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

1"""User-facing durable step operation.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from collections.abc import Awaitable, Callable 

7from typing import TypeVar 

8 

9from .._core import Duration, OperationSubType, SerDes 

10from .._primitive.step import ( 

11 StepContext, 

12 StepInterruptedError, 

13 StepSemantics, 

14 get_step_context, 

15) 

16from ..extension import get_extension_context 

17 

18T = TypeVar("T") 

19 

20 

21def step( 

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

23 *, 

24 name: str | None = None, 

25 retry_strategy: Callable[[Exception, int], Duration | None] | None = None, 

26 step_semantics: StepSemantics = StepSemantics.AT_LEAST_ONCE_PER_RETRY, 

27 serdes: SerDes | None = None, 

28) -> asyncio.Task[T]: 

29 """Run user code as a checkpointed durable step. 

30 

31 Durable steps are the main way to isolate non-deterministic work such as API 

32 calls, clock reads, UUID generation, and database access from replayed code. 

33 """ 

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

35 return ( 

36 get_extension_context() 

37 ._reserve_sdk_operation( # noqa: SLF001 

38 step_name, 

39 ) 

40 ._run_step( # noqa: SLF001 

41 func, 

42 sub_type=OperationSubType.STEP, 

43 retry_strategy=retry_strategy, 

44 step_semantics=step_semantics, 

45 serdes=serdes, 

46 ) 

47 ) 

48 

49 

50__all__ = [ 

51 "StepContext", 

52 "StepInterruptedError", 

53 "StepSemantics", 

54 "get_step_context", 

55 "step", 

56]