Coverage for async_durable_execution/_runner/cloud/__init__.py: 94%
334 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-08-30 23:43 +0000
1from __future__ import annotations
3import asyncio
4import inspect
5import json
6import logging
7import time
8from typing import Any, cast
9from uuid import uuid4
11from botocore.config import Config
12from botocore.session import get_session
14from ..._core import (
15 DurableExecutionInvocationInput,
16 DurableExecutionInvocationOutput,
17 ErrorObject,
18 create_default_async_client,
19 create_default_sync_client,
20 httpx_is_installed,
21)
22from ..exceptions import (
23 DurableFunctionsTestError,
24 InvalidParameterValueException,
25 ResourceNotFoundException,
26)
27from ..model import (
28 DurableFunctionTestResult,
29 GetDurableExecutionResponse,
30 GetDurableExecutionHistoryResponse,
31 InvokeResponse,
32 _get_callback_id_from_events,
33)
36logger = logging.getLogger(__name__)
39class ThreadedSyncCloudLambdaClient:
40 """Adapt a sync Lambda client to the async cloud runner interface."""
42 def __init__(self, client: Any) -> None:
43 self.client = client
45 @property
46 def exceptions(self) -> Any:
47 return self.client.exceptions
49 async def invoke(self, **kwargs: Any) -> dict[str, Any]:
50 return cast(
51 dict[str, Any], await asyncio.to_thread(self.client.invoke, **kwargs)
52 )
54 async def get_durable_execution(self, **kwargs: Any) -> dict[str, Any]:
55 return cast(
56 dict[str, Any],
57 await asyncio.to_thread(self.client.get_durable_execution, **kwargs),
58 )
60 async def get_durable_execution_history(self, **kwargs: Any) -> dict[str, Any]:
61 return cast(
62 dict[str, Any],
63 await asyncio.to_thread(
64 self.client.get_durable_execution_history, **kwargs
65 ),
66 )
68 async def send_durable_execution_callback_success(
69 self, **kwargs: Any
70 ) -> dict[str, Any]:
71 return cast(
72 dict[str, Any],
73 await asyncio.to_thread(
74 self.client.send_durable_execution_callback_success, **kwargs
75 ),
76 )
78 async def send_durable_execution_callback_failure(
79 self, **kwargs: Any
80 ) -> dict[str, Any]:
81 return cast(
82 dict[str, Any],
83 await asyncio.to_thread(
84 self.client.send_durable_execution_callback_failure, **kwargs
85 ),
86 )
88 async def send_durable_execution_callback_heartbeat(
89 self, **kwargs: Any
90 ) -> dict[str, Any]:
91 return cast(
92 dict[str, Any],
93 await asyncio.to_thread(
94 self.client.send_durable_execution_callback_heartbeat, **kwargs
95 ),
96 )
98 def close(self) -> None:
99 close = getattr(self.client, "close", None)
100 if callable(close): 100 ↛ exitline 100 didn't return from function 'close' because the condition on line 100 was always true
101 close()
104class AsyncCloudLambdaClient:
105 """Adapt an async Lambda client to the cloud runner interface."""
107 def __init__(self, client: Any) -> None:
108 self._client_context = client if hasattr(client, "__aenter__") else None
109 self._client = None if self._client_context is not None else client
110 self._entered_client: Any | None = None
112 @property
113 def exceptions(self) -> Any:
114 client = self._entered_client or self._client
115 if client is None:
116 msg = "Async Lambda client has not been initialized"
117 raise AttributeError(msg)
118 return client.exceptions
120 async def _get_client(self) -> Any:
121 if self._client is not None:
122 return self._client
123 if self._entered_client is None: 123 ↛ 126line 123 didn't jump to line 126 because the condition on line 123 was always true
124 assert self._client_context is not None
125 self._entered_client = await self._client_context.__aenter__()
126 return self._entered_client
128 async def _call(self, method_name: str, **kwargs: Any) -> dict[str, Any]:
129 client = await self._get_client()
130 result = getattr(client, method_name)(**kwargs)
131 if inspect.isawaitable(result): 131 ↛ 133line 131 didn't jump to line 133 because the condition on line 131 was always true
132 result = await result
133 return cast(dict[str, Any], result)
135 async def invoke(self, **kwargs: Any) -> dict[str, Any]:
136 return await self._call("invoke", **kwargs)
138 async def get_durable_execution(self, **kwargs: Any) -> dict[str, Any]:
139 return await self._call("get_durable_execution", **kwargs)
141 async def get_durable_execution_history(self, **kwargs: Any) -> dict[str, Any]:
142 return await self._call("get_durable_execution_history", **kwargs)
144 async def send_durable_execution_callback_success(
145 self, **kwargs: Any
146 ) -> dict[str, Any]:
147 return await self._call("send_durable_execution_callback_success", **kwargs)
149 async def send_durable_execution_callback_failure(
150 self, **kwargs: Any
151 ) -> dict[str, Any]:
152 return await self._call("send_durable_execution_callback_failure", **kwargs)
154 async def send_durable_execution_callback_heartbeat(
155 self, **kwargs: Any
156 ) -> dict[str, Any]:
157 return await self._call("send_durable_execution_callback_heartbeat", **kwargs)
159 async def aclose(self) -> None:
160 if self._entered_client is not None:
161 assert self._client_context is not None
162 await self._client_context.__aexit__(None, None, None)
163 self._entered_client = None
164 return
166 close = getattr(self._client, "aclose", None)
167 if callable(close): 167 ↛ exitline 167 didn't return from function 'aclose' because the condition on line 167 was always true
168 await close()
171async def _read_payload(payload: Any) -> str:
172 read = getattr(payload, "read", None)
173 if callable(read): 173 ↛ 181line 173 didn't jump to line 181 because the condition on line 173 was always true
174 if inspect.iscoroutinefunction(read): 174 ↛ 175line 174 didn't jump to line 175 because the condition on line 174 was never true
175 data = await read()
176 else:
177 data = await asyncio.to_thread(read)
178 if inspect.isawaitable(data):
179 data = await data
180 else:
181 data = payload
183 if isinstance(data, bytes): 183 ↛ 185line 183 didn't jump to line 185 because the condition on line 183 was always true
184 return data.decode("utf-8")
185 return str(data)
188def _cloud_lambda_client_is_async(client: Any) -> bool:
189 return inspect.iscoroutinefunction(getattr(client, "invoke", None))
192def adapt_lambda_client(client: Any) -> Any:
193 """Adapt a raw Lambda client to the async cloud runner interface."""
194 if isinstance(client, ThreadedSyncCloudLambdaClient | AsyncCloudLambdaClient): 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 return client
196 if _cloud_lambda_client_is_async(client): 196 ↛ 197line 196 didn't jump to line 197 because the condition on line 196 was never true
197 return AsyncCloudLambdaClient(client)
198 return ThreadedSyncCloudLambdaClient(client)
201_KNOWN_LAMBDA_ERROR_CODES = (
202 "ResourceNotFoundException",
203 "InvalidParameterValueException",
204 "TooManyRequestsException",
205 "ServiceException",
206 "ResourceConflictException",
207 "InvalidRequestContentException",
208 "RequestTooLargeException",
209 "UnsupportedMediaTypeException",
210 "InvalidRuntimeException",
211 "InvalidZipFileException",
212 "ResourceNotReadyException",
213 "SnapStartTimeoutException",
214 "SnapStartNotReadyException",
215 "SnapStartException",
216 "RecursiveInvocationException",
217 "InvalidSecurityGroupIDException",
218 "EC2ThrottledException",
219 "EFSMountConnectivityException",
220 "SubnetIPAddressLimitReachedException",
221 "EC2UnexpectedException",
222 "InvalidSubnetIDException",
223 "EC2AccessDeniedException",
224 "EFSIOException",
225 "ENILimitReachedException",
226 "EFSMountTimeoutException",
227 "EFSMountFailureException",
228 "KMSAccessDeniedException",
229 "KMSDisabledException",
230 "KMSNotFoundException",
231 "KMSInvalidStateException",
232 "DurableExecutionAlreadyStartedException",
233)
236def _aws_error_code(error: Exception, client: Any = None) -> str:
237 response = getattr(error, "response", None)
238 if isinstance(response, dict):
239 aws_error = response.get("Error")
240 if isinstance(aws_error, dict) and aws_error.get("Code"): 240 ↛ 243line 240 didn't jump to line 243 because the condition on line 240 was always true
241 return str(aws_error["Code"])
243 exceptions = getattr(client, "exceptions", None)
244 if exceptions is not None: 244 ↛ 249line 244 didn't jump to line 249 because the condition on line 244 was always true
245 for error_code in _KNOWN_LAMBDA_ERROR_CODES:
246 exception_type = getattr(exceptions, error_code, None)
247 if isinstance(exception_type, type) and isinstance(error, exception_type):
248 return error_code
249 return type(error).__name__
252def create_cloud_runner(
253 *,
254 function_name: str,
255 region: str = "us-west-2",
256 lambda_endpoint: str | None = None,
257 poll_interval: float = 1.0,
258 input: Any = None, # noqa: A002
259 timeout: int = 60,
260) -> DurableFunctionCloudTestRunner:
261 """Create a configured cloud durable function runner."""
262 return DurableFunctionCloudTestRunner(
263 function_name=function_name,
264 region=region,
265 lambda_endpoint=lambda_endpoint,
266 poll_interval=poll_interval,
267 input=input,
268 timeout=timeout,
269 )
272class DurableFunctionCloudTestRunner:
273 """Test runner that executes durable functions against actual AWS Lambda backend.
275 This runner invokes deployed Lambda functions and polls for execution completion,
276 providing the same interface as DurableFunctionLocalTestRunner for seamless test
277 compatibility between local and cloud modes.
278 """
280 def __init__(
281 self,
282 function_name: str,
283 region: str = "us-west-2",
284 lambda_endpoint: str | None = None,
285 poll_interval: float = 1.0,
286 input: Any = None, # noqa: A002
287 timeout: int = 60,
288 ) -> None:
289 """Initialize cloud test runner."""
290 self.mode = "cloud"
291 self.function_name = function_name
292 self.region = region
293 self.lambda_endpoint = lambda_endpoint
294 self.poll_interval = poll_interval
295 self._default_input = input
296 self._default_timeout = timeout
298 self.lambda_client: Any = create_lambda_client(lambda_endpoint, region)
300 async def __aenter__(self) -> DurableFunctionCloudTestRunner:
301 return self
303 async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
304 await self.aclose()
306 def close(self) -> None:
307 """Close the underlying sync client when supported."""
308 close = getattr(self.lambda_client, "close", None)
309 if callable(close):
310 close()
312 async def aclose(self) -> None:
313 """Close the underlying client when supported."""
314 aclose = getattr(self.lambda_client, "aclose", None)
315 if callable(aclose):
316 await aclose()
317 return
318 self.close()
320 async def run(
321 self,
322 ) -> DurableFunctionTestResult:
323 """Execute function on AWS Lambda and wait for completion."""
324 execution_arn = await self._invoke_for_execution(
325 invocation_type="RequestResponse",
326 expected_status_code=200,
327 )
328 return await self.wait_for_result(
329 execution_arn=execution_arn, timeout=self._default_timeout
330 )
332 async def run_async(
333 self,
334 ) -> str:
335 """Execute function on AWS Lambda asynchronously"""
336 return await self._invoke_for_execution(
337 invocation_type="Event",
338 expected_status_code=202,
339 )
341 async def _invoke_for_execution(
342 self,
343 *,
344 invocation_type: str,
345 expected_status_code: int,
346 ) -> str:
347 logger.info(
348 "Invoking Lambda function: %s (timeout: %ds)",
349 self.function_name,
350 self._default_timeout,
351 )
352 payload = json.dumps(self._default_input)
353 try:
354 response = cast(
355 dict[str, Any],
356 await self.lambda_client.invoke(
357 FunctionName=self.function_name,
358 InvocationType=invocation_type,
359 Payload=payload,
360 ),
361 )
362 except Exception as e:
363 msg = f"Failed to invoke Lambda function {self.function_name}: {e}"
364 raise DurableFunctionsTestError(msg) from e
366 status_code = response.get("StatusCode")
367 if status_code != expected_status_code:
368 error_payload = await _read_payload(response["Payload"])
369 msg = f"Lambda invocation failed with status {status_code}: {error_payload}"
370 raise DurableFunctionsTestError(msg)
372 if "FunctionError" in response: 372 ↛ 373line 372 didn't jump to line 373 because the condition on line 372 was never true
373 error_payload = await _read_payload(response["Payload"])
374 logger.warning("Lambda function failed: %s", error_payload)
376 execution_arn = cast(str | None, response.get("DurableExecutionArn"))
377 if execution_arn is None:
378 msg = (
379 f"No DurableExecutionArn in response for function {self.function_name}"
380 )
381 raise DurableFunctionsTestError(msg)
382 return execution_arn
384 async def send_callback_success(
385 self, callback_id: str, result: bytes | None = None
386 ) -> None:
387 try:
388 await self.lambda_client.send_durable_execution_callback_success(
389 CallbackId=callback_id,
390 Result=cast(Any, result),
391 )
392 except Exception as e:
393 msg = f"Failed to send callback success for {self.function_name}, callback_id {callback_id}: {e}"
394 raise DurableFunctionsTestError(msg) from e
396 async def send_callback_failure(
397 self, callback_id: str, error: ErrorObject | None = None
398 ) -> None:
399 try:
400 await self.lambda_client.send_durable_execution_callback_failure(
401 CallbackId=callback_id,
402 Error=cast(Any, error.to_dict() if error else None),
403 )
404 except Exception as e:
405 msg = f"Failed to send callback failure for {self.function_name}, callback_id {callback_id}: {e}"
406 raise DurableFunctionsTestError(msg) from e
408 async def send_callback_heartbeat(self, callback_id: str) -> None:
409 try:
410 await self.lambda_client.send_durable_execution_callback_heartbeat(
411 CallbackId=callback_id,
412 )
413 except Exception as e:
414 msg = f"Failed to send callback heartbeat for {self.function_name}, callback_id {callback_id}: {e}"
415 raise DurableFunctionsTestError(msg) from e
417 async def _wait_for_completion(
418 self, execution_arn: str, timeout: int
419 ) -> GetDurableExecutionResponse:
420 """Poll execution status until completion or timeout.
422 Args:
423 execution_arn: ARN of the durable execution
424 timeout: Maximum seconds to wait
426 Returns:
427 GetDurableExecutionResponse with typed execution details
429 Raises:
430 TimeoutError: If execution doesn't complete within timeout
431 DurableFunctionsTestError: If status check fails
432 """
433 start_time = time.time()
434 last_status = None
436 while time.time() - start_time < timeout:
437 try:
438 execution_dict = await self.lambda_client.get_durable_execution(
439 DurableExecutionArn=execution_arn,
440 IncludeExecutionData=True,
441 )
442 execution = GetDurableExecutionResponse.from_dict(execution_dict)
443 except Exception as e:
444 if (
445 _aws_error_code(e, self.lambda_client)
446 == "ResourceNotFoundException"
447 ):
448 logger.info(
449 "Execution status not available yet for %s; retrying",
450 execution_arn,
451 )
452 else:
453 msg = f"Failed to get execution status: {e}"
454 raise DurableFunctionsTestError(msg) from e
455 else:
456 # Log status changes
457 if execution.status != last_status:
458 logger.info("Execution status: %s", execution.status)
459 last_status = execution.status
461 # Check if execution completed
462 if execution.status == "SUCCEEDED":
463 logger.info("Execution succeeded")
464 return execution
465 if execution.status == "FAILED":
466 logger.warning("Execution failed")
467 return execution
468 if execution.status in ["TIMED_OUT", "ABORTED"]:
469 logger.warning("Execution terminated: %s", execution.status)
470 return execution
472 await asyncio.sleep(self.poll_interval)
474 # Timeout reached
475 elapsed = time.time() - start_time
476 msg = (
477 f"Execution did not complete within {timeout}s "
478 f"(elapsed: {elapsed:.1f}s, last status: {last_status})"
479 )
480 raise TimeoutError(msg)
482 async def wait_for_result(
483 self, execution_arn: str, timeout: int = 60
484 ) -> DurableFunctionTestResult:
485 execution_result = self._wait_for_completion(execution_arn, timeout)
486 execution_response = (
487 await execution_result
488 if inspect.isawaitable(execution_result)
489 else execution_result
490 )
492 try:
493 history_result = self._fetch_execution_history(execution_arn)
494 history_response = (
495 await history_result
496 if inspect.isawaitable(history_result)
497 else history_result
498 )
499 except Exception as e:
500 msg = f"Failed to fetch execution history: {e}"
501 raise DurableFunctionsTestError(msg) from e
503 # Build test result from execution history
504 return DurableFunctionTestResult.from_execution_history(
505 execution_response, history_response
506 )
508 async def wait_for_callback(
509 self, execution_arn: str, name: str | None = None, timeout: int = 60
510 ) -> str:
511 """
512 Wait for and retrieve a callback ID from a durable execution.
514 Polls the execution history at regular intervals until a callback ID is found
515 or the timeout is reached.
517 Args:
518 execution_arn: Execution Arn
519 name: Specific callback name, default to None
520 timeout: Maximum time in seconds to wait for callback. Defaults to 60.
522 Returns:
523 str: The callback ID/token retrieved from the execution history
525 Raises:
526 TimeoutError: If callback is not found within the specified timeout period
527 DurableFunctionsTestError: If there's an error fetching execution history
528 (excluding retryable errors)
529 """
530 start_time = time.time()
532 while time.time() - start_time < timeout:
533 try:
534 history_response = await self._fetch_execution_history(execution_arn)
535 callback_id = _get_callback_id_from_events(
536 events=history_response.events, name=name
537 )
538 if callback_id:
539 return callback_id
540 except DurableFunctionsTestError:
541 raise
542 except Exception as e:
543 # Retry while an asynchronously invoked execution is not yet visible.
544 if (
545 _aws_error_code(e, self.lambda_client)
546 != "ResourceNotFoundException"
547 ):
548 msg = f"Failed to fetch execution history: {e}"
549 raise DurableFunctionsTestError(msg) from e
551 await asyncio.sleep(self.poll_interval)
553 # Timeout reached
554 elapsed = time.time() - start_time
555 msg = f"Callback was not available within {timeout}s (elapsed: {elapsed:.1f}s)."
556 raise TimeoutError(msg)
558 async def _fetch_execution_history(
559 self, execution_arn: str
560 ) -> GetDurableExecutionHistoryResponse:
561 """Retrieve the complete execution history from Lambda service.
563 Args:
564 execution_arn: ARN of the durable execution
566 Returns:
567 GetDurableExecutionHistoryResponse with typed Event objects
569 Raises:
570 Exception: If the Lambda API client encounters an error
571 """
572 events = []
573 next_marker: str | None = None
574 seen_markers: set[str] = set()
575 page_count = 0
577 while True:
578 request: dict[str, Any] = {
579 "DurableExecutionArn": execution_arn,
580 "IncludeExecutionData": True,
581 }
582 if next_marker:
583 request["Marker"] = next_marker
585 history_dict = await self.lambda_client.get_durable_execution_history(
586 **request
587 )
588 history_response = GetDurableExecutionHistoryResponse.from_dict(
589 history_dict
590 )
591 page_count += 1
592 events.extend(history_response.events)
594 next_marker = history_response.next_marker
595 if not next_marker:
596 break
597 if next_marker in seen_markers:
598 msg = (
599 "Execution history pagination returned a repeated marker: "
600 f"{next_marker}"
601 )
602 raise DurableFunctionsTestError(msg)
603 seen_markers.add(next_marker)
605 logger.info(
606 "Retrieved %d events from history across %d page(s)",
607 len(events),
608 page_count,
609 )
611 return GetDurableExecutionHistoryResponse(events=events)
614class LambdaInvoker:
615 def __init__(self, lambda_client: Any) -> None:
616 self.lambda_client = adapt_lambda_client(lambda_client)
617 # Maps execution_arn -> endpoint for that execution
618 # Maps endpoint -> client to reuse clients across executions
619 self._execution_endpoints: dict[str, str] = {}
620 self._endpoint_clients: dict[str, Any] = {}
621 self._current_endpoint: str = "" # Track current endpoint for new executions
623 def _get_client_for_execution(
624 self,
625 durable_execution_arn: str,
626 lambda_endpoint: str | None = None,
627 region_name: str | None = None,
628 ) -> Any:
629 """Get the appropriate client for this execution."""
630 # Use provided endpoint or fall back to cached endpoint for this execution
631 if lambda_endpoint:
632 if lambda_endpoint not in self._endpoint_clients:
633 self._endpoint_clients[lambda_endpoint] = adapt_lambda_client(
634 create_lambda_client(lambda_endpoint, region_name or "us-east-1")
635 )
636 return self._endpoint_clients[lambda_endpoint]
638 # Fallback to cached endpoint
639 if durable_execution_arn not in self._execution_endpoints: 639 ↛ 642line 639 didn't jump to line 642 because the condition on line 639 was always true
640 self._execution_endpoints[durable_execution_arn] = self._current_endpoint
642 endpoint = self._execution_endpoints[durable_execution_arn]
644 # If no endpoint configured, fall back to default client
645 if not endpoint:
646 return self.lambda_client
648 return self._endpoint_clients[endpoint]
650 async def invoke(
651 self,
652 function_name: str,
653 input: DurableExecutionInvocationInput,
654 endpoint_url: str | None = None,
655 ) -> InvokeResponse:
656 """Invoke AWS Lambda function and return durable execution result.
658 Args:
659 function_name: Name of the Lambda function to invoke
660 input: Durable execution invocation input
661 endpoint_url: Lambda endpoint url
663 Returns:
664 InvokeResponse: Response containing invocation output and request ID
666 Raises:
667 ResourceNotFoundException: If function does not exist
668 InvalidParameterValueException: If parameters are invalid
669 DurableFunctionsTestError: For other invocation failures
670 """
672 # Parameter validation
673 if not function_name or not function_name.strip():
674 msg = "Function name is required"
675 raise InvalidParameterValueException(msg)
677 # Get the client for this execution
678 client = self._get_client_for_execution(
679 input.durable_execution_arn, endpoint_url
680 )
682 try:
683 # Invoke AWS Lambda function using standard invoke method
684 response = await client.invoke(
685 FunctionName=function_name,
686 InvocationType="RequestResponse", # Synchronous invocation
687 Payload=json.dumps(input.to_dict()),
688 )
690 # Check HTTP status code
691 status_code = response.get("StatusCode")
692 if status_code not in (200, 202, 204):
693 msg = f"Lambda invocation failed with status code: {status_code}"
694 raise DurableFunctionsTestError(msg)
696 # Check for function errors
697 if "FunctionError" in response:
698 error_payload = await _read_payload(response["Payload"])
699 msg = f"Lambda invocation failed with status {status_code}: {error_payload}"
700 raise DurableFunctionsTestError(msg)
702 # Parse response payload
703 response_payload = await _read_payload(response["Payload"])
704 response_dict = json.loads(response_payload)
706 # Extract request ID from response headers (x-amzn-RequestId or x-amzn-request-id)
707 headers = response.get("ResponseMetadata", {}).get("HTTPHeaders", {})
708 request_id = (
709 headers.get("x-amzn-RequestId")
710 or headers.get("x-amzn-request-id")
711 or headers.get("x-amzn-requestid")
712 or f"local-{uuid4()}"
713 )
715 # Convert to DurableExecutionInvocationOutput
716 output = DurableExecutionInvocationOutput.from_dict(response_dict)
717 return InvokeResponse(invocation_output=output, request_id=request_id)
719 except Exception as e:
720 error_code = _aws_error_code(e, client)
721 if error_code == "ResourceNotFoundException":
722 msg = f"Function not found: {function_name}"
723 raise ResourceNotFoundException(msg) from e
724 if error_code == "InvalidParameterValueException":
725 msg = f"Invalid parameter: {e}"
726 raise InvalidParameterValueException(msg) from e
727 if error_code in {
728 "TooManyRequestsException",
729 "ServiceException",
730 "ResourceConflictException",
731 "InvalidRequestContentException",
732 "RequestTooLargeException",
733 "UnsupportedMediaTypeException",
734 "InvalidRuntimeException",
735 "InvalidZipFileException",
736 "ResourceNotReadyException",
737 "SnapStartTimeoutException",
738 "SnapStartNotReadyException",
739 "SnapStartException",
740 "RecursiveInvocationException",
741 }:
742 msg = f"Lambda invocation failed: {e}"
743 raise DurableFunctionsTestError(msg) from e
744 if error_code in {
745 "InvalidSecurityGroupIDException",
746 "EC2ThrottledException",
747 "EFSMountConnectivityException",
748 "SubnetIPAddressLimitReachedException",
749 "EC2UnexpectedException",
750 "InvalidSubnetIDException",
751 "EC2AccessDeniedException",
752 "EFSIOException",
753 "ENILimitReachedException",
754 "EFSMountTimeoutException",
755 "EFSMountFailureException",
756 }:
757 msg = f"Lambda infrastructure error: {e}"
758 raise DurableFunctionsTestError(msg) from e
759 if error_code in {
760 "KMSAccessDeniedException",
761 "KMSDisabledException",
762 "KMSNotFoundException",
763 "KMSInvalidStateException",
764 }:
765 msg = f"Lambda KMS error: {e}"
766 raise DurableFunctionsTestError(msg) from e
767 if error_code == "DurableExecutionAlreadyStartedException":
768 msg = f"Durable execution already started: {e}"
769 raise DurableFunctionsTestError(msg) from e
770 msg = f"Unexpected error during Lambda invocation: {e}"
771 raise DurableFunctionsTestError(msg) from e
774def create_sync_lambda_client(endpoint_url: str | None, region_name: str) -> Any:
775 """Create a sync Lambda client adapted for cloud runner calls."""
776 return ThreadedSyncCloudLambdaClient(
777 create_default_sync_client(
778 session=get_session(),
779 endpoint_url=endpoint_url,
780 region_name=region_name,
781 config=_LAMBDA_CLIENT_CONFIG,
782 )
783 )
786def create_async_lambda_client(endpoint_url: str | None, region_name: str) -> Any:
787 """Create an async Lambda client adapted for cloud runner calls."""
788 return AsyncCloudLambdaClient(
789 create_default_async_client(
790 session=get_session(),
791 endpoint_url=endpoint_url,
792 region_name=region_name,
793 config=_LAMBDA_CLIENT_CONFIG,
794 )
795 )
798def create_lambda_client(endpoint_url: str | None, region_name: str) -> Any:
799 """Create a Lambda client, preferring HTTPX when installed."""
800 if httpx_is_installed():
801 return create_async_lambda_client(endpoint_url, region_name)
802 return create_sync_lambda_client(endpoint_url, region_name)
805_LAMBDA_READ_TIMEOUT_SECONDS = 960
806_LAMBDA_CLIENT_CONFIG = Config(
807 parameter_validation=False,
808 read_timeout=_LAMBDA_READ_TIMEOUT_SECONDS,
809 retries={"max_attempts": 0},
810)