From bfb633c9d7e3b98f94ea35b44e79554ce87c4da1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 14 Aug 2025 09:29:48 -0300 Subject: [PATCH] SIENTIAPDE-1184 docs: enhance docstrings in Formatters class - Improved clarity and consistency of docstrings across multiple methods in the Formatters class. - Added detailed descriptions for method arguments and return values to facilitate better understanding and usage. - Ensured that all public methods now have comprehensive documentation, enhancing maintainability and readability. --- orchestrator/activities/formatters.py | 95 +++++++++++++++++++++++++-- 1 file changed, 88 insertions(+), 7 deletions(-) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index e0452b0..0bfbc15 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -168,11 +168,13 @@ class Formatters(BaseActivity): async def format_schedule_config(self, input_data: dict[str, Any]) -> dict[str, Any]: """ Formats the schedule config to a dictionary with the schedule name as the key. - input_data: - - schedule_config (list[dict[str, Any]]): The schedule config to format. + + Args: + input_data (dict[str, Any]): The input data containing the schedule config to format. + - schedule_config (list[dict[str, Any]]): The schedule config to format. Returns: - - dict[str, Any]: The formatted schedule config. + dict[str, Any]: The formatted schedule config. """ schedule_config = input_data['schedule_config'] @@ -194,7 +196,16 @@ class Formatters(BaseActivity): to_update: dict[str, Any], to_create: dict[str, Any], namespace: str, metadata: dict[str, Any]): """ - Compares the timestamps of the schedule and the current schedule. + Compares the timestamps of the schedule and the current schedule to determine + which schedules need to be updated or created. + + Args: + schedules (dict[str, Any]): The new schedules to compare. + current_schedules (dict[str, Any]): The existing schedules to compare against. + to_update (dict[str, Any]): Dictionary to populate with schedules that need updating. + to_create (dict[str, Any]): Dictionary to populate with schedules that need creating. + namespace (str): The namespace for the schedules. + metadata (dict[str, Any]): Metadata for logging purposes. """ for schedule_name, schedule in schedules.items(): if schedule_name in current_schedules: @@ -319,6 +330,15 @@ class Formatters(BaseActivity): return output def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str = None) -> None: + """ + Sends a success notification report. + + Args: + metadata (dict[str, Any]): Metadata for the notification. + message (str): The success message to send. + notification_id (str): The ID of the notification. + attachment (str, optional): Optional attachment content for the notification. + """ self.send_notification( metadata=metadata, notification_id=notification_id, @@ -330,6 +350,15 @@ class Formatters(BaseActivity): def send_error_report(self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str) -> None: + """ + Sends an error notification report. + + Args: + metadata (dict[str, Any]): Metadata for the notification. + message (str): The error message to send. + notification_id (str): The ID of the notification. + attachment (str): The attachment content for the notification. + """ self.send_notification( metadata=metadata, notification_id=notification_id, @@ -340,6 +369,19 @@ class Formatters(BaseActivity): ) def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], dict[str, Any]]: + """ + Parses the report schedule data to extract success and error information. + + Args: + input_data (dict[str, Any]): The input data containing schedule reports. + Each item should have 'namespace', 'schedule_name', 'success', 'message', + and optionally 'attachment' fields. + + Returns: + tuple[list[str], dict[str, Any]]: A tuple containing: + - List of successful schedule keys in format "namespace/schedule_name" + - Dictionary of error keys mapped to their error details + """ success_keys = [f"{value['namespace']}/{value['schedule_name']}" for value in input_data if value['success']] @@ -352,6 +394,18 @@ class Formatters(BaseActivity): return success_keys, error_keys def parse_report(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]: + """ + Parses the report data to extract success and error keys. + + Args: + input_data (dict[str, Any]): The input data containing report items. + Each item should have a 'success' field indicating success/failure. + + Returns: + tuple[list[str], list[str]]: A tuple containing: + - List of successful keys + - List of error keys + """ success_keys = [key for key, value in input_data.items() if value['success']] @@ -361,6 +415,16 @@ class Formatters(BaseActivity): return success_keys, error_keys def manage_and_send_report(self, metadata: dict[str, Any], success_keys: list[str], error_keys: dict[str, Any], schedule_type: str, schedule_data: dict[str, Any]): + """ + Manages and sends success and error reports based on the provided keys and data. + + Args: + metadata (dict[str, Any]): Metadata for logging and notifications. + success_keys (list[str]): List of keys that were successful. + error_keys (dict[str, Any]): Dictionary of error keys mapped to error details. + schedule_type (str): The type of schedule being reported (e.g., 'created schedules'). + schedule_data (dict[str, Any]): The schedule data containing items and notification ID. + """ if len(success_keys) > 0: self.send_success_report( metadata=metadata, @@ -494,8 +558,15 @@ class Formatters(BaseActivity): async def format_log_report(self, input_data: dict[str, Any]) -> dict[str, Any]: """ Formats the receiver_groups status to a dataframe to be stored in the database. - input_data: - - receiver_groups (dict): The receiver groups. + + Args: + input_data (dict[str, Any]): The input data containing: + - receiver_groups (dict): The receiver groups configuration. + - mail_type (str): The type of mail for the report. + - metadata (dict): Metadata for logging purposes. + + Returns: + dict[str, Any]: The formatted log report as a dictionary representation of a DataFrame. """ metadata = input_data["metadata"] mail_type = input_data["mail_type"] @@ -540,7 +611,17 @@ class Formatters(BaseActivity): @activity.defn(name="filter_notification_reports") async def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]: """ - Filter notification reports. + Filters notification reports based on sending configurations and notification package. + + Args: + input_data (dict[str, Any]): The input data containing: + - metadata (dict): Metadata for logging purposes. + - notification_package (list): The package of notifications to filter. + - sending_configs (list): The configurations for sending notifications. + Each config should have 'group_name', 'contents', and optionally 'ignore' fields. + + Returns: + dict[str, Any]: The filtered receiver groups with their notifications. """ metadata = input_data['metadata'] notification_package = input_data['notification_package']