feat: add support for notifying external workflows upon simulation completion via CallbackWorkflowId

main
Cizz22 2 months ago
parent a59ae49c1b
commit 7df6a24cbc

@ -28,6 +28,7 @@ class SimulationInput(DefultBase):
AhmJobId: Optional[str] = Field(None, max_length=50)
deratings: Optional[List[dict]] = []
maintenance_outages: Optional[List[dict]] = []
CallbackWorkflowId: Optional[str] = Field(None, max_length=100)
class SimulationNode(DefultBase):
id: UUID

@ -1,10 +1,11 @@
from datetime import timedelta
from temporalio import workflow
from temporalio.client import Client
import os
with workflow.unsafe.imports_passed_through():
from temporal.activity import calculate_plant_eaf_activity, execute_simulation_activity, update_contribution_bulk_mappings_activity, update_equipment_for_simulation_activity,call_callback_ahm, acquire_lock, release_lock,execute_update_node
TEMPORAL_URL = os.environ.get("TEMPORAL_URL", "http://192.168.1.86:7233")
@workflow.defn
class SimulationWorkflow:
@ -79,9 +80,21 @@ class SimulationWorkflow:
self.status = "COMPLETED"
# etc…
self.isDone = True
await workflow.execute_activity(
release_lock,
sim_data["HubCnnId"],
start_to_close_timeout=timedelta(minutes=10),
)
if sim_data.get("CallbackWorkflowId"):
client = await Client.connect(TEMPORAL_URL)
handle = client.get_workflow_handle(sim_data["CallbackWorkflowId"])
await handle.signal("notify_done", {
"simulation_id": sim_data["HubCnnId"],
"status": "success"
})
return sim_data["HubCnnId"] # simulation_id

Loading…
Cancel
Save