diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 0bf2649..51606fa 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -453,7 +453,7 @@ class MongoDB(SientiaMonitoring): - metadata (dict[str, Any]): Workflow execution metadata - collection_name (str): Name of the MongoDB collection - last_data_timestamp (str | None): Last processed timestamp - (ISO-format string) for filtering + (formatted per ``DATETIME_FORMAT_MS_WITH_TZ``) for filtering - base_data_filter (dict[str, Any]): Base query filter conditions Returns: @@ -475,12 +475,15 @@ class MongoDB(SientiaMonitoring): else: # ``notification_queue.timestamp`` is stored as a native BSON # ``Date`` (``Notification.timestamp`` is a ``datetime``). - # ``last_data_timestamp`` arrives here as an ISO-format string - # (round-tripped through Redis), so it must be parsed back to - # ``datetime`` for the ``$gt`` comparison to be type-correct. + # ``last_data_timestamp`` arrives here as a string formatted per + # ``DATETIME_FORMAT_MS_WITH_TZ`` (round-tripped through Redis), + # so it must be parsed back to ``datetime`` for the ``$gt`` + # comparison to be type-correct. data_filter = { **base_data_filter, - 'timestamp': {'$gt': datetime.fromisoformat(last_data_timestamp)}, + 'timestamp': { + '$gt': datetime.strptime(last_data_timestamp, DATETIME_FORMAT_MS_WITH_TZ) + }, } self.debug(f'Data filter: {data_filter}', metadata=metadata) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index 6a6192b..e0f778a 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -337,8 +337,9 @@ class SlotManager(SientiaMonitoring): # ``timestamp`` is a native ``datetime``/``Timestamp`` (BSON Date from # Mongo), which the Redis repository's plain ``json.dumps`` cannot - # serialize. Store it as an ISO-format string instead. - last_data_timestamp = last_data_timestamp.isoformat() + # serialize. Store it as a string formatted per + # ``DATETIME_FORMAT_MS_WITH_TZ`` instead. + last_data_timestamp = last_data_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ) try: self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5) diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 4e55f12..927f860 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -504,7 +504,7 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db): { 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': '2023-01-01T12:00:00+00:00', + 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000', 'base_data_filter': {'level': 'ERROR'}, } ) @@ -561,7 +561,7 @@ def test_load_latest_data_run_boundary_no_skip_or_duplicate(mongo_db): { 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': boundary_timestamp.isoformat(), + 'last_data_timestamp': boundary_timestamp.strftime(DATETIME_FORMAT_MS_WITH_TZ), 'base_data_filter': {'level': 'ERROR'}, } ) @@ -578,7 +578,7 @@ def test_load_latest_data_error(mongo_db): { 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'collection_name': 'test_collection', - 'last_data_timestamp': '2023-01-01T12:00:00+00:00', + 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000', 'base_data_filter': {'level': 'ERROR'}, } ) diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index b16c41f..12873ca 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -297,10 +297,10 @@ def test_put_last_data_timestamp_not_empty_dataframe(slot_manager): result = slot_manager.put_last_data_timestamp(test_data) - assert result == '2023-01-01T12:00:01+00:00' + assert result == '2023-01-01 12:00:01.000000+0000' slot_manager.redis_repository.set.assert_called_once_with( - 'notification_last_timestamp:test_mail_type', '2023-01-01T12:00:01+00:00', ttl=18000 + 'notification_last_timestamp:test_mail_type', '2023-01-01 12:00:01.000000+0000', ttl=18000 )