Async

What you’ll learn

• Register an async def factory and resolve it via await container.aresolve(...).

• Use an async-generator provider for deterministic async cleanup within a scope.

Async factory + aresolve

Run locally

uv run python examples/ex_18_async/01_async_factory_aresolve.py

"""Focused example: async factory registration with ``await container.aresolve(...)``."""

from __future__ import annotations

import asyncio

from diwire import Container


class AsyncService:
    def __init__(self, value: str) -> None:
        self.value = value


async def main() -> None:
    container = Container()

    async def build_async_service() -> AsyncService:
        await asyncio.sleep(0)
        return AsyncService(value="ok")

    container.add_factory(build_async_service, provides=AsyncService)

    service = await container.aresolve(AsyncService)
    print(f"async_factory_value={service.value}")  # => async_factory_value=ok


if __name__ == "__main__":
    asyncio.run(main())

Async generator cleanup

Run locally

uv run python examples/ex_18_async/02_async_generator_cleanup.py

"""Focused example: async-generator provider cleanup on async scope exit."""

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator

from diwire import Container, Lifetime, Scope


class AsyncResource:
    pass


async def main() -> None:
    container = Container()
    state = {"closed": 0}

    async def provide_async_resource() -> AsyncGenerator[AsyncResource, None]:
        try:
            yield AsyncResource()
        finally:
            await asyncio.sleep(0)
            state["closed"] += 1

    container.add_generator(
        provide_async_resource,
        provides=AsyncResource,
        scope=Scope.REQUEST,
        lifetime=Lifetime.SCOPED,
    )

    async with container.enter_scope() as request_scope:
        _ = await request_scope.aresolve(AsyncResource)
        closed_before_exit = state["closed"]

    print(
        f"async_cleanup_after_exit={closed_before_exit == 0 and state['closed'] == 1}",
    )  # => async_cleanup_after_exit=True


if __name__ == "__main__":
    asyncio.run(main())

Cancelled job cleanup

A background job can be cancelled while it waits for input. Give each job its own request scope. Keep the scope inside the task that uses the resource. An async-generator provider can then return a borrowed connection in its finally block when the job stops.

Cancel the task once, then await it. Catch asyncio.CancelledError outside the job’s scope. In this example, release:done appears before cancelled. The event signals that the job has acquired its connection before cancellation. No timed delay is needed to select that point.

The connection is an in-memory stand-in. Replace release() with the real driver’s async release operation. Use the async scope on success and failure too. A scoped connection is reused inside one job; the next job gets a new one.

This example covers one cancellation during work. A second cancellation during cleanup, a failed release operation, or process termination needs a separate shutdown policy. Do not assume that a finally block makes cleanup immune to these events.

Run locally

uv run python examples/ex_18_async/03_cancelled_job_cleanup.py

"""Return a borrowed connection when an async job is cancelled.

Cancel once and await the task. The job owns its request scope, so async
cleanup finishes before the caller receives CancelledError. This in-memory
connection needs no database or network service.
"""

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator
from dataclasses import dataclass

from diwire import Container, Lifetime, Scope


@dataclass
class Connection:
    events: list[str]
    released: bool = False

    async def release(self) -> None:
        self.events.append("release:start")
        # A real driver would await its connection release operation here.
        await asyncio.sleep(0)
        self.released = True
        self.events.append("release:done")


def create_container(events: list[str]) -> Container:
    container = Container()

    async def borrow_connection() -> AsyncGenerator[Connection, None]:
        connection = Connection(events)
        events.append("acquire")
        try:
            yield connection
        finally:
            await connection.release()

    container.add_generator(
        borrow_connection,
        provides=Connection,
        lifetime=Lifetime.SCOPED,
        scope=Scope.REQUEST,
    )
    return container


async def run_job(
    container: Container,
    started: asyncio.Event,
    finish: asyncio.Event,
) -> None:
    async with container.enter_scope(Scope.REQUEST) as job:
        connection = await job.aresolve(Connection)
        connection.events.append("run")
        started.set()
        # Stand in for work that waits for more input.
        await finish.wait()
        connection.events.append("complete")


async def main() -> None:
    events: list[str] = []
    container = create_container(events)
    started = asyncio.Event()
    task = asyncio.create_task(run_job(container, started, asyncio.Event()))
    try:
        await started.wait()
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            events.append("cancelled")
    finally:
        await container.aclose()

    print(
        f"lifecycle={','.join(events)}"
    )  # => lifecycle=acquire,run,release:start,release:done,cancelled


if __name__ == "__main__":
    asyncio.run(main())