Custom Operation SPI

Public module: async_durable_execution.extension.

Use these contracts in independently maintained packages that implement reusable durable operations. See Custom Durable Operations for the extension-author guide and replay compatibility rules.

Stable contracts for authoring third-party durable operations.

Attributes

ExtensionStepFunction module-attribute

ExtensionStepFunction: TypeAlias = Callable[
    [T | None], Awaitable[ExtensionStepResult[T]]
]

ExtensionStepRetryStrategy module-attribute

ExtensionStepRetryStrategy: TypeAlias = Callable[
    [Exception, T | None, int],
    ExtensionStepResult[T] | None,
]

Classes

ExtensionContext

ExtensionContext(context: DurableContext)

Stable extension-author view of the current durable execution scope.

Attributes

lambda_context property

lambda_context: LambdaContext | None

Return the active AWS Lambda context, when available.

recursive_level property

recursive_level: int

Return the durable self-invocation recursion level.

Methods:

get_current classmethod

get_current() -> ExtensionContext

Return the extension context for the active handler or child scope.

is_replaying

is_replaying() -> bool

Return whether the current scope is replaying checkpointed work.

reserve

reserve(
    name: str | None = None,
    *,
    local_operation_id: str | None = None,
) -> ExtensionOperation

Reserve a stable one-shot primitive identity.

Sequential reservations depend on deterministic reservation order. A caller-provided local id remains stable when reservation order changes, but it must be unique within the current durable context.

ExtensionOperation

ExtensionOperation()

Opaque one-shot reservation for one SDK-owned durable primitive.

Instances are created only by :meth:ExtensionContext.reserve.

Methods:

step

step(
    func: ExtensionStepFunction[T],
    *,
    sub_type: str | OperationSubType,
    initial_state: T | None = None,
    retry_strategy: ExtensionStepRetryStrategy[T]
    | None = None,
    step_semantics: StepSemantics = AT_LEAST_ONCE_PER_RETRY,
    serdes: SerDes[T] | None = None,
) -> Task[T]

Use this reservation for a stateful STEP primitive.

wait

wait(
    duration: Duration, *, sub_type: str | OperationSubType
) -> Task[None]

Use this reservation for a WAIT primitive.

invoke

invoke(
    function_name: str,
    payload: P,
    *,
    sub_type: str | OperationSubType,
    serdes_payload: SerDes[P] | None = None,
    serdes_result: SerDes[R] | None = None,
    tenant_id: str | None = None,
) -> Task[R]

Use this reservation for a CHAINED_INVOKE primitive.

create_callback

create_callback(
    *,
    sub_type: str | OperationSubType,
    timeout: Duration | None = None,
    heartbeat_timeout: Duration | None = None,
    serdes: SerDes[T] | None = None,
) -> Task[Callback[T]]

Use this reservation for a CALLBACK primitive.

run_in_child_context

run_in_child_context(
    func: Callable[[], Awaitable[T]],
    *,
    sub_type: str | OperationSubType,
    serdes: SerDes[T] | None = None,
    summary_generator: SummaryGenerator[T] | None = None,
    is_virtual: bool = False,
) -> Task[T]

Use this reservation for a CONTEXT primitive.

ExtensionStepResult dataclass

ExtensionStepResult(
    value: T, retry_delay: Duration | None = None
)

Bases: Generic[T]

Outcome returned by one attempt of a stateful extension step.

Attributes

is_retry property

is_retry: bool

Return whether this outcome schedules another attempt.

Methods:

succeed classmethod

succeed(value: T) -> ExtensionStepResult[T]

Complete the extension step with value.

retry classmethod

retry(
    state: T | None, delay: Duration
) -> ExtensionStepResult[T]

Checkpoint state and retry after delay.

Functions:

get_extension_context

get_extension_context() -> ExtensionContext

Return the stable extension-author context for the active durable scope.