gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1184

Browse Source 
refactor: consolidate report handling in Formatters class

- Introduced a new method, manage_and_send_report, to streamline success and error report generation.
- Improved code readability by reducing duplication in report handling logic.
- Maintained existing functionality while enhancing the structure of report messages.
This commit is contained in:
vitor-aignosi
2025-08-14 09:26:10 -03:00
parent 4791665642
commit 36c34d3208
1 changed files with 32 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
orchestrator/activities/formatters.py
View File

@@ -360,6 +360,31 @@ class Formatters(BaseActivity):
return success_keys, error_keys 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]):
if len(success_keys) > 0:
self.send_success_report(
metadata=metadata,
message=f"Successfully {schedule_type}: \n {', '.join(success_keys)}",
notification_id=schedule_data['id'],
attachment=schedule_data['items']
)
if len(error_keys) > 0:
attachment = []
for key, value in error_keys.items():
if value['attachment'] is not None:
attachment.append(
f"{key}:\n{value['message']}\n{value['attachment']}")
else:
attachment.append(f"{key}:\n{value['message']}")
self.send_error_report(
metadata=metadata,
message=f"Fails on {schedule_type}: \n {', '.join(error_keys)}",
notification_id=f"{schedule_data['id']}_ERROR",
attachment=topic_separator.join(attachment)
)
@activity.defn(name="report_schedule_orchestration") @activity.defn(name="report_schedule_orchestration")
async def report_schedule_orchestration(self, async def report_schedule_orchestration(self,
input_data: dict[str, Any]) -> None: input_data: dict[str, Any]) -> None:
@@ -401,29 +426,13 @@ class Formatters(BaseActivity):
success_keys, error_keys = self.parse_report_schedule( success_keys, error_keys = self.parse_report_schedule(
schedule_data['items']) schedule_data['items'])
if len(success_keys) > 0: self.manage_and_send_report(
self.send_success_report( metadata=metadata,
metadata=metadata, success_keys=success_keys,
message=f"Successfully {schedule_type}: \n {', '.join(success_keys)}", error_keys=error_keys,
notification_id=schedule_data['id'], schedule_type=schedule_type,
attachment=schedule_data['items'] schedule_data=schedule_data
) )
if len(error_keys) > 0:
attachment = []
for key, value in error_keys.items():
if value['attachment'] is not None:
attachment.append(
f"{key}:\n{value['message']}\n{value['attachment']}")
else:
attachment.append(f"{key}:\n{value['message']}")
self.send_error_report(
metadata=metadata,
message=f"Fails on {schedule_type}: \n {', '.join(error_keys)}",
notification_id=f"{schedule_data['id']}_ERROR",
attachment=topic_separator.join(attachment)
)
@activity.defn(name="report_slot_orchestration") @activity.defn(name="report_slot_orchestration")
async def report_slot_orchestration(self, async def report_slot_orchestration(self,