without_async

The asyncio primitives without is built from: scoped background tasks, bounded concurrency, optional timeouts, and durations that cross an integer boundary.

without_async

Milliseconds dataclass

Milliseconds(count: int)

A duration a boundary carrying integer milliseconds can express exactly.

The millisecond counterpart of Seconds, with the same shape: a count in, duration back out, and of to parse one from a timedelta.

yield Retry(Milliseconds.of(timedelta(seconds=30)))

count instance-attribute

count: int

duration property

duration: timedelta

The timedelta this count of milliseconds names.

of classmethod

of(duration: timedelta) -> Milliseconds

Parse a timedelta into a count of milliseconds, refusing a finer duration.

Seconds dataclass

Seconds(count: int)

A duration a boundary carrying integer seconds can express exactly.

The count itself, rather than a timedelta that happens to divide by one second: a duration finer than the boundary carries is not something this can be constructed from, so nothing downstream has a truncation left to do.

tcp_keepalive(idle=Seconds(60), interval=Seconds(10))

duration is the timedelta back out, for arithmetic and for anything that takes a plain duration. of is the way in from one, and the only place the question of whether it divides is asked:

Seconds.of(settings.keepalive_idle)  # raises on a duration finer than a second

count instance-attribute

count: int

duration property

duration: timedelta

The timedelta this count of seconds names.

of classmethod

of(duration: timedelta) -> Seconds

Parse a timedelta into a count of seconds, refusing a finer duration.

as_async_iterator async

as_async_iterator(
    items: AsyncIterable[T] | Iterable[T],
) -> AsyncIterator[T]

Normalize a sync or async iterable into a single async iterator.

Lets code that consumes via async for/anext accept either kind without branching on the iteration protocol at every use.

background_task async

background_task(
    coro: Coroutine[object, object, T],
) -> AsyncIterator[Task[T]]

Run coro as a task for the duration of the with block.

The task is started on entry and cancelled (then awaited) on exit, so it is bounded by the block and never leaks. If it finishes on its own with an exception, that surfaces when the block exits.

cancel_futures async

cancel_futures(
    futures: Iterable[Future[T] | None],
) -> None

Cancel every future, then await them all so their teardown completes.

Two phases on purpose: cancelling the whole set before awaiting any of them lets them tear down concurrently, instead of serially cancelling and waiting for one at a time. None entries are skipped, so a caller holding an optional task (task: asyncio.Task | None) can pass it without a guard. The futures are materialized first, so a caller may pass a live set the awaits will mutate. Each future's own CancelledError is suppressed; any other exception it raises during teardown propagates, which also ends the loop, so the futures behind it in the set are cancelled but never awaited. That is the right order for a set of pending futures, where nothing has an exception to raise; pass one that may already hold a failure and the ones behind it lose their teardown to it. Filter to what is still running when the set can contain both.

limit_concurrency async

limit_concurrency(
    aws: AsyncIterable[Awaitable[T]]
    | Iterable[Awaitable[T]],
    limit: int,
) -> AsyncIterator[Future[T]]

Run awaitables from aws with at most limit in flight, yielding each as it finishes.

A bounded-concurrency driver: it pulls the next awaitable from aws only while fewer than limit are already running. So a lazy source (an async generator that produces each unit of work on demand) is never advanced past the limit. That is what lets it gate a side-effecting source: an accept loop whose generator awaits socket.accept() only when pulled will never accept more connections than it can serve.

Each completed awaitable is yielded as a Future; call .result() on it to read the value or re-raise its exception. On early exit or cancellation, any still-running awaitables are cancelled and awaited, so none outlive the iteration.

limit must be at least 1; a non-positive limit is a ValueError, since it could only ever stall the source rather than run it.

Adapted from Limiting concurrency in asyncio.

settled async

settled(future: Future[T]) -> T

Await future, and hold on through a cancellation of the caller until it is done.

For a future whose work has already happened by the time it is awaited (a store write recording an effect that has been performed, a renewal a server has already applied), where cancelling the await would not undo the work but would lose the record of it. The future is shielded, so the caller's cancellation does not reach it, and when one arrives the caller waits for the future to finish before the CancelledError propagates. What the future came to is then there to read off it, on the cancellation path as on the ordinary one.

It waits through repeated cancellation, which is the ordinary case rather than stubbornness: a task torn down by a task group or a gather is commonly cancelled twice, once by the combinator and once by its caller's own teardown. wait rather than an await for that, so the future's own failure belongs to whoever reads it afterwards and does not replace the cancellation.

sleep_forever async

sleep_forever() -> None

Suspend the current task until it is cancelled.

The idiom for a coroutine whose job is to stay alive until its surrounding scope tears it down: a server's run loop holding a bound socket open, a process that should idle until signalled. It awaits a future that never resolves, so it consumes nothing and ends only on cancellation.

timeout async

timeout(duration: timedelta | None) -> AsyncIterator[None]

Bound the with block by duration, or leave it unbounded when None.

A timedelta-typed, nullable wrapper over asyncio.timeout: None disables the bound (an always-open context), and a duration raises TimeoutError if the block outlives it. Modelling "no limit" as None keeps that choice a first-class value at the call site, rather than a sentinel float threaded through the same parameter.