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

1"""User-facing durable child-context operation.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from collections.abc import Awaitable, Callable 

7from typing import TypeVar 

8 

9from .._core import OperationSubType, SerDes 

10from .._primitive.child import SummaryGenerator 

11from ..extension import get_extension_context 

12 

13T = TypeVar("T") 

14 

15 

16def run_in_child_context( 

17 func: Callable[[], Awaitable[T]], 

18 *, 

19 name: str | None = None, 

20 serdes: SerDes | None = None, 

21 summary_generator: SummaryGenerator | None = None, 

22 is_virtual: bool = False, 

23) -> asyncio.Task[T]: 

24 """Execute a durable sub-workflow inside its own child context. 

25 

26 Args: 

27 func: The child context function to execute. 

28 name: Optional durable operation name. 

29 serdes: Optional serializer for the child context result. 

30 summary_generator: Optional summary generator for large child results. 

31 is_virtual: Whether this child context should skip lifecycle checkpoints. 

32 """ 

33 step_name = name if name is not None else getattr(func, "__name__", None) 

34 return ( 

35 get_extension_context() 

36 ._reserve_sdk_operation(step_name) # noqa: SLF001 

37 ._run_in_child_context( # noqa: SLF001 

38 func, 

39 sub_type=OperationSubType.RUN_IN_CHILD_CONTEXT, 

40 serdes=serdes, 

41 summary_generator=summary_generator, 

42 is_virtual=is_virtual, 

43 ) 

44 ) 

45 

46 

47__all__ = ["SummaryGenerator", "run_in_child_context"]