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

SIENTIAPDE-1150

Browse Source 
refactor: streamline MongoDB activity methods to improve readability and maintainability; replace inline query construction with variable assignments for updated, created, and deleted pipelines
This commit is contained in:
vitor-aignosi
2025-07-14 16:16:29 -03:00
parent f823d1813f
commit 7f162a7d4e
1 changed files with 23 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
orchestrator/activities/mongo_db.py
View File

@@ -198,12 +198,14 @@ class MongoDB(BaseActivity):
updated_pipelines = input_data.get("updated_pipelines", [])
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
argument = [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"]}
for pipeline in updated_pipelines if pipeline["success"]
]
data_filter = {"$or": argument} if argument else {}
self.database["pipelines"].update_many(
{"$or": [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"]}
for pipeline in updated_pipelines if pipeline["success"]
]},
data_filter,
{"$set": {"updated_at": now}}
)
@@ -218,11 +220,14 @@ class MongoDB(BaseActivity):
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
self.database["pipelines"].insert_many([{
"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"],
"updated_at": now
} for pipeline in created_pipelines if pipeline["success"]])
argument = [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"],
"updated_at": now}
for pipeline in created_pipelines if pipeline["success"]
]
data_filter = argument if argument else {}
self.database["pipelines"].insert_many(data_filter)
@activity.defn(name="delete_pipelines_timestamps")
async def delete_pipelines_timestamps(self, input_data: dict[str, Any]) -> None:
@@ -233,10 +238,11 @@ class MongoDB(BaseActivity):
"""
deleted_pipelines = input_data.get("deleted_pipelines", [])
self.database["pipelines"].delete_many({
"$or": [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"]}
for pipeline in deleted_pipelines if pipeline["success"]
]
})
argument = [
{"schedule_name": pipeline["schedule_name"],
"namespace": pipeline["namespace"]}
for pipeline in deleted_pipelines if pipeline["success"]
]
data_filter = {"$or": argument} if argument else {}
self.database["pipelines"].delete_many(data_filter)