Merge pull request #11 from Aignosi/SIENTIAPDE-1171-criar-pipeline-de-retreino-laborious

Sientiapde 1171 criar pipeline de retreino laborious

orchestrator/activities/formatters.py

@@ -1,6 +1,8 @@
 
 from temporalio import activity, workflow
 
+from orchestrator.utils.orchestrator_functions import minimal_retrain
+
 
 with workflow.unsafe.imports_passed_through():
     import json
@@ -67,6 +69,13 @@ class Formatters(BaseActivity):
                     "updated_at": pipeline.get(
                         "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT))
                 }
+            elif pipeline['workflow_type'] == 'minimal_retrain':
+                schedule_config[self.laborious_namespace][pipeline['schedule_name']
+                                                          ] = {
+                    **minimal_retrain(pipeline),
+                    "updated_at": pipeline.get(
+                        "updated_at", datetime.now().strftime(DEFAULT_DATE_FORMAT))
+                }
 
         self.info("Processed schedules", metadata=metadata)
         self.debug(json.dumps(

orchestrator/utils/orchestrator_functions.py

@@ -13,6 +13,17 @@ def common_config(config: dict[str, Any]):
     }
 
 
+def minimal_retrain(config: dict[str, Any]):
+    return {
+        **common_config(config),
+        "workflow_type": "minimal_retrain",
+        "schedule_name": config['schedule_name'],
+        "query": config['query'],
+        "schema": "sientia_data",
+        "table_name": "log_retrain",
+    }
+
+
 def scouter(config: dict[str, Any]):
     filters = {}
     for f in config.get('filters', []):

samples.json

@@ -120,7 +120,17 @@
             ],
             "active": true,
             "updated_at": "2025-07-14 10:00:00.000000"
-          }
+          },
+        "3": {
+            "schedule_name": "minimal-retrain-pipeline",
+            "model_id": "1",
+            "workflow_type": "minimal_retrain",
+            "frequency": "5m",
+            "max_retry_policy": 1,
+            "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
+            "active": true,
+            "updated_at": "2025-07-23 10:00:00.000000"
+        }
     },
     "opc-servers": {
         "1": {

tests/orchestrator/activities/test_formatters.py

@@ -34,7 +34,9 @@ metadata = {
        return_value={"test_scouter": "test_scouter"})
 @patch("orchestrator.activities.formatters.predictions_batch",
        return_value={"test_predictions_batch": "test_predictions_batch"})
-async def test_process_schedules(mock_predictions_batch, mock_scouter, formatters):
+@patch("orchestrator.activities.formatters.minimal_retrain",
+       return_value={"test_minimal_retrain": "test_minimal_retrain"})
+async def test_process_schedules(mock_minimal_retrain, mock_predictions_batch, mock_scouter, formatters):
     input_data = {
         "pipelines": [
             {
@@ -50,6 +52,13 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter
                 "model_name": "test_model_name",
                 "model_id": "test_model_id",
                 "updated_at": "2021-01-02"
+            },
+            {
+                "schedule_name": "test_schedule_name3",
+                "workflow_type": "minimal_retrain",
+                "model_name": "test_model_name",
+                "model_id": "test_model_id",
+                "updated_at": "2021-01-03"
             }
         ]
     }
@@ -67,6 +76,10 @@ async def test_process_schedules(mock_predictions_batch, mock_scouter, formatter
             "test_schedule_name2": {
                 "test_predictions_batch": "test_predictions_batch",
                 "updated_at": "2021-01-02"
+            },
+            "test_schedule_name3": {
+                "test_minimal_retrain": "test_minimal_retrain",
+                "updated_at": "2021-01-03"
             }
         }
     }

tests/orchestrator/utils/test_orchestrator_functions.py

@@ -1,6 +1,7 @@
 from unittest.mock import patch, call
 from orchestrator.utils.orchestrator_functions import (
     common_config,
+    minimal_retrain,
     scouter,
     predictions_batch,
     overlap_filter_config,
@@ -31,6 +32,31 @@ def test_common_config():
     assert result == expected
 
 
+def test_minimal_retrain():
+    config = {
+        "workflow_type": "minimal_retrain",
+        "schedule_name": "test_schedule",
+        "model_id": "test_model_id",
+        "models": {
+            "name": "test_model_name"
+        },
+        "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
+    }
+    result = minimal_retrain(config)
+    expected = {
+        "workflow_type": "minimal_retrain",
+        "schedule_name": "test_schedule",
+        "frequency": "1m",
+        "max_retry_policy": 1,
+        "model_id": "test_model_id",
+        "model_name": "test_model_name",
+        "query": "select * from sientia_data.laborious_data order by \"timestamp\" desc limit 30;",
+        "schema": "sientia_data",
+        "table_name": "log_retrain",
+    }
+    assert result == expected
+
+
 def test_scouter():
     config = {
         "workflow_type": "scouter",

values.yaml

@@ -132,7 +132,7 @@ env:
   - name: GITHUB_REPO_URL
     value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
   - name: GITHUB_BRANCH
-    value: "SIENTIAPDE-1166-alterar-orquestrador-para-criar-collections-com-ttl-no-mongo]"
+    value: "SIENTIAPDE-1171-criar-pipeline-de-retreino-laborious"
   - name: PYTHON_APP
     value: "orchestrator.worker.worker"