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

@@ -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]
 

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",

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']

orchestrator/utils/patterns.py

@@ -1 +0,0 @@
-DEFAULT_DATE_FORMAT = '%Y-%m-%d %H:%M:%S.%f'

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),

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