Map¶

Internal implementation module: async_durable_execution._operation.map.

Use map() for durable item fan-out with bounded concurrency, completion policies, and aggregated results.

Implementation for Durable Map operation.

Classes¶

MapItemContext `dataclass` ¶

MapItemContext(
    execution_state: ExecutionState,
    operation_identifier: OperationIdentifier,
    step_id_prefix: str | None = None,
    replaying: bool = False,
    index: int = 0,
    items: Sequence[T] = tuple(),
)

Bases: DurableContext, Generic[T]

Context exposed while a map item function is executing.

Functions:¶

get_map_item_context ¶

get_map_item_context() -> MapItemContext[Any]

Return the active MapItemContext.

map ¶

map(
    func: Callable[
        [U | BatchedInput[Any, U]], Awaitable[T]
    ],
    items: Iterable[U],
    *,
    name: str | None = None,
    max_concurrency: int | None = None,
    completion_config: CompletionConfig | None = None,
    serdes: SerDes | None = None,
    item_serdes: SerDes | None = None,
    summary_generator: SummaryGenerator
    | None = MapSummaryGenerator(),
    nesting_type: NestingType = NESTED,
    item_namer: Callable[[U, int], str] | None = None,
) -> Task[BatchResult[T]]

Start a durable map operation over a collection of items.

map() creates one durable child context per item and calls func with that item. The item function must be async and may contain durable operations such as step() or wait().

The returned object is an asyncio.Task; awaiting it yields a BatchResult. Calling map() without immediately awaiting it schedules the durable operation in the background, consistent with other operation helpers.

By default, map() uses CompletionConfig() with no explicit success threshold or failure tolerance: all-successful completion produces CompletionReason.ALL_COMPLETED, while any observed failure completes the operation as failed. Pass completion_config to use threshold-based or custom completion.

Parameters:

Name	Type	Description	Default
`func`	`Callable[[U \| BatchedInput[Any, U]], Awaitable[T]]`	Async callable that processes each item. It receives the original item value and returns that item's result.	required
`items`	`Iterable[U]`	Items to process.	required
`name`	`str \| None`	Optional durable operation name.	`None`
`max_concurrency`	`int \| None`	Optional limit for in-flight items. A suspended item retains its slot until it reaches a terminal state.	`None`
`completion_config`	`CompletionConfig \| None`	Optional completion policy. Use `CompletionConfig.thresholds()`, `first_successful()`, `all_completed()`, `all_successful()`, or `custom()`.	`None`
`serdes`	`SerDes \| None`	Optional serializer for the final `BatchResult`.	`None`
`item_serdes`	`SerDes \| None`	Optional serializer for each item result.	`None`
`summary_generator`	`SummaryGenerator \| None`	Optional callable used to summarize oversized checkpoint payloads.	`MapSummaryGenerator()`
`nesting_type`	`NestingType`	Whether map iterations use nested or flat operation identifiers.	`NESTED`
`item_namer`	`Callable[[U, int], str] \| None`	Optional callable for naming map item iterations.	`None`

Returns:

Type	Description
`Task[BatchResult[T]]`	An `asyncio.Task` that resolves to a `BatchResult` containing one `BatchItem` per input item.

Raises:

Type	Description
`RuntimeError`	If called outside a durable context.