Merge pull request #17 from Aignosi/SIENTIAPDE-1193-conferir-como-a-escrita-de-datetime-ocorre-no-temporal

SIENTIAPDE-1193: Improve Datetime Handling and Update Dependencies

coverage.sh

@@ -0,0 +1 @@
+pytest --cov=laborious --cov-report=html && xdg-open htmlcov/index.html

laborious/activities/gates.py

@@ -7,6 +7,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.notifications.models import NotificationLevel
     from sientia_do.temporal.activities.base import BaseActivity
     from sientia_do.observability.logger import Logger
+    from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, now
     from laborious.utils.filters.mlflow_filters import nan_values_filter, api_error_filter
     from typing import Any
     from laborious.utils.filters.conditional_filters import (
@@ -14,7 +15,6 @@ with workflow.unsafe.imports_passed_through():
         filter_specific_variables_null_values
     )
     from pandas import DataFrame
-    from datetime import datetime
     from laborious import metrics
 
 input_filter_functions = {
@@ -318,13 +318,18 @@ class Gates(BaseActivity):
 
         data = DataFrame(input_data['data'])
 
+        self.debug(f"Input data: {data.to_string()}", metadata)
+
         if data.empty:
-            return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+            return now().strftime(DATETIME_FORMAT_WITH_TZ)
+
+        max_timestamp = max(
+            data['timestamp'].values.tolist())
 
         self.info(
-            f"Last timestamp: {max(data['timestamp'].values.tolist())}", metadata)
+            f"Last timestamp: {max_timestamp}", metadata)
 
-        return max(data['timestamp'].values.tolist())
+        return max_timestamp
 
     @activity.defn(name="write_metrics")
     async def write_metrics(self, input_data: dict[str, Any]):

laborious/workflows/minimal_retrain.py

@@ -52,6 +52,7 @@ class MinimalRetrain():
             {
                 **metadata,
                 'query': input_data['query'],
+                'datetime_columns': input_data.get('datetime_columns', [])
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60)

laborious/workflows/predictions_batch.py

@@ -54,6 +54,7 @@ class PredictionsBatch():
             {
                 **metadata,
                 'query': input_data['query'],
+                'datetime_columns': input_data.get('datetime_columns', [])
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60)

laborious/workflows/sub_workflows/format_and_export_prediction.py

@@ -5,6 +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_WITH_TZ
 
 
 @workflow.defn(name="format_and_export_prediction")
@@ -91,6 +92,10 @@ class FormatAndExportPrediction():
                 'schema': input_data['schema'],
                 'table_name': input_data['table_name'],
                 'data': prediction,
+                'timestamp_conversion': {
+                    'column': 'timestamp',
+                    'format': DATETIME_FORMAT_WITH_TZ
+                }
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=60)

requirements.txt

@@ -3,6 +3,6 @@ psycopg2-binary
 sqlalchemy
 asyncua
 redis
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.1
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.4
 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.38.5
 prometheus-client

tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py

@@ -3,6 +3,7 @@ from pytest import mark, fixture
 
 from laborious.activities.activities import Activities
 from laborious.workflows.sub_workflows.format_and_export_prediction import FormatAndExportPrediction
+from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
 
 
 @fixture
@@ -73,7 +74,11 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction):
                 'schema': input_data['schema'],
                 'table_name': input_data['table_name'],
                 'data': workflow_mock.execute_activity_method.return_value,
-                **metadata
+                **metadata,
+                'timestamp_conversion': {
+                    'column': 'timestamp',
+                    'format': DATETIME_FORMAT_WITH_TZ
+                }
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY
@@ -138,7 +143,11 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction
                 'schema': input_data['schema'],
                 'table_name': input_data['table_name'],
                 'data': workflow_mock.execute_activity_method.return_value,
-                **metadata
+                **metadata,
+                'timestamp_conversion': {
+                    'column': 'timestamp',
+                    'format': DATETIME_FORMAT_WITH_TZ
+                }
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY

tests/laborious/workflows/test_minimal_retrain.py

@@ -48,6 +48,7 @@ async def test_run(workflow_mock: AsyncMock, minimal_retrain: MinimalRetrain):
                 {
                     **metadata,
                     "query": input_data["query"],
+                    'datetime_columns': input_data.get('datetime_columns', [])
                 },
                 retry_policy=ANY,
                 start_to_close_timeout=ANY

tests/laborious/workflows/test_predictions_batch.py

@@ -32,7 +32,8 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch
         'query': 'SELECT * FROM test',
         'schema': 'test_schema',
         'table_name': 'test_table',
-        'opc_output_config': 'test_opc_output_config'
+        'opc_output_config': 'test_opc_output_config',
+        'datetime_columns': ['timestamp', 'created_at']
     }
 
     await predictions_batch.run(input_data)
@@ -43,6 +44,7 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch
             {
                 **metadata,
                 'query': input_data['query'],
+                'datetime_columns': input_data.get('datetime_columns', [])
             },
             retry_policy=ANY,
             start_to_close_timeout=ANY

values.yaml

@@ -11,7 +11,7 @@ image:
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.4.2"
+  tag: "0.4.4"
 
 # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets:
@@ -151,7 +151,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1199-revisar-e-testar-observabilidade"
+    value: "SIENTIAPDE-1193-conferir-como-a-escrita-de-datetime-ocorre-no-temporal"
   - name: PYTHON_APP
     value: "laborious.worker.worker"