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())