Coverage for async_durable_execution/_operation/invoke.py: 100%

1"""User-facing durable chained-invoke operation.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from typing import TypeVar 

7 

8from .._core import OperationSubType, SerDes 

9from ..extension import get_extension_context 

10 

11P = TypeVar("P") 

12R = TypeVar("R") 

13 

14 

15def invoke( 

16 function_name: str, 

17 payload: P, 

18 *, 

19 name: str | None = None, 

20 serdes_payload: SerDes[P] | None = None, 

21 serdes_result: SerDes[R] | None = None, 

22 tenant_id: str | None = None, 

23) -> asyncio.Task[R]: 

24 """Invoke another durable Lambda function and wait for its durable result. 

25 

26 Args: 

27 function_name: Qualified Lambda function name or ARN to invoke. 

28 payload: Payload to send to the invoked function. 

29 name: Optional durable operation name. 

30 serdes_payload: Optional serializer for the invocation payload. 

31 serdes_result: Optional deserializer for the invocation result. 

32 tenant_id: Optional tenant identifier for the chained invocation. 

33 """ 

34 return ( 

35 get_extension_context() 

36 ._reserve_sdk_operation( # noqa: SLF001 

37 name, 

38 ) 

39 ._run_invoke( # noqa: SLF001 

40 function_name, 

41 payload, 

42 sub_type=OperationSubType.CHAINED_INVOKE, 

43 serdes_payload=serdes_payload, 

44 serdes_result=serdes_result, 

45 tenant_id=tenant_id, 

46 ) 

47 ) 

48 

49 

50__all__ = ["invoke"]