From 9672a9cb7e2e4a0806418af8deb48ea3b538605b Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 22 Aug 2025 11:45:11 -0300 Subject: [PATCH] SIENTIAPDE-1193 refactor: update date handling and remove deprecated constants - Replaced the usage of datetime.now() with a new now() function from sientia_do.temporal.constants for consistent timestamp handling across multiple files. - Updated the date format from DEFAULT_DATE_FORMAT to DATETIME_FORMAT_MS_WITH_TZ in various activities to ensure uniformity in timestamp formatting. - Removed the obsolete patterns.py file that contained the DEFAULT_DATE_FORMAT constant. --- orchestrator/activities/formatters.py | 11 +++++------ orchestrator/activities/mongo_db.py | 12 ++++++------ orchestrator/activities/slot_manager.py | 8 ++++---- orchestrator/utils/patterns.py | 1 - .../workflows/subworkflows/process_notifications.py | 4 ++-- requirements.txt | 2 +- 6 files changed, 18 insertions(+), 20 deletions(-) delete mode 100644 orchestrator/utils/patterns.py diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 34732be..f3212ab 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -9,14 +9,13 @@ with workflow.unsafe.imports_passed_through(): import json from typing import Any from logging import Logger - from datetime import datetime from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.temporal.activities.base import BaseActivity from sientia_do.notifications.models import NotificationLevel from orchestrator.utils.orchestrator_functions import ( scouter, predictions_batch, gather_read_tags, build_tag_config ) - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now from math import ceil topic_separator = "\n ========== \n" @@ -63,21 +62,21 @@ class Formatters(BaseActivity): schedule_config[self.scouter_namespace][pipeline['schedule_name']] = { **scouter(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } elif pipeline['workflow_type'] == 'predictions_batch': schedule_config[self.laborious_namespace][pipeline['schedule_name'] ] = { **predictions_batch(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } elif pipeline['workflow_type'] == 'minimal_retrain': schedule_config[self.laborious_namespace][pipeline['schedule_name'] ] = { **minimal_retrain(pipeline), "updated_at": pipeline.get( - "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT)) + "updated_at", now().strftime(DATETIME_FORMAT_MS_WITH_TZ)) } self.info("Processed schedules", metadata=metadata) @@ -219,7 +218,7 @@ class Formatters(BaseActivity): if schedule_name in current_schedules: update_timestamp = schedule.get( - 'updated_at', datetime.now()) + 'updated_at', now()) old_timestamp = current_schedules[schedule_name] diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index e7a3ac1..3e094b4 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -2,7 +2,6 @@ from temporalio import workflow, activity with workflow.unsafe.imports_passed_through(): - from datetime import datetime from typing import Any import traceback from logging import Logger @@ -10,7 +9,8 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.activities.base import BaseActivity - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now + from datetime import datetime def clear_mongo_id(docs: list) -> list: @@ -216,7 +216,7 @@ class MongoDB(BaseActivity): """ updated_pipelines = input_data.get("updated_pipelines", []) metadata = input_data.get("metadata", {}) - now = datetime.now() + now = now() collection = self.database["orchestrated_schedules"] self.info("Updating pipelines timestamps...", metadata=metadata) @@ -267,7 +267,7 @@ class MongoDB(BaseActivity): success_count = 0 - now = datetime.now() + now = now() argument = [ {"schedule_name": pipeline["schedule_name"], @@ -442,7 +442,7 @@ class MongoDB(BaseActivity): data_filter = { **base_data_filter, "timestamp": { - "$gt": datetime.strptime(last_data_timestamp, DEFAULT_DATE_FORMAT) + "$gt": datetime.strptime(last_data_timestamp, DATETIME_FORMAT_MS_WITH_TZ) } } @@ -460,7 +460,7 @@ class MongoDB(BaseActivity): for item in data: item['timestamp'] = item['timestamp'].strftime( - DEFAULT_DATE_FORMAT) + DATETIME_FORMAT_MS_WITH_TZ) self.info( f"Loaded {len(data)} documents from MongoDB", diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index a7dab02..bc68d23 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -9,7 +9,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.redis_base import Redis from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel - from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now from pandas import DataFrame from datetime import datetime, timedelta @@ -344,10 +344,10 @@ class SlotManager(Redis): else: last_sent = datetime.strptime( - last_sent, DEFAULT_DATE_FORMAT) + last_sent, DATETIME_FORMAT_MS_WITH_TZ) # Check if "notification_ttl" seconds has passed since last sent - if (datetime.now() - last_sent) > timedelta(seconds=notification_ttl): + if (now() - last_sent) > timedelta(seconds=notification_ttl): alert_type = "persistent_alerts" # Check if this group must be notified @@ -382,7 +382,7 @@ class SlotManager(Redis): self.info("Storing notification cache...", metadata=metadata) - now = datetime.now().strftime(DEFAULT_DATE_FORMAT) + now = now().strftime(DATETIME_FORMAT_MS_WITH_TZ) for index, row in log_report.iterrows(): status = row['status'] diff --git a/orchestrator/utils/patterns.py b/orchestrator/utils/patterns.py deleted file mode 100644 index fe00beb..0000000 --- a/orchestrator/utils/patterns.py +++ /dev/null @@ -1 +0,0 @@ -DEFAULT_DATE_FORMAT = '%Y-%m-%d %H:%M:%S.%f' diff --git a/orchestrator/workflows/subworkflows/process_notifications.py b/orchestrator/workflows/subworkflows/process_notifications.py index 1c2a0bd..e02fc12 100644 --- a/orchestrator/workflows/subworkflows/process_notifications.py +++ b/orchestrator/workflows/subworkflows/process_notifications.py @@ -5,7 +5,7 @@ with workflow.unsafe.imports_passed_through(): from typing import Any from datetime import timedelta from sientia_do.temporal.policies import retry_policy - from sientia_do.temporal.constants import DATETIME_FORMAT + from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ @workflow.defn(name="process_notifications") @@ -84,7 +84,7 @@ class ProcessNotifications: "data": log_report, 'timestamp_conversion': { 'column': 'timestamp', - 'format': DATETIME_FORMAT + 'format': DATETIME_FORMAT_MS_WITH_TZ } }, schedule_to_close_timeout=timedelta(seconds=60), diff --git a/requirements.txt b/requirements.txt index 5d8b5e9..f27fcce 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,5 +5,5 @@ redis couchbase pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.2 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.3 prometheus-client