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

SIENTIAPDE-1478

Browse Source 
Refactor validation script and improve logging in API and model repository

- Updated validation script to include 'e2e/' directory in code formatting and linting checks.
- Enhanced error logging in API class to improve readability of error messages.
- Refactored debug logging in model repository for better structured output.
- Cleaned up import statements in various files for improved organization.
This commit is contained in:
vitor-aignosi
2026-01-16 16:47:33 -03:00
parent 241f283724
commit 69b2f93ab2
8 changed files with 42 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
laborious/activities/api.py
View File

@@ -148,7 +148,10 @@ class API(SientiaMonitoring):
if len(written_tags) != len(tag_names):
message = f'The number of written tags does not match the number of tag names: Expected {tag_names} tags, but {written_tags} tags were written.'
self.error(f"{message}\nResponse:\n {json.dumps(response_data, indent=4)}\nTags:\n {json.dumps(tags, indent=4)}", metadata)
self.error(
f'{message}\nResponse:\n {json.dumps(response_data, indent=4)}\nTags:\n {json.dumps(tags, indent=4)}',
metadata,
)
await self.send_notification_async(
metadata=metadata,

2
laborious/activities/gates.py
View File

@@ -6,13 +6,13 @@ with workflow.unsafe.imports_passed_through():
from typing import Any
from pandas import DataFrame
from sientia_do.utils.formatters import create_sample_dict
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, now
from sientia_do.utils.formatters import create_sample_dict
from laborious import metrics
from laborious.utils.filters.conditional_filters import (

3
laborious/activities/mlflow.py
View File

@@ -6,7 +6,6 @@ with workflow.unsafe.imports_passed_through():
import numpy as np
from pandas import DataFrame, to_datetime
from sientia_do.utils.formatters import create_sample_dict
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
@@ -18,6 +17,7 @@ with workflow.unsafe.imports_passed_through():
DATETIME_FORMAT_WITH_TZ,
now,
)
from sientia_do.utils.formatters import create_sample_dict
from laborious.utils.repository.minio_repository import MinioRepository
from laborious.utils.repository.model_repository import MLFlowRepository
@@ -159,7 +159,6 @@ class MLFlow(SientiaMonitoring):
self.debug(f'Processed input data: \n {data.to_csv()}', metadata)
# Request transformation from MLFlow model
response_data = await self.model_monitoring_repository.transform(
model_name, data, model_config, metadata

7
laborious/utils/repository/model_repository.py
View File

@@ -1254,7 +1254,9 @@ class MLFlowRepository(SientiaMonitoring):
input_index = data.index
start_time = datetime.now()
self.debug(f'Data received for model prediction: {data.to_dict(orient="records")}', metadata)
self.debug(
f'Data received for model prediction: {data.to_dict(orient="records")}', metadata
)
# data.to_csv(
# f"tmp/treated_data_{model_name}.csv", index=True)
@@ -1272,7 +1274,8 @@ class MLFlowRepository(SientiaMonitoring):
if isinstance(predict_data, pd.DataFrame):
self.debug(
f'Data received from model prediction: {predict_data.to_dict(orient="records")}', metadata
f'Data received from model prediction: {predict_data.to_dict(orient="records")}',
metadata,
)
# predict_data.to_csv(

16
laborious/worker/worker.py
View File

@@ -31,27 +31,20 @@ Environment Variables:
from temporalio import client, workflow
from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig
from temporalio.worker import (
PollerBehaviorAutoscaling,
ResourceBasedSlotConfig,
Worker,
WorkerTuner,
)
with workflow.unsafe.imports_passed_through():
import asyncio
import os
import sys
from datetime import timedelta
from prometheus_client import start_http_server
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import get_logger
from sientia_do.utils.connectors_config import (
build_api_config,
build_mongodb_config,
build_postgres_config,
)
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import get_logger
from laborious import metrics
from laborious.activities.activities import Activities
@@ -60,6 +53,7 @@ with workflow.unsafe.imports_passed_through():
build_mlflow_config,
build_opc_config,
)
from laborious.worker.prepare_worker import prepare_worker
from laborious.workflows.drift import Drift
from laborious.workflows.minimal_retrain import MinimalRetrain
from laborious.workflows.predictions_batch import PredictionsBatch
@@ -68,11 +62,11 @@ with workflow.unsafe.imports_passed_through():
FormatAndExportPrediction,
)
from laborious.workflows.sub_workflows.prediction_process import PredictionProcess
from laborious.worker.prepare_worker import prepare_worker
POD_ID = os.getenv('POD_ID')
SDK_METRICS_PORT = int(os.getenv('HTTP_SDK_METRICS_PORT', '9091'))
async def main():
"""
Main entry point for the Laborious worker application.
@@ -215,7 +209,7 @@ async def main():
activities.write_metrics,
],
logger=logger,
)
),
]
handlers = []

21
tests/laborious/activities/test_api.py
View File

@@ -333,11 +333,13 @@ async def test_process_pi_web_api_response_with_errors(api):
)
assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE
assert message == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag2'] tags were written."
assert (
message
== "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag2'] tags were written."
)
assert api.emit_metric.call_count == 2
@mark.asyncio
async def test_process_pi_web_api_response_missing_tags(api):
"""Test processing response when number of written tags doesn't match expected."""
@@ -361,7 +363,10 @@ async def test_process_pi_web_api_response_missing_tags(api):
)
assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE
assert message == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag1'] tags were written."
assert (
message
== "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag1'] tags were written."
)
api.send_notification_async.assert_called_once()
call_args = api.send_notification_async.call_args
assert call_args.kwargs['notification_id'] == 'WRITE_PI_WEB_API_PREDICTION_ERROR'
@@ -392,7 +397,10 @@ async def test_process_pi_web_api_response_missing_webid(api):
)
assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE
assert message == "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag2'] tags were written."
assert (
message
== "The number of written tags does not match the number of tag names: Expected ['tag1', 'tag2'] tags, but ['tag2'] tags were written."
)
api.error.assert_any_call('The response did not contain some WebIds', metadata['metadata'])
@@ -419,7 +427,10 @@ async def test_process_pi_web_api_response_missing_tag_name(api):
)
assert confidence == PI_WEB_API_PREDICTION_ERROR_CONFIDENCE
assert message == "The number of written tags does not match the number of tag names: Expected ['tag1'] tags, but [] tags were written."
assert (
message
== "The number of written tags does not match the number of tag names: Expected ['tag1'] tags, but [] tags were written."
)
api.error.assert_any_call(
'The response did not contain the tag name for WebId unknown_web_id', metadata['metadata']
)

9
tests/laborious/utils/repository/test_model_repository.py
View File

@@ -812,7 +812,14 @@ async def test_fit_models_not_df_target_name_none_and_not_in_model(
data = MagicMock()
output = await mlflow_repository.fit_models(
'model_name', data, 'latest_production_id', metadata['metadata'], 'sklearn', False, 'pyfunc', None
'model_name',
data,
'latest_production_id',
metadata['metadata'],
'sklearn',
False,
'pyfunc',
None,
)
mlflow_repository.download_model.assert_has_calls(

4
validate.sh
View File

@@ -70,14 +70,14 @@ FAILED_STEPS=()
# Step 1: Code Formatting Check (Ruff)
# - default: check only
# - --fix: write changes
if ! run_step "1. Code Formatting (Ruff)" "if \$FIX_MODE; then ruff format laborious/ tests/; else ruff format --check laborious/ tests/; fi"; then
if ! run_step "1. Code Formatting (Ruff)" "if \$FIX_MODE; then ruff format laborious/ tests/; else ruff format --check laborious/ tests/ e2e/; fi"; then
FAILED_STEPS+=("Code Formatting")
fi
# Step 2: Linting (Ruff)
# - default: check only
# - --fix: apply autofixes
if ! run_step "2. Code Linting (Ruff)" "if \$FIX_MODE; then ruff check --fix laborious/ tests/; else ruff check laborious/ tests/; fi"; then
if ! run_step "2. Code Linting (Ruff)" "if \$FIX_MODE; then ruff check --fix laborious/ tests/; else ruff check laborious/ tests/ e2e/; fi"; then
FAILED_STEPS+=("Linting")
fi