diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index bd9c548..312729e 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -202,7 +202,7 @@ class MongoDB(BaseActivity): {"$or": [ {"schedule_name": pipeline["schedule_name"], "namespace": pipeline["namespace"]} - for pipeline in updated_pipelines + for pipeline in updated_pipelines if pipeline["success"] ]}, {"$set": {"updated_at": now}} ) @@ -222,7 +222,7 @@ class MongoDB(BaseActivity): "schedule_name": pipeline["schedule_name"], "namespace": pipeline["namespace"], "updated_at": now - } for pipeline in created_pipelines]) + } for pipeline in created_pipelines if pipeline["success"]]) @activity.defn(name="delete_pipelines_timestamps") async def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None: @@ -237,6 +237,6 @@ class MongoDB(BaseActivity): "$or": [ {"schedule_name": pipeline["schedule_name"], "namespace": pipeline["namespace"]} - for pipeline in deleted_pipelines + for pipeline in deleted_pipelines if pipeline["success"] ] }) diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index d195446..afb24fc 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -199,5 +199,35 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) + update_pipelines_timestamps_handler = workflow.execute_activity_method( + Activities.update_pipelines_timestamps, + { + 'updated_pipelines': schedule_update_report + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + create_pipelines_timestamps_handler = workflow.execute_activity_method( + Activities.create_pipelines_timestamps, + { + 'created_pipelines': schedule_insertion_report + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + delete_pipelines_timestamps_handler = workflow.execute_activity_method( + Activities.delete_pipelines_timestamps, + { + 'deleted_pipelines': schedule_deletion_report + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + await schedule_report_handler await slot_report_handler + await update_pipelines_timestamps_handler + await create_pipelines_timestamps_handler + await delete_pipelines_timestamps_handler diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 3db578e..7beb29d 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -250,8 +250,8 @@ async def test_aggregate_documents_in_mongodb_missing_aggregation(mongo_db): @patch("orchestrator.activities.mongo_db.datetime") async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): input_data = {"updated_pipelines": [ - {"schedule_name": "test1", "namespace": "test1"}, - {"schedule_name": "test2", "namespace": "test2"} + {"schedule_name": "test1", "namespace": "test1", "success": True}, + {"schedule_name": "test2", "namespace": "test2", "success": True} ]} mongo_db.database["pipelines"].update_many.return_value = MagicMock() await mongo_db.update_pipelines_timestamps(input_data) @@ -269,8 +269,8 @@ async def test_update_pipelines_timestamps_success(datetime_mock, mongo_db): @patch("orchestrator.activities.mongo_db.datetime") async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): input_data = {"created_pipelines": [ - {"schedule_name": "test1", "namespace": "test1"}, - {"schedule_name": "test2", "namespace": "test2"} + {"schedule_name": "test1", "namespace": "test1", "success": True}, + {"schedule_name": "test2", "namespace": "test2", "success": True} ]} mongo_db.database["pipelines"].insert_many.return_value = MagicMock() await mongo_db.create_pipelines_timestamps(input_data) @@ -288,8 +288,8 @@ async def test_create_pipelines_timestamps_success(datetime_mock, mongo_db): @patch("orchestrator.activities.mongo_db.datetime") async def test_delete_pipelines_timestamps_success(datetime_mock, mongo_db): input_data = {"deleted_pipelines": [ - {"schedule_name": "test1", "namespace": "test1"}, - {"schedule_name": "test2", "namespace": "test2"} + {"schedule_name": "test1", "namespace": "test1", "success": True}, + {"schedule_name": "test2", "namespace": "test2", "success": True} ]} mongo_db.database["pipelines"].delete_many.return_value = MagicMock() await mongo_db.delete_pipelines_timestamps(input_data) diff --git a/values.yaml b/values.yaml index bff3de6..422f589 100644 --- a/values.yaml +++ b/values.yaml @@ -123,7 +123,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1148-separar-scouter-laborious-e-orchestrator-por-namespaces" + value: "SIENTIAPDE-1150-ajustar-orquestrador-para-manter-uma-store-de-detalhes-dos-schedules" - name: PYTHON_APP value: "orchestrator.worker.worker"