3083c5edc4
Update version in init_orchestration.ipynb to 3.11.14 and enhance README.md with detailed architecture principles, execution flows, and improved filtering mechanisms in workflows. Refactor orchestrator_functions.py for clarity in filter configuration and add support for PI Web API output configuration in predictions_batch. Update tests to reflect new configurations.
212 lines
6.5 KiB
Plaintext
212 lines
6.5 KiB
Plaintext
{
|
|
"cells": [
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 5,
|
|
"id": "3f8b77a4",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"from temporalio import client\n",
|
|
"from orchestrator.activities.temporal_manager import TemporalManager\n",
|
|
"import os\n",
|
|
"from unittest.mock import MagicMock\n",
|
|
"\n",
|
|
"host = \"localhost:7233\"\n",
|
|
"logger = MagicMock(info=MagicMock(side_effect=print), debug=MagicMock(side_effect=print))\n",
|
|
"\n",
|
|
"temporal_client = await client.Client.connect(\n",
|
|
" target_host=host,\n",
|
|
" namespace=os.getenv('TEMPORAL_NAMESPACE', 'default')\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 6,
|
|
"id": "d9d2a242",
|
|
"metadata": {},
|
|
"outputs": [
|
|
{
|
|
"data": {
|
|
"text/plain": [
|
|
"<temporalio.client.ScheduleHandle at 0x788340e41b90>"
|
|
]
|
|
},
|
|
"execution_count": 6,
|
|
"metadata": {},
|
|
"output_type": "execute_result"
|
|
}
|
|
],
|
|
"source": [
|
|
"from datetime import timedelta\n",
|
|
"from temporalio.client import (\n",
|
|
" Schedule,\n",
|
|
" ScheduleActionStartWorkflow,\n",
|
|
" ScheduleIntervalSpec,\n",
|
|
" ScheduleSpec,\n",
|
|
")\n",
|
|
"\n",
|
|
"# temporal operator search-attribute create --namespace scouter --name model_id --type Text && temporal operator search-attribute create --namespace scouter --name orchestrated --type Text && temporal operator search-attribute create --namespace scouter --name model_name --type Text && temporal operator search-attribute create --namespace laborious --name model_id --type Text && temporal operator search-attribute create --namespace laborious --name orchestrated --type Text && temporal operator search-attribute create --namespace laborious --name model_name --type Text\n",
|
|
"\n",
|
|
"\n",
|
|
"\n",
|
|
"await temporal_client.create_schedule(\n",
|
|
" \"orchestrator\",\n",
|
|
" Schedule(\n",
|
|
" action=ScheduleActionStartWorkflow(\n",
|
|
" 'orchestrator',\n",
|
|
" {\n",
|
|
" \"schedule_name\": \"orchestrator\",\n",
|
|
" \"pipelines_query\": {\n",
|
|
" \"collection\": \"pipelines\",\n",
|
|
" \"aggregation\": [\n",
|
|
" {\n",
|
|
" \"$lookup\": {\n",
|
|
" \"from\": \"models\",\n",
|
|
" \"localField\": \"model_id\",\n",
|
|
" \"foreignField\": \"id\",\n",
|
|
" \"as\": \"model\"\n",
|
|
" }\n",
|
|
" },\n",
|
|
" {\n",
|
|
" \"$unwind\": \"$model\"\n",
|
|
" },\n",
|
|
" {\n",
|
|
" \"$match\": {\n",
|
|
" \"active\": True\n",
|
|
" }\n",
|
|
" },\n",
|
|
" {\n",
|
|
" \"$match\": {\n",
|
|
" \"model.active\": True\n",
|
|
" }\n",
|
|
" }\n",
|
|
" ]\n",
|
|
" },\n",
|
|
" \"opc_servers_query\": {\n",
|
|
" \"collection\": \"opc-servers\",\n",
|
|
" \"filters\": {\n",
|
|
"\n",
|
|
" }\n",
|
|
" }\n",
|
|
" },\n",
|
|
" id=\"orchestrator\",\n",
|
|
" task_queue=\"orchestrator-queue\",\n",
|
|
" execution_timeout=timedelta(minutes=600)\n",
|
|
" ),\n",
|
|
" spec=ScheduleSpec(\n",
|
|
" intervals=[ScheduleIntervalSpec(every=timedelta(minutes=60))]\n",
|
|
" )\n",
|
|
" )\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 8,
|
|
"id": "6ea0f616",
|
|
"metadata": {},
|
|
"outputs": [
|
|
{
|
|
"data": {
|
|
"text/plain": [
|
|
"<temporalio.client.ScheduleHandle at 0x78834037c3d0>"
|
|
]
|
|
},
|
|
"execution_count": 8,
|
|
"metadata": {},
|
|
"output_type": "execute_result"
|
|
}
|
|
],
|
|
"source": [
|
|
"await temporal_client.create_schedule(\n",
|
|
" \"alerts\",\n",
|
|
" Schedule(\n",
|
|
" action=ScheduleActionStartWorkflow(\n",
|
|
" 'alerts',\n",
|
|
" {\n",
|
|
" \"schedule_name\": \"alerts\",\n",
|
|
" \"notification_ttl\": 5*60,\n",
|
|
" \"sent_ttl\": 10*60\n",
|
|
" },\n",
|
|
" id=\"alerts\",\n",
|
|
" task_queue=\"alerts-queue\",\n",
|
|
" execution_timeout=timedelta(minutes=600)\n",
|
|
" ),\n",
|
|
" spec=ScheduleSpec(\n",
|
|
" intervals=[ScheduleIntervalSpec(every=timedelta(seconds=30))]\n",
|
|
" )\n",
|
|
" )\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 7,
|
|
"id": "05c46ec8",
|
|
"metadata": {},
|
|
"outputs": [
|
|
{
|
|
"data": {
|
|
"text/plain": [
|
|
"<temporalio.client.ScheduleHandle at 0x78832fd30dd0>"
|
|
]
|
|
},
|
|
"execution_count": 7,
|
|
"metadata": {},
|
|
"output_type": "execute_result"
|
|
}
|
|
],
|
|
"source": [
|
|
"await temporal_client.create_schedule(\n",
|
|
" \"reports\",\n",
|
|
" Schedule(\n",
|
|
" action=ScheduleActionStartWorkflow(\n",
|
|
" 'reports',\n",
|
|
" {\n",
|
|
" \"schedule_name\": \"reports\",\n",
|
|
" },\n",
|
|
" id=\"reports\",\n",
|
|
" task_queue=\"reports-queue\",\n",
|
|
" execution_timeout=timedelta(minutes=600)\n",
|
|
" ),\n",
|
|
" spec=ScheduleSpec(\n",
|
|
" intervals=[ScheduleIntervalSpec(every=timedelta(minutes=20))]\n",
|
|
" )\n",
|
|
" )\n",
|
|
")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"id": "76e3d8a7",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": []
|
|
}
|
|
],
|
|
"metadata": {
|
|
"kernelspec": {
|
|
"display_name": "venv",
|
|
"language": "python",
|
|
"name": "python3"
|
|
},
|
|
"language_info": {
|
|
"codemirror_mode": {
|
|
"name": "ipython",
|
|
"version": 3
|
|
},
|
|
"file_extension": ".py",
|
|
"mimetype": "text/x-python",
|
|
"name": "python",
|
|
"nbconvert_exporter": "python",
|
|
"pygments_lexer": "ipython3",
|
|
"version": "3.11.14"
|
|
}
|
|
},
|
|
"nbformat": 4,
|
|
"nbformat_minor": 5
|
|
}
|