Coverage for async_durable_execution/_primitive/invoke.py: 97%
52 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-30 23:43 +0000
1"""Implement the Durable invoke operation."""
3from __future__ import annotations
5import asyncio
6import logging
7from typing import TypeVar, cast
9# Import base classes for operation executor pattern
10from .base import OperationExecutor
11from .._core import (
12 DEFAULT_JSON_SERDES,
13 CallableRuntimeError,
14 ChainedInvokeOptions,
15 DurableContext,
16 ExecutionError,
17 ExecutionState,
18 Operation,
19 OperationIdentifier,
20 OperationStatus,
21 OperationType,
22 OperationUpdate,
23 SerDes,
24 suspend_with_optional_resume_delay,
25)
27P = TypeVar("P") # Payload type
28R = TypeVar("R") # Result type
30logger = logging.getLogger(__name__)
33class InvokeOperationExecutor(OperationExecutor[R]):
34 """Executor for invoke operations."""
36 SERDES_OPERATION_TYPE = OperationType.CHAINED_INVOKE
38 def __init__(
39 self,
40 function_name: str,
41 payload: P,
42 state: ExecutionState,
43 operation_identifier: OperationIdentifier,
44 serdes_payload: SerDes[P] | None = None,
45 serdes_result: SerDes[R] | None = None,
46 tenant_id: str | None = None,
47 ) -> None:
48 """Initialize the invoke operation executor.
50 Args:
51 function_name: Name of the function to invoke
52 payload: The payload to pass to the invoked function
53 state: The execution state
54 operation_identifier: The operation identifier
55 serdes_payload: Optional serializer for the invocation payload
56 serdes_result: Optional deserializer for the invocation result
57 tenant_id: Optional tenant identifier for the chained invocation
58 """
59 super().__init__(state=state, operation_identifier=operation_identifier)
60 self.function_name = function_name
61 self.payload = payload
62 self.serdes_payload = serdes_payload
63 self.serdes_result = serdes_result
64 self.tenant_id = tenant_id
66 async def start(self) -> R:
67 """Start a new invoke operation."""
68 serialized_payload: str = await self.serialize_value(
69 value=self.payload,
70 serdes=self.serdes_payload or DEFAULT_JSON_SERDES,
71 )
72 start_operation: OperationUpdate = OperationUpdate.create_invoke_start(
73 identifier=self.operation_identifier,
74 payload=serialized_payload,
75 chained_invoke_options=ChainedInvokeOptions(
76 function_name=self.function_name,
77 tenant_id=self.tenant_id,
78 ),
79 )
80 await self.create_checkpoint(start_operation, is_sync=True)
82 logger.debug(
83 "🚀 Invoke %s started, will suspend for completion",
84 self.operation_name or self.function_name,
85 )
87 return await self.execute()
89 async def replay(self, operation: Operation) -> R:
90 """Replay an existing invoke operation from its checkpoint."""
91 invoke_details = operation.chained_invoke_details
92 if operation.status is OperationStatus.SUCCEEDED:
93 result_data = invoke_details.result if invoke_details else None
94 if result_data is None:
95 return cast("R", None)
97 result: R = await self.deserialize_value(
98 data=result_data,
99 serdes=self.serdes_result or DEFAULT_JSON_SERDES,
100 operation=operation,
101 )
102 return result
104 # Terminal failures
105 if (
106 operation.status is OperationStatus.FAILED
107 or operation.status is OperationStatus.TIMED_OUT
108 or operation.status is OperationStatus.STOPPED
109 ):
110 error = invoke_details.error if invoke_details else None
111 if error is None:
112 raise CallableRuntimeError(
113 message="Unknown error. No ErrorObject exists on the Checkpoint Operation.",
114 error_type=None,
115 data=None,
116 stack_trace=None,
117 )
119 raise CallableRuntimeError.from_error_object(error)
121 if operation.status is OperationStatus.STARTED:
122 logger.debug(
123 "⏳ Invoke %s still in progress, will suspend",
124 self.operation_name or self.function_name,
125 )
126 return await self.execute()
128 return await self.execute()
130 async def execute(self, operation: Operation | None = None) -> R:
131 """Execute invoke operation by suspending to wait for async completion.
133 The invoke operation doesn't execute synchronously - it suspends and
134 the backend executes the invoked function asynchronously.
136 Returns:
137 Never returns - always suspends
139 Raises:
140 Always suspends via suspend_with_optional_resume_delay
141 ExecutionError: If suspend doesn't raise (should never happen)
142 """
143 msg: str = f"Invoke {self.operation_identifier.operation_id} started, suspending for completion"
144 suspend_with_optional_resume_delay(msg)
145 # This line should never be reached since suspend_with_optional_resume_delay always raises
146 error_msg: str = "suspend_with_optional_resume_delay should have raised an exception, but did not."
147 raise ExecutionError(error_msg) from None
150def invoke(
151 function_name: str,
152 payload: P,
153 *,
154 name: str | None = None,
155 serdes_payload: SerDes[P] | None = None,
156 serdes_result: SerDes[R] | None = None,
157 tenant_id: str | None = None,
158) -> asyncio.Task[R]:
159 """Compatibility import for the canonical operation-layer helper."""
160 from .._operation.invoke import invoke as operation_invoke
162 return operation_invoke(
163 function_name,
164 payload,
165 name=name,
166 serdes_payload=serdes_payload,
167 serdes_result=serdes_result,
168 tenant_id=tenant_id,
169 )
172async def _invoke(
173 function_name: str,
174 payload: P,
175 *,
176 context: DurableContext,
177 operation_identifier: OperationIdentifier,
178 serdes_payload: SerDes[P] | None = None,
179 serdes_result: SerDes[R] | None = None,
180 tenant_id: str | None = None,
181) -> R:
182 executor: InvokeOperationExecutor[R] = InvokeOperationExecutor(
183 function_name=function_name,
184 payload=payload,
185 state=context.execution_state,
186 operation_identifier=operation_identifier,
187 serdes_payload=serdes_payload,
188 serdes_result=serdes_result,
189 tenant_id=tenant_id,
190 )
191 return await executor.process()