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

SIENTIAPDE-1193

Browse Source 
Enhance timestamp handling in Redis and CoreScouter workflows by updating to DATETIME_FORMAT_MS_WITH_TZ. Modify tests to reflect new timestamp format, ensuring consistency across data structures and improving overall datetime management.
This commit is contained in:
vitor-aignosi
2025-08-22 14:05:41 -03:00
parent 0b337d00f9
commit 96a28e91e5
4 changed files with 17 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
scouter/activities/redis.py
View File

@@ -149,6 +149,7 @@ class Redis(RedisBase):
# Remove possibly removed tags # Remove possibly removed tags
tags = list(model_tags.keys()) tags = list(model_tags.keys())
tags.append('timestamp')
self.debug( self.debug(
f"Tags to keep: {tags}", f"Tags to keep: {tags}",
metadata=metadata metadata=metadata

4
scouter/workflow/sub_workflows/core_scouter.py
View File

@@ -5,7 +5,7 @@ with workflow.unsafe.imports_passed_through():
from typing import Any from typing import Any
from datetime import timedelta from datetime import timedelta
from sientia_do.temporal.policies import retry_policy 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="core_scouter") @workflow.defn(name="core_scouter")
@@ -85,7 +85,7 @@ class CoreScouter:
'data': held_data, 'data': held_data,
'timestamp_conversion': { 'timestamp_conversion': {
'column': 'timestamp', 'column': 'timestamp',
'format': DATETIME_FORMAT 'format': DATETIME_FORMAT_MS_WITH_TZ
} }
}, },
retry_policy=retry_policy, retry_policy=retry_policy,

14
tests/activities/test_mongo.py
View File

@@ -96,7 +96,7 @@ async def test_load_latest_data_none_last_data_timestamp(mongodb_activity):
'name': 'test1', 'name': 'test1',
'value': 1, 'value': 1,
'inserted_at': datetime.strptime( 'inserted_at': datetime.strptime(
'2023-01-01 12:00:00.000000', DATETIME_FORMAT_MS_WITH_TZ) '2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ)
} }
] ]
@@ -122,7 +122,7 @@ async def test_load_latest_data_none_last_data_timestamp(mongodb_activity):
0: 1 0: 1
}, },
'inserted_at': { 'inserted_at': {
0: '2023-01-01 12:00:00.000000' 0: '2023-01-01 12:00:00.000000+0000'
} }
} }
@@ -138,14 +138,14 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongodb_activity):
'name': 'test1', 'name': 'test1',
'value': 1, 'value': 1,
'inserted_at': datetime.strptime( 'inserted_at': datetime.strptime(
'2023-01-01 12:00:00.000000', DATETIME_FORMAT_MS_WITH_TZ) '2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ)
} }
] ]
result = await mongodb_activity.load_latest_data({ result = await mongodb_activity.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection', 'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000' 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000'
}) })
mongodb_activity.database.__getitem__.assert_called_once_with( mongodb_activity.database.__getitem__.assert_called_once_with(
@@ -155,7 +155,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongodb_activity):
{ {
'inserted_at': { 'inserted_at': {
'$gt': datetime.strptime( '$gt': datetime.strptime(
'2023-01-01 12:00:00.000000', DATETIME_FORMAT_MS_WITH_TZ) '2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ)
} }
}, },
{"_id": 0} {"_id": 0}
@@ -169,7 +169,7 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongodb_activity):
0: 1 0: 1
}, },
'inserted_at': { 'inserted_at': {
0: '2023-01-01 12:00:00.000000' 0: '2023-01-01 12:00:00.000000+0000'
} }
} }
@@ -187,7 +187,7 @@ async def test_load_latest_data_error(mongodb_activity):
await mongodb_activity.load_latest_data({ await mongodb_activity.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'}, 'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection', 'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000' 'last_data_timestamp': '2023-01-01 12:00:00.000000+0000'
}) })
except Exception as e: except Exception as e:
assert str(e) == 'test' assert str(e) == 'test'

8
tests/workflow/sub_workflows/test_core_scouter.py
View File

@@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, patch, call, ANY
import pytest import pytest
from scouter.workflow.sub_workflows.core_scouter import CoreScouter from scouter.workflow.sub_workflows.core_scouter import CoreScouter
from scouter.activities.activities import Activities from scouter.activities.activities import Activities
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
@pytest.fixture @pytest.fixture
@@ -94,7 +95,12 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter):
**expected_metadata, **expected_metadata,
'schema': 'test_schema', 'schema': 'test_schema',
'table_name': 'test_table', 'table_name': 'test_table',
'data': 'held_data'}, 'data': 'held_data',
'timestamp_conversion': {
'column': 'timestamp',
'format': DATETIME_FORMAT_MS_WITH_TZ
}
},
retry_policy=ANY, retry_policy=ANY,
start_to_close_timeout=ANY start_to_close_timeout=ANY
) )