Coverage for async-durable-execution/src/async_durable_execution/runner/cloud/__init__.py: 89%
346 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 16:54 +0000
1from __future__ import annotations
3import asyncio
4import importlib
5import inspect
6import json
7import logging
8import time
9from typing import Any, cast
10from uuid import uuid4
12from botocore.config import Config
13from botocore.exceptions import ClientError
14from botocore.session import get_session
16from ... import client as durable_client
17from ...execution import (
18 DurableExecutionInvocationInput,
19)
20from ...models import DurableExecutionInvocationOutput, ErrorObject
21from ..exceptions import (
22 DurableFunctionsTestError,
23 InvalidParameterValueException,
24 ResourceNotFoundException,
25)
26from ..model import (
27 DurableFunctionTestResult,
28 GetDurableExecutionResponse,
29 GetDurableExecutionHistoryResponse,
30 InvokeResponse,
31 _get_callback_id_from_events,
32)
35logger = logging.getLogger(__name__)
38class ThreadedSyncCloudLambdaClient:
39 """Adapt a sync Lambda client to the async cloud runner interface."""
41 def __init__(self, client: Any) -> None:
42 self.client = client
44 @property
45 def exceptions(self) -> Any:
46 return self.client.exceptions
48 async def invoke(self, **kwargs: Any) -> dict[str, Any]:
49 return cast(
50 dict[str, Any], await asyncio.to_thread(self.client.invoke, **kwargs)
51 )
53 async def get_durable_execution(self, **kwargs: Any) -> dict[str, Any]:
54 return cast(
55 dict[str, Any],
56 await asyncio.to_thread(self.client.get_durable_execution, **kwargs),
57 )
59 async def get_durable_execution_history(self, **kwargs: Any) -> dict[str, Any]:
60 return cast(
61 dict[str, Any],
62 await asyncio.to_thread(
63 self.client.get_durable_execution_history, **kwargs
64 ),
65 )
67 async def send_durable_execution_callback_success(
68 self, **kwargs: Any
69 ) -> dict[str, Any]:
70 return cast(
71 dict[str, Any],
72 await asyncio.to_thread(
73 self.client.send_durable_execution_callback_success, **kwargs
74 ),
75 )
77 async def send_durable_execution_callback_failure(
78 self, **kwargs: Any
79 ) -> dict[str, Any]:
80 return cast(
81 dict[str, Any],
82 await asyncio.to_thread(
83 self.client.send_durable_execution_callback_failure, **kwargs
84 ),
85 )
87 async def send_durable_execution_callback_heartbeat(
88 self, **kwargs: Any
89 ) -> dict[str, Any]:
90 return cast(
91 dict[str, Any],
92 await asyncio.to_thread(
93 self.client.send_durable_execution_callback_heartbeat, **kwargs
94 ),
95 )
97 def close(self) -> None:
98 close = getattr(self.client, "close", None)
99 if callable(close):
100 close()
103class AsyncCloudLambdaClient:
104 """Adapt an async aioboto Lambda client to the cloud runner interface."""
106 def __init__(self, client: Any) -> None:
107 self._client_context = client if hasattr(client, "__aenter__") else None
108 self._client = None if self._client_context is not None else client
109 self._entered_client: Any | None = None
111 @property
112 def exceptions(self) -> Any:
113 client = self._entered_client or self._client
114 if client is None:
115 msg = "Async Lambda client has not been initialized"
116 raise AttributeError(msg)
117 return client.exceptions
119 async def _get_client(self) -> Any:
120 if self._client is not None:
121 return self._client
122 if self._entered_client is None:
123 assert self._client_context is not None
124 self._entered_client = await self._client_context.__aenter__()
125 return self._entered_client
127 async def _call(self, method_name: str, **kwargs: Any) -> dict[str, Any]:
128 client = await self._get_client()
129 result = getattr(client, method_name)(**kwargs)
130 if inspect.isawaitable(result):
131 result = await result
132 return cast(dict[str, Any], result)
134 async def invoke(self, **kwargs: Any) -> dict[str, Any]:
135 return await self._call("invoke", **kwargs)
137 async def get_durable_execution(self, **kwargs: Any) -> dict[str, Any]:
138 return await self._call("get_durable_execution", **kwargs)
140 async def get_durable_execution_history(self, **kwargs: Any) -> dict[str, Any]:
141 return await self._call("get_durable_execution_history", **kwargs)
143 async def send_durable_execution_callback_success(
144 self, **kwargs: Any
145 ) -> dict[str, Any]:
146 return await self._call("send_durable_execution_callback_success", **kwargs)
148 async def send_durable_execution_callback_failure(
149 self, **kwargs: Any
150 ) -> dict[str, Any]:
151 return await self._call("send_durable_execution_callback_failure", **kwargs)
153 async def send_durable_execution_callback_heartbeat(
154 self, **kwargs: Any
155 ) -> dict[str, Any]:
156 return await self._call("send_durable_execution_callback_heartbeat", **kwargs)
158 async def aclose(self) -> None:
159 if self._entered_client is not None:
160 assert self._client_context is not None
161 await self._client_context.__aexit__(None, None, None)
162 self._entered_client = None
163 return
165 close = getattr(self._client, "aclose", None)
166 if callable(close):
167 await close()
170async def _read_payload(payload: Any) -> str:
171 read = getattr(payload, "read", None)
172 if callable(read):
173 if inspect.iscoroutinefunction(read):
174 data = await read()
175 else:
176 data = await asyncio.to_thread(read)
177 if inspect.isawaitable(data):
178 data = await data
179 else:
180 data = payload
182 if isinstance(data, bytes):
183 return data.decode("utf-8")
184 return str(data)
187def _cloud_lambda_client_is_async(client: Any) -> bool:
188 return inspect.iscoroutinefunction(getattr(client, "invoke", None))
191def adapt_lambda_client(client: Any) -> Any:
192 """Adapt a raw Lambda client to the async cloud runner interface."""
193 if isinstance(client, ThreadedSyncCloudLambdaClient | AsyncCloudLambdaClient):
194 return client
195 if _cloud_lambda_client_is_async(client):
196 return AsyncCloudLambdaClient(client)
197 return ThreadedSyncCloudLambdaClient(client)
200def create_cloud_runner(
201 *,
202 function_name: str,
203 region: str = "us-west-2",
204 lambda_endpoint: str | None = None,
205 poll_interval: float = 1.0,
206 input: Any = None, # noqa: A002
207 timeout: int = 60,
208) -> DurableFunctionCloudTestRunner:
209 """Create a configured cloud durable function runner."""
210 return DurableFunctionCloudTestRunner(
211 function_name=function_name,
212 region=region,
213 lambda_endpoint=lambda_endpoint,
214 poll_interval=poll_interval,
215 input=input,
216 timeout=timeout,
217 )
220class DurableFunctionCloudTestRunner:
221 """Test runner that executes durable functions against actual AWS Lambda backend.
223 This runner invokes deployed Lambda functions and polls for execution completion,
224 providing the same interface as DurableFunctionLocalTestRunner for seamless test
225 compatibility between local and cloud modes.
226 """
228 def __init__(
229 self,
230 function_name: str,
231 region: str = "us-west-2",
232 lambda_endpoint: str | None = None,
233 poll_interval: float = 1.0,
234 input: Any = None, # noqa: A002
235 timeout: int = 60,
236 ):
237 """Initialize cloud test runner."""
238 self.mode = "cloud"
239 self.function_name = function_name
240 self.region = region
241 self.lambda_endpoint = lambda_endpoint
242 self.poll_interval = poll_interval
243 self._default_input = input
244 self._default_timeout = timeout
246 self.lambda_client: Any = create_lambda_client(lambda_endpoint, region)
248 async def __aenter__(self) -> DurableFunctionCloudTestRunner:
249 return self
251 async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
252 await self.aclose()
254 def close(self) -> None:
255 """Close the underlying sync client when supported."""
256 close = getattr(self.lambda_client, "close", None)
257 if callable(close):
258 close()
260 async def aclose(self) -> None:
261 """Close the underlying client when supported."""
262 aclose = getattr(self.lambda_client, "aclose", None)
263 if callable(aclose):
264 await aclose()
265 return
266 self.close()
268 async def run(
269 self,
270 ) -> DurableFunctionTestResult:
271 """Execute function on AWS Lambda and wait for completion."""
272 execution_arn = await self._invoke_for_execution(
273 invocation_type="RequestResponse",
274 expected_status_code=200,
275 )
276 return await self.wait_for_result(
277 execution_arn=execution_arn, timeout=self._default_timeout
278 )
280 async def run_async(
281 self,
282 ) -> str:
283 """Execute function on AWS Lambda asynchronously"""
284 return await self._invoke_for_execution(
285 invocation_type="Event",
286 expected_status_code=202,
287 )
289 async def _invoke_for_execution(
290 self,
291 *,
292 invocation_type: str,
293 expected_status_code: int,
294 ) -> str:
295 logger.info(
296 "Invoking Lambda function: %s (timeout: %ds)",
297 self.function_name,
298 self._default_timeout,
299 )
300 payload = json.dumps(self._default_input)
301 try:
302 response = cast(
303 dict[str, Any],
304 await self.lambda_client.invoke(
305 FunctionName=self.function_name,
306 InvocationType=invocation_type,
307 Payload=payload,
308 ),
309 )
310 except Exception as e:
311 msg = f"Failed to invoke Lambda function {self.function_name}: {e}"
312 raise DurableFunctionsTestError(msg) from e
314 status_code = response.get("StatusCode")
315 if status_code != expected_status_code:
316 error_payload = await _read_payload(response["Payload"])
317 msg = f"Lambda invocation failed with status {status_code}: {error_payload}"
318 raise DurableFunctionsTestError(msg)
320 if "FunctionError" in response:
321 error_payload = await _read_payload(response["Payload"])
322 logger.warning("Lambda function failed: %s", error_payload)
324 execution_arn = cast(str | None, response.get("DurableExecutionArn"))
325 if execution_arn is None:
326 msg = (
327 f"No DurableExecutionArn in response for function {self.function_name}"
328 )
329 raise DurableFunctionsTestError(msg)
330 return execution_arn
332 async def send_callback_success(
333 self, callback_id: str, result: bytes | None = None
334 ) -> None:
335 try:
336 await self.lambda_client.send_durable_execution_callback_success(
337 CallbackId=callback_id,
338 Result=cast(Any, result),
339 )
340 except Exception as e:
341 msg = f"Failed to send callback success for {self.function_name}, callback_id {callback_id}: {e}"
342 raise DurableFunctionsTestError(msg) from e
344 async def send_callback_failure(
345 self, callback_id: str, error: ErrorObject | None = None
346 ) -> None:
347 try:
348 await self.lambda_client.send_durable_execution_callback_failure(
349 CallbackId=callback_id,
350 Error=cast(Any, error.to_dict() if error else None),
351 )
352 except Exception as e:
353 msg = f"Failed to send callback failure for {self.function_name}, callback_id {callback_id}: {e}"
354 raise DurableFunctionsTestError(msg) from e
356 async def send_callback_heartbeat(self, callback_id: str) -> None:
357 try:
358 await self.lambda_client.send_durable_execution_callback_heartbeat(
359 CallbackId=callback_id,
360 )
361 except Exception as e:
362 msg = f"Failed to send callback heartbeat for {self.function_name}, callback_id {callback_id}: {e}"
363 raise DurableFunctionsTestError(msg) from e
365 async def _wait_for_completion(
366 self, execution_arn: str, timeout: int
367 ) -> GetDurableExecutionResponse:
368 """Poll execution status until completion or timeout.
370 Args:
371 execution_arn: ARN of the durable execution
372 timeout: Maximum seconds to wait
374 Returns:
375 GetDurableExecutionResponse with typed execution details
377 Raises:
378 TimeoutError: If execution doesn't complete within timeout
379 DurableFunctionsTestError: If status check fails
380 """
381 start_time = time.time()
382 last_status = None
384 while time.time() - start_time < timeout:
385 try:
386 execution_dict = await self.lambda_client.get_durable_execution(
387 DurableExecutionArn=execution_arn
388 )
389 execution = GetDurableExecutionResponse.from_dict(execution_dict)
390 except ClientError as e:
391 error_code = e.response.get("Error", {}).get("Code")
392 if error_code == "ResourceNotFoundException":
393 logger.info(
394 "Execution status not available yet for %s; retrying",
395 execution_arn,
396 )
397 else:
398 msg = f"Failed to get execution status: {e}"
399 raise DurableFunctionsTestError(msg) from e
400 except Exception as e:
401 msg = f"Failed to get execution status: {e}"
402 raise DurableFunctionsTestError(msg) from e
403 else:
404 # Log status changes
405 if execution.status != last_status:
406 logger.info("Execution status: %s", execution.status)
407 last_status = execution.status
409 # Check if execution completed
410 if execution.status == "SUCCEEDED":
411 logger.info("Execution succeeded")
412 return execution
413 if execution.status == "FAILED":
414 logger.warning("Execution failed")
415 return execution
416 if execution.status in ["TIMED_OUT", "ABORTED"]:
417 logger.warning("Execution terminated: %s", execution.status)
418 return execution
420 await asyncio.sleep(self.poll_interval)
422 # Timeout reached
423 elapsed = time.time() - start_time
424 msg = (
425 f"Execution did not complete within {timeout}s "
426 f"(elapsed: {elapsed:.1f}s, last status: {last_status})"
427 )
428 raise TimeoutError(msg)
430 async def wait_for_result(
431 self, execution_arn: str, timeout: int = 60
432 ) -> DurableFunctionTestResult:
433 execution_result = self._wait_for_completion(execution_arn, timeout)
434 execution_response = (
435 await execution_result
436 if inspect.isawaitable(execution_result)
437 else execution_result
438 )
440 try:
441 history_result = self._fetch_execution_history(execution_arn)
442 history_response = (
443 await history_result
444 if inspect.isawaitable(history_result)
445 else history_result
446 )
447 except Exception as e:
448 msg = f"Failed to fetch execution history: {e}"
449 raise DurableFunctionsTestError(msg) from e
451 # Build test result from execution history
452 return DurableFunctionTestResult.from_execution_history(
453 execution_response, history_response
454 )
456 async def wait_for_callback(
457 self, execution_arn: str, name: str | None = None, timeout: int = 60
458 ) -> str:
459 """
460 Wait for and retrieve a callback ID from a durable execution.
462 Polls the execution history at regular intervals until a callback ID is found
463 or the timeout is reached.
465 Args:
466 execution_arn: Execution Arn
467 name: Specific callback name, default to None
468 timeout: Maximum time in seconds to wait for callback. Defaults to 60.
470 Returns:
471 str: The callback ID/token retrieved from the execution history
473 Raises:
474 TimeoutError: If callback is not found within the specified timeout period
475 DurableFunctionsTestError: If there's an error fetching execution history
476 (excluding retryable errors)
477 """
478 start_time = time.time()
480 while time.time() - start_time < timeout:
481 try:
482 history_response = await self._fetch_execution_history(execution_arn)
483 callback_id = _get_callback_id_from_events(
484 events=history_response.events, name=name
485 )
486 if callback_id:
487 return callback_id
488 except ClientError as e:
489 error_code = e.response["Error"]["Code"]
490 # retryable error, the execution may not start yet in async invoke situation
491 if error_code in ["ResourceNotFoundException"]:
492 pass
493 else:
494 msg = f"Failed to fetch execution history: {e}"
495 raise DurableFunctionsTestError(msg) from e
496 except DurableFunctionsTestError:
497 raise
498 except Exception as e:
499 msg = f"Failed to fetch execution history: {e}"
500 raise DurableFunctionsTestError(msg) from e
502 await asyncio.sleep(self.poll_interval)
504 # Timeout reached
505 elapsed = time.time() - start_time
506 msg = f"Callback was not available within {timeout}s (elapsed: {elapsed:.1f}s)."
507 raise TimeoutError(msg)
509 async def _fetch_execution_history(
510 self, execution_arn: str
511 ) -> GetDurableExecutionHistoryResponse:
512 """Retrieve the complete execution history from Lambda service.
514 Args:
515 execution_arn: ARN of the durable execution
517 Returns:
518 GetDurableExecutionHistoryResponse with typed Event objects
520 Raises:
521 ClientError: If lambda client encounter error
522 """
523 events = []
524 next_marker: str | None = None
525 seen_markers: set[str] = set()
526 page_count = 0
528 while True:
529 request: dict[str, Any] = {
530 "DurableExecutionArn": execution_arn,
531 "IncludeExecutionData": True,
532 }
533 if next_marker:
534 request["Marker"] = next_marker
536 history_dict = await self.lambda_client.get_durable_execution_history(
537 **request
538 )
539 history_response = GetDurableExecutionHistoryResponse.from_dict(
540 history_dict
541 )
542 page_count += 1
543 events.extend(history_response.events)
545 next_marker = history_response.next_marker
546 if not next_marker:
547 break
548 if next_marker in seen_markers:
549 msg = (
550 "Execution history pagination returned a repeated marker: "
551 f"{next_marker}"
552 )
553 raise DurableFunctionsTestError(msg)
554 seen_markers.add(next_marker)
556 logger.info(
557 "Retrieved %d events from history across %d page(s)",
558 len(events),
559 page_count,
560 )
562 return GetDurableExecutionHistoryResponse(events=events)
565class LambdaInvoker:
566 def __init__(self, lambda_client: Any) -> None:
567 self.lambda_client = adapt_lambda_client(lambda_client)
568 # Maps execution_arn -> endpoint for that execution
569 # Maps endpoint -> client to reuse clients across executions
570 self._execution_endpoints: dict[str, str] = {}
571 self._endpoint_clients: dict[str, Any] = {}
572 self._current_endpoint: str = "" # Track current endpoint for new executions
574 @staticmethod
575 def create(endpoint_url: str, region_name: str) -> LambdaInvoker:
576 """Create with the boto lambda client."""
577 invoker = LambdaInvoker(create_lambda_client(endpoint_url, region_name))
578 invoker._current_endpoint = endpoint_url
579 invoker._endpoint_clients[endpoint_url] = invoker.lambda_client
580 return invoker
582 def update_endpoint(self, endpoint_url: str, region_name: str) -> None:
583 """Update the Lambda client endpoint."""
584 # Cache client by endpoint to reuse across executions
585 if endpoint_url not in self._endpoint_clients:
586 self._endpoint_clients[endpoint_url] = adapt_lambda_client(
587 create_lambda_client(endpoint_url, region_name)
588 )
589 self.lambda_client = self._endpoint_clients[endpoint_url]
590 self._current_endpoint = endpoint_url
592 def _get_client_for_execution(
593 self,
594 durable_execution_arn: str,
595 lambda_endpoint: str | None = None,
596 region_name: str | None = None,
597 ) -> Any:
598 """Get the appropriate client for this execution."""
599 # Use provided endpoint or fall back to cached endpoint for this execution
600 if lambda_endpoint:
601 if lambda_endpoint not in self._endpoint_clients:
602 self._endpoint_clients[lambda_endpoint] = adapt_lambda_client(
603 create_lambda_client(lambda_endpoint, region_name or "us-east-1")
604 )
605 return self._endpoint_clients[lambda_endpoint]
607 # Fallback to cached endpoint
608 if durable_execution_arn not in self._execution_endpoints:
609 self._execution_endpoints[durable_execution_arn] = self._current_endpoint
611 endpoint = self._execution_endpoints[durable_execution_arn]
613 # If no endpoint configured, fall back to default client
614 if not endpoint:
615 return self.lambda_client
617 return self._endpoint_clients[endpoint]
619 async def invoke(
620 self,
621 function_name: str,
622 input: DurableExecutionInvocationInput,
623 endpoint_url: str | None = None,
624 ) -> InvokeResponse:
625 """Invoke AWS Lambda function and return durable execution result.
627 Args:
628 function_name: Name of the Lambda function to invoke
629 input: Durable execution invocation input
630 endpoint_url: Lambda endpoint url
632 Returns:
633 InvokeResponse: Response containing invocation output and request ID
635 Raises:
636 ResourceNotFoundException: If function does not exist
637 InvalidParameterValueException: If parameters are invalid
638 DurableFunctionsTestError: For other invocation failures
639 """
641 # Parameter validation
642 if not function_name or not function_name.strip():
643 msg = "Function name is required"
644 raise InvalidParameterValueException(msg)
646 # Get the client for this execution
647 client = self._get_client_for_execution(
648 input.durable_execution_arn, endpoint_url
649 )
651 try:
652 # Invoke AWS Lambda function using standard invoke method
653 response = await client.invoke(
654 FunctionName=function_name,
655 InvocationType="RequestResponse", # Synchronous invocation
656 Payload=json.dumps(input.to_json_dict()),
657 )
659 # Check HTTP status code
660 status_code = response.get("StatusCode")
661 if status_code not in (200, 202, 204):
662 msg = f"Lambda invocation failed with status code: {status_code}"
663 raise DurableFunctionsTestError(msg)
665 # Check for function errors
666 if "FunctionError" in response:
667 error_payload = await _read_payload(response["Payload"])
668 msg = f"Lambda invocation failed with status {status_code}: {error_payload}"
669 raise DurableFunctionsTestError(msg)
671 # Parse response payload
672 response_payload = await _read_payload(response["Payload"])
673 response_dict = json.loads(response_payload)
675 # Extract request ID from response headers (x-amzn-RequestId or x-amzn-request-id)
676 headers = response.get("ResponseMetadata", {}).get("HTTPHeaders", {})
677 request_id = (
678 headers.get("x-amzn-RequestId")
679 or headers.get("x-amzn-request-id")
680 or f"local-{uuid4()}"
681 )
683 # Convert to DurableExecutionInvocationOutput
684 output = DurableExecutionInvocationOutput.from_dict(response_dict)
685 return InvokeResponse(invocation_output=output, request_id=request_id)
687 except client.exceptions.ResourceNotFoundException as e:
688 msg = f"Function not found: {function_name}"
689 raise ResourceNotFoundException(msg) from e
690 except client.exceptions.InvalidParameterValueException as e:
691 msg = f"Invalid parameter: {e}"
692 raise InvalidParameterValueException(msg) from e
693 except (
694 client.exceptions.TooManyRequestsException,
695 client.exceptions.ServiceException,
696 client.exceptions.ResourceConflictException,
697 client.exceptions.InvalidRequestContentException,
698 client.exceptions.RequestTooLargeException,
699 client.exceptions.UnsupportedMediaTypeException,
700 client.exceptions.InvalidRuntimeException,
701 client.exceptions.InvalidZipFileException,
702 client.exceptions.ResourceNotReadyException,
703 client.exceptions.SnapStartTimeoutException,
704 client.exceptions.SnapStartNotReadyException,
705 client.exceptions.SnapStartException,
706 client.exceptions.RecursiveInvocationException,
707 ) as e:
708 msg = f"Lambda invocation failed: {e}"
709 raise DurableFunctionsTestError(msg) from e
710 except (
711 client.exceptions.InvalidSecurityGroupIDException,
712 client.exceptions.EC2ThrottledException,
713 client.exceptions.EFSMountConnectivityException,
714 client.exceptions.SubnetIPAddressLimitReachedException,
715 client.exceptions.EC2UnexpectedException,
716 client.exceptions.InvalidSubnetIDException,
717 client.exceptions.EC2AccessDeniedException,
718 client.exceptions.EFSIOException,
719 client.exceptions.ENILimitReachedException,
720 client.exceptions.EFSMountTimeoutException,
721 client.exceptions.EFSMountFailureException,
722 ) as e:
723 msg = f"Lambda infrastructure error: {e}"
724 raise DurableFunctionsTestError(msg) from e
725 except (
726 client.exceptions.KMSAccessDeniedException,
727 client.exceptions.KMSDisabledException,
728 client.exceptions.KMSNotFoundException,
729 client.exceptions.KMSInvalidStateException,
730 ) as e:
731 msg = f"Lambda KMS error: {e}"
732 raise DurableFunctionsTestError(msg) from e
733 except Exception as e:
734 # Handle any remaining exceptions, including custom ones like DurableExecutionAlreadyStartedException
735 if "DurableExecutionAlreadyStartedException" in str(type(e)):
736 msg = f"Durable execution already started: {e}"
737 raise DurableFunctionsTestError(msg) from e
738 msg = f"Unexpected error during Lambda invocation: {e}"
739 raise DurableFunctionsTestError(msg) from e
742def create_sync_lambda_client(endpoint_url: str | None, region_name: str) -> Any:
743 """Create a sync Lambda client adapted for cloud runner calls."""
744 session = get_session()
745 return ThreadedSyncCloudLambdaClient(
746 session.create_client(
747 "lambda",
748 endpoint_url=endpoint_url,
749 region_name=region_name,
750 config=_LAMBDA_CLIENT_CONFIG,
751 )
752 )
755def create_async_lambda_client(endpoint_url: str | None, region_name: str) -> Any:
756 """Create an async Lambda client adapted for cloud runner calls."""
757 aiobotocore_session = importlib.import_module("aiobotocore.session")
758 session = aiobotocore_session.get_session()
759 return AsyncCloudLambdaClient(
760 session.create_client(
761 "lambda",
762 endpoint_url=endpoint_url,
763 region_name=region_name,
764 config=_LAMBDA_CLIENT_CONFIG,
765 )
766 )
769def create_lambda_client(endpoint_url: str | None, region_name: str) -> Any:
770 """Create a Lambda client, preferring aioboto when installed."""
771 if durable_client.aioboto_is_installed():
772 return create_async_lambda_client(endpoint_url, region_name)
773 return create_sync_lambda_client(endpoint_url, region_name)
776_LAMBDA_READ_TIMEOUT_SECONDS = 960
777_LAMBDA_CLIENT_CONFIG = Config(
778 parameter_validation=False,
779 read_timeout=_LAMBDA_READ_TIMEOUT_SECONDS,
780 retries={"max_attempts": 0},
781)