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

SIENTIAPDE-1184

Browse Source 
SIENTIAPDE-1184
 refactor: improve schedule reporting in Formatters class

- Introduced a unified reporting structure for created, updated, and deleted schedules.
- Enhanced success and error report messages for clarity, including specific schedule types in notifications.
- Updated attachment handling in success and error reports to improve context and readability.
This commit is contained in:
vitor-aignosi
2025-08-14 09:19:48 -03:00
parent 26a31a39d6
commit 4791665642
2 changed files with 73 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
orchestrator/activities/formatters.py
View File

@@ -19,6 +19,8 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT
from math import ceil from math import ceil
topic_separator = "\n ========== \n"
class Formatters(BaseActivity): class Formatters(BaseActivity):
def __init__(self, def __init__(self,
@@ -316,24 +318,25 @@ class Formatters(BaseActivity):
return output return output
def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str) -> None: def send_success_report(self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str = None) -> None:
self.send_notification( self.send_notification(
metadata=metadata, metadata=metadata,
notification_id=notification_id, notification_id=notification_id,
message=message, message=message,
block="report_orchestration", block="report_orchestration",
level=NotificationLevel.INFO level=NotificationLevel.INFO,
attachment_content=json.dumps(attachment, indent=4, sort_keys=True)
) )
def send_error_report(self, metadata: dict[str, Any], message: str, notification_id: str, def send_error_report(self, metadata: dict[str, Any], message: str, notification_id: str,
attachment: dict[str, Any]) -> None: attachment: str) -> None:
self.send_notification( self.send_notification(
metadata=metadata, metadata=metadata,
notification_id=notification_id, notification_id=notification_id,
message=message, message=message,
block="report_orchestration", block="report_orchestration",
level=NotificationLevel.ERROR, level=NotificationLevel.ERROR,
attachment_content=json.dumps(attachment, indent=4, sort_keys=True) attachment_content=attachment
) )
def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], dict[str, Any]]: def parse_report_schedule(self, input_data: dict[str, Any]) -> tuple[list[str], dict[str, Any]]:
@@ -378,91 +381,49 @@ class Formatters(BaseActivity):
updated_schedules = input_data['updated_schedules'] updated_schedules = input_data['updated_schedules']
deleted_schedules = input_data['deleted_schedules'] deleted_schedules = input_data['deleted_schedules']
# Send report for created schedules schedules_report = {
if len(created_schedules) > 0: 'created schedules': {
success_keys, error_keys = self.parse_report_schedule( 'items': created_schedules,
created_schedules) 'id': 'REPORT_ORCHESTRATION_CREATED_SCHEDULES'
},
'updated schedules': {
'items': updated_schedules,
'id': 'REPORT_ORCHESTRATION_UPDATED_SCHEDULES'
},
'deleted schedules': {
'items': deleted_schedules,
'id': 'REPORT_ORCHESTRATION_DELETED_SCHEDULES'
}
}
if len(success_keys) > 0: for schedule_type, schedule_data in schedules_report.items():
self.send_success_report( if len(schedule_data['items']) > 0:
metadata=metadata, success_keys, error_keys = self.parse_report_schedule(
message=f"Created schedules: \n {', '.join(success_keys)}", schedule_data['items'])
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES",
attachment=created_schedules
)
if len(error_keys) > 0: if len(success_keys) > 0:
attachment = [] self.send_success_report(
for value in error_keys.values(): metadata=metadata,
if value['attachment'] is not None: message=f"Successfully {schedule_type}: \n {', '.join(success_keys)}",
attachment.append( notification_id=schedule_data['id'],
f"{value['message']}\n{value['attachment']}") attachment=schedule_data['items']
else: )
attachment.append(value['message'])
self.send_error_report( if len(error_keys) > 0:
metadata=metadata, attachment = []
message=f"Failed to create schedules: \n {', '.join(error_keys)}", for key, value in error_keys.items():
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR", if value['attachment'] is not None:
attachment="\n ========== \n".join(attachment) attachment.append(
) f"{key}:\n{value['message']}\n{value['attachment']}")
else:
attachment.append(f"{key}:\n{value['message']}")
# Send report for updated schedules self.send_error_report(
if len(updated_schedules) > 0: metadata=metadata,
success_keys, error_keys = self.parse_report_schedule( message=f"Fails on {schedule_type}: \n {', '.join(error_keys)}",
updated_schedules) notification_id=f"{schedule_data['id']}_ERROR",
attachment=topic_separator.join(attachment)
if len(success_keys) > 0: )
self.send_success_report(
metadata=metadata,
message=f"Updated schedules: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
attachment=updated_schedules
)
if len(error_keys) > 0:
attachment = []
for value in error_keys.values():
if value['attachment'] is not None:
attachment.append(
f"{value['message']}\n{value['attachment']}")
else:
attachment.append(value['message'])
self.send_error_report(
metadata=metadata,
message=f"Failed to update schedules: \n {', '.join(error_keys)}",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR",
attachment="\n ========== \n".join(attachment)
)
if len(deleted_schedules) > 0:
success_keys, error_keys = self.parse_report_schedule(
deleted_schedules)
if len(success_keys) > 0:
self.send_success_report(
metadata=metadata,
message=f"Deleted schedules: \n {', '.join(success_keys)}",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES",
attachment=deleted_schedules
)
if len(error_keys) > 0:
attachment = []
for value in error_keys.values():
if value['attachment'] is not None:
attachment.append(
f"{value['message']}\n{value['attachment']}")
else:
attachment.append(value['message'])
self.send_error_report(
metadata=metadata,
message=f"Failed to delete schedules: \n {', '.join(error_keys)}",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR",
attachment="\n ========== \n".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,

52
tests/orchestrator/activities/test_formatters.py
View File

@@ -466,21 +466,6 @@ async def test_create_slot_config(formatters):
def test_send_success_report(formatters): def test_send_success_report(formatters):
formatters.send_success_report( formatters.send_success_report(
metadata=metadata,
message="test_message",
notification_id="test_notification_id"
)
formatters.send_notification.assert_called_once_with(
metadata=metadata,
notification_id="test_notification_id",
message="test_message",
block="report_orchestration",
level=NotificationLevel.INFO
)
def test_send_error_report(formatters):
formatters.send_error_report(
metadata=metadata, metadata=metadata,
message="test_message", message="test_message",
notification_id="test_notification_id", notification_id="test_notification_id",
@@ -491,12 +476,29 @@ def test_send_error_report(formatters):
notification_id="test_notification_id", notification_id="test_notification_id",
message="test_message", message="test_message",
block="report_orchestration", block="report_orchestration",
level=NotificationLevel.ERROR, level=NotificationLevel.INFO,
attachment_content=json.dumps( attachment_content=json.dumps(
{"test": "test"}, indent=4, sort_keys=True) {"test": "test"}, indent=4, sort_keys=True)
) )
def test_send_error_report(formatters):
formatters.send_error_report(
metadata=metadata,
message="test_message",
notification_id="test_notification_id",
attachment="test_attachment"
)
formatters.send_notification.assert_called_once_with(
metadata=metadata,
notification_id="test_notification_id",
message="test_message",
block="report_orchestration",
level=NotificationLevel.ERROR,
attachment_content="test_attachment"
)
def test_parse_report(formatters): def test_parse_report(formatters):
input_data = { input_data = {
"test_key": { "test_key": {
@@ -631,19 +633,19 @@ async def test_report_schedule_orchestration(formatters):
formatters.send_success_report.assert_has_calls([ formatters.send_success_report.assert_has_calls([
call( call(
metadata=metadata['metadata'], metadata=metadata['metadata'],
message="Created schedules: \n test_namespace/test_schedule_name_to_create", message="Successfully created schedules: \n test_namespace/test_schedule_name_to_create",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES", notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES",
attachment=input_data['created_schedules'] attachment=input_data['created_schedules']
), ),
call( call(
metadata=metadata['metadata'], metadata=metadata['metadata'],
message="Updated schedules: \n test_namespace/test_schedule_name_to_update", message="Successfully updated schedules: \n test_namespace/test_schedule_name_to_update",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES", notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES",
attachment=input_data['updated_schedules'] attachment=input_data['updated_schedules']
), ),
call( call(
metadata=metadata['metadata'], metadata=metadata['metadata'],
message="Deleted schedules: \n test_namespace/test_schedule_name_to_delete", message="Successfully deleted schedules: \n test_namespace/test_schedule_name_to_delete",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES", notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES",
attachment=input_data['deleted_schedules'] attachment=input_data['deleted_schedules']
) )
@@ -651,21 +653,21 @@ async def test_report_schedule_orchestration(formatters):
formatters.send_error_report.assert_has_calls([ formatters.send_error_report.assert_has_calls([
call( call(
metadata=metadata['metadata'], metadata=metadata['metadata'],
message="Failed to create schedules: \n test_namespace/test_schedule_name_to_create_error, test_namespace/test_schedule_name_to_create_error2", message="Fails on created schedules: \n test_namespace/test_schedule_name_to_create_error, test_namespace/test_schedule_name_to_create_error2",
notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR", notification_id="REPORT_ORCHESTRATION_CREATED_SCHEDULES_ERROR",
attachment="test_error1\ntest_attachment1\n ========== \ntest_error2" attachment="test_namespace/test_schedule_name_to_create_error:\ntest_error1\ntest_attachment1\n ========== \ntest_namespace/test_schedule_name_to_create_error2:\ntest_error2"
), ),
call( call(
metadata=metadata['metadata'], metadata=metadata['metadata'],
message="Failed to update schedules: \n test_namespace/test_schedule_name_to_update_error, test_namespace/test_schedule_name_to_update_error2, test_namespace/test_schedule_name_to_update_error3", message="Fails on updated schedules: \n test_namespace/test_schedule_name_to_update_error, test_namespace/test_schedule_name_to_update_error2, test_namespace/test_schedule_name_to_update_error3",
notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR", notification_id="REPORT_ORCHESTRATION_UPDATED_SCHEDULES_ERROR",
attachment="test_error2\ntest_attachment2\n ========== \ntest_error3\ntest_attachment3\n ========== \ntest_error4" attachment="test_namespace/test_schedule_name_to_update_error:\ntest_error2\ntest_attachment2\n ========== \ntest_namespace/test_schedule_name_to_update_error2:\ntest_error3\ntest_attachment3\n ========== \ntest_namespace/test_schedule_name_to_update_error3:\ntest_error4"
), ),
call( call(
metadata=metadata['metadata'], metadata=metadata['metadata'],
message="Failed to delete schedules: \n test_namespace/test_schedule_name_to_delete_error, test_namespace/test_schedule_name_to_delete_error2", message="Fails on deleted schedules: \n test_namespace/test_schedule_name_to_delete_error, test_namespace/test_schedule_name_to_delete_error2",
notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR", notification_id="REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR",
attachment="test_error4\ntest_attachment4\n ========== \ntest_error5" attachment="test_namespace/test_schedule_name_to_delete_error:\ntest_error4\ntest_attachment4\n ========== \ntest_namespace/test_schedule_name_to_delete_error2:\ntest_error5"
) )
]) ])