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

SIENTIAPDE-1141

Browse Source 
fix: update host configuration and execution counts in test.ipynb; refactor schedule creation parameters in orchestrator
This commit is contained in:
vitor-aignosi
2025-07-09 16:22:15 -03:00
parent 4df1b0240a
commit bb5d0f15f9
2 changed files with 58 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
orchestrator/worker/worker.py
View File

@@ -41,7 +41,6 @@ async def main():
activities = Activities( activities = Activities(
temporal_client=temporal_client, temporal_client=temporal_client,
# couchbase_config=build_couchbase_config(),
redis_config=build_redis_config(), redis_config=build_redis_config(),
mongodb_config=build_mongodb_config(), mongodb_config=build_mongodb_config(),
logger=logger, logger=logger,

75
test.ipynb
View File

@@ -120,7 +120,7 @@
"import os\n", "import os\n",
"from unittest.mock import MagicMock\n", "from unittest.mock import MagicMock\n",
"\n", "\n",
"host = \"localhost:7233\"\n", "host = \"localhost:44795\"\n",
"logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n", "logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n",
"\n", "\n",
"temporal_client = await client.Client.connect(\n", "temporal_client = await client.Client.connect(\n",
@@ -136,17 +136,17 @@
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 6, "execution_count": 4,
"id": "bb750ae6", "id": "bb750ae6",
"metadata": {}, "metadata": {},
"outputs": [ "outputs": [
{ {
"data": { "data": {
"text/plain": [ "text/plain": [
"<temporalio.client.ScheduleHandle at 0x7dfad094e610>" "<temporalio.client.ScheduleHandle at 0x7ddf807199d0>"
] ]
}, },
"execution_count": 6, "execution_count": 4,
"metadata": {}, "metadata": {},
"output_type": "execute_result" "output_type": "execute_result"
} }
@@ -163,29 +163,70 @@
")\n", ")\n",
"from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n", "from temporalio.common import TypedSearchAttributes, SearchAttributeKey, SearchAttributePair\n",
"\n", "\n",
"# temporal operator search-attribute create --namespace default --name model_id --type Text\n",
"# temporal operator search-attribute create --namespace default --name orchestrated --type Text\n",
"# temporal operator search-attribute create --namespace default --name model_name --type Text\n",
"\n", "\n",
"customer_id_key = SearchAttributeKey.for_keyword(\"orchestrated\")\n",
"search_attributes = TypedSearchAttributes([\n",
" SearchAttributePair(customer_id_key, \"true\")\n",
"])\n",
"await temporal_client.create_schedule(\n", "await temporal_client.create_schedule(\n",
" \"meu-schedule-id5\",\n", " \"orchestrator\",\n",
" Schedule(\n", " Schedule(\n",
" action=ScheduleActionStartWorkflow(\n", " action=ScheduleActionStartWorkflow(\n",
" 'scouter-test2',\n", " 'orchestrator',\n",
" {\n", " {\n",
" 'args': {\n", " \"schedule_name\": \"orchestrator-test\",\n",
" 'arg1': 'value1'\n", " \"pipelines_query\": {\n",
" \"collection\": \"pipelines\",\n",
" \"aggregation\": [\n",
" {\n",
" \"$lookup\": {\n",
" \"from\": \"models\",\n",
" \"localField\": \"model_id\",\n",
" \"foreignField\": \"id\",\n",
" \"as\": \"model_docs\"\n",
" }\n",
" },\n",
" {\n",
" \"$match\": {\n",
" \"models.active\": True\n",
" }\n",
" },\n",
" {\n",
" \"$match\": {\n",
" \"pipelines.active\": True\n",
" }\n",
" },\n",
" {\n",
" \"$addFields\": {\n",
" \"models\": {\n",
" \"$arrayElemAt\": [\n",
" \"$model_docs\",\n",
" 0\n",
" ]\n",
" }\n",
" }\n",
" },\n",
" {\n",
" \"$project\": {\n",
" \"model_docs\": 0\n",
" }\n",
" }\n",
" ]\n",
" },\n",
" \"opc_servers_query\": {\n",
" \"collection\": \"opc-servers\",\n",
" \"filters\": {\n",
"\n",
" }\n",
" }\n", " }\n",
" },\n", " },\n",
" id=\"workflow-id-unico\",\n", " id=\"orchestrator\",\n",
" task_queue=\"nome-da-task-queue\",\n", " task_queue=\"orchestrator-queue\",\n",
" execution_timeout=timedelta(minutes=600)\n",
" ),\n", " ),\n",
" spec=ScheduleSpec(\n", " spec=ScheduleSpec(\n",
" intervals=[ScheduleIntervalSpec(every=timedelta(minutes=10))]\n", " intervals=[ScheduleIntervalSpec(every=timedelta(minutes=60))]\n",
" )\n", " )\n",
" ),\n", " )\n",
" search_attributes=search_attributes,\n",
")\n" ")\n"
] ]
}, },