"""Reports Logic.""" from typing import Any, Dict, Optional from collaborator import config from collaborator.models.rds.report_run_persister import ReportRunPersister from collaborator.utils.lambda_fn import invoke_lambda from collaborator.utils.typing import User def report_runs_dataloader( report_run_ids: list[int], ): """Get report runs dataloader.""" results = ReportRunPersister.get_report_runs_by_ids(report_run_ids) report_runs_by_id = {result.report_run_id: result for result in results} return [report_runs_by_id.get(report_run_id) for report_run_id in report_run_ids] def trigger_auto_report_run( statement_period_id: int, user: User, ) -> Dict[str, Any]: """Trigger automatic report run creation. Args: statement_period_id: ID of the statement period user: User who made the request Returns: Dictionary containing the Lambda response """ return invoke_lambda( function_name=config.LAMBDA_FUNCTION_AUTO_REPORT_RUN, payload={ "target_type": "statement_period_id", "target_id": statement_period_id, "requestor_identity_uuid": user.id, }, ) def get_report_run_participations( account_id: Optional[int] = None, report_run_id: Optional[str] = None, sort_key: Optional[str] = None, sort_direction: Optional[str] = None, limit: int = 0, offset: int = 0, ) -> tuple: """Get a list of the reports runs for an account. Args: account (Account): Account to get reports for sort_key (str?): Key to sort by (defaults to no sorting) sort_direction (str?): Direction to sort by (ASC or DESC) limit (int): Amount of reports runs to get offset (int): Used to paginate amount of report runs Returns: Response """ return ReportRunPersister.get_report_run_participations( account_id=account_id, report_run_id=report_run_id, sort_key=sort_key, sort_direction=sort_direction, limit=limit, offset=offset, )