diff --git a/coverage.sh b/coverage.sh new file mode 100755 index 0000000..5867692 --- /dev/null +++ b/coverage.sh @@ -0,0 +1 @@ +pytest --cov=laborious --cov-report=html && xdg-open htmlcov/index.html \ No newline at end of file diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 32d1006..919a2b2 100644 --- a/laborious/activities/gates.py +++ b/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]): diff --git a/laborious/workflows/minimal_retrain.py b/laborious/workflows/minimal_retrain.py index b3da70a..0baf647 100644 --- a/laborious/workflows/minimal_retrain.py +++ b/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) diff --git a/laborious/workflows/predictions_batch.py b/laborious/workflows/predictions_batch.py index a4b14f0..263e893 100644 --- a/laborious/workflows/predictions_batch.py +++ b/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) diff --git a/laborious/workflows/sub_workflows/format_and_export_prediction.py b/laborious/workflows/sub_workflows/format_and_export_prediction.py index a5c3afd..ee7c424 100644 --- a/laborious/workflows/sub_workflows/format_and_export_prediction.py +++ b/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) diff --git a/requirements.txt b/requirements.txt index fdc9371..5a8d019 100644 --- a/requirements.txt +++ b/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 diff --git a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py index 648a38f..54909a6 100644 --- a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py +++ b/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 diff --git a/tests/laborious/workflows/test_minimal_retrain.py b/tests/laborious/workflows/test_minimal_retrain.py index bcb2288..b3b03b5 100644 --- a/tests/laborious/workflows/test_minimal_retrain.py +++ b/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 diff --git a/tests/laborious/workflows/test_predictions_batch.py b/tests/laborious/workflows/test_predictions_batch.py index 3c8fc82..e9b9bb6 100644 --- a/tests/laborious/workflows/test_predictions_batch.py +++ b/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 diff --git a/values.yaml b/values.yaml index 25180fb..0a59346 100644 --- a/values.yaml +++ b/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"