Scopes & cleanup

What you’ll learn

  • Use enter_scope() to create nested scopes.

  • Use Lifetime.SCOPED for per-scope caching.

  • Use generator providers for deterministic cleanup.

Scope transitions

Run locally

uv run python examples/ex_04_scopes_and_cleanup/01_scope_transitions.py
"""Focused example: default and explicit scope transitions."""

from __future__ import annotations

from diwire import Container, Lifetime, Scope


class RequestDependency:
    pass


def _resolver_scope_name(resolver: object) -> str:
    inner_resolver = getattr(resolver, "_resolver", resolver)
    return type(inner_resolver).__name__.removeprefix("_").removesuffix("Resolver").upper()


def main() -> None:
    container = Container()
    container.add(
        RequestDependency,
        provides=RequestDependency,
        scope=Scope.REQUEST,
        lifetime=Lifetime.SCOPED,
    )

    with container.enter_scope() as request_scope:
        default_scope = _resolver_scope_name(request_scope)

    with container.enter_scope(Scope.ACTION) as action_scope:
        resolved = action_scope.resolve(RequestDependency)

    print(f"enter_scope_default={default_scope}")  # => enter_scope_default=REQUEST
    print(
        f"action_scope_can_resolve_request_scoped={isinstance(resolved, RequestDependency)}",
    )  # => action_scope_can_resolve_request_scoped=True


if __name__ == "__main__":
    main()

Scope mismatch

Run locally

uv run python examples/ex_04_scopes_and_cleanup/02_scope_mismatch.py
"""Focused example: ``DIWireScopeMismatchError`` from root resolution."""

from __future__ import annotations

from diwire import Container, Scope
from diwire.exceptions import DIWireScopeMismatchError


class RequestDependency:
    pass


def main() -> None:
    container = Container()
    container.add(
        RequestDependency,
        provides=RequestDependency,
        scope=Scope.REQUEST,
    )

    try:
        container.resolve(RequestDependency)
    except DIWireScopeMismatchError as error:
        error_name = type(error).__name__

    print(f"scope_mismatch_error={error_name}")  # => scope_mismatch_error=DIWireScopeMismatchError


if __name__ == "__main__":
    main()

Scoped cleanup

Run locally

uv run python examples/ex_04_scopes_and_cleanup/03_scoped_cleanup.py
"""Focused example: scoped resource cleanup on scope exit."""

from __future__ import annotations

from collections.abc import Generator

from diwire import Container, Lifetime, Scope


class ScopedResource:
    pass


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

    def provide_resource() -> Generator[ScopedResource, None, None]:
        try:
            yield ScopedResource()
        finally:
            state["closed"] += 1

    container.add_generator(
        provide_resource,
        provides=ScopedResource,
        scope=Scope.REQUEST,
        lifetime=Lifetime.SCOPED,
    )

    with container.enter_scope() as request_scope:
        _ = request_scope.resolve(ScopedResource)
        closed_inside_scope = state["closed"]

    closed_after_exit = state["closed"]
    print(
        f"scoped_cleanup_after_exit={closed_inside_scope == 0 and closed_after_exit == 1}",
    )  # => scoped_cleanup_after_exit=True


if __name__ == "__main__":
    main()

Singleton cleanup

Run locally

uv run python examples/ex_04_scopes_and_cleanup/04_singleton_cleanup.py
"""Focused example: singleton generator cleanup on ``container.close()``."""

from __future__ import annotations

from collections.abc import Generator

from diwire import Container, Lifetime, Scope


class SingletonResource:
    pass


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

    def provide_resource() -> Generator[SingletonResource, None, None]:
        try:
            yield SingletonResource()
        finally:
            state["closed"] += 1

    container.add_generator(
        provide_resource,
        provides=SingletonResource,
        scope=Scope.APP,
        lifetime=Lifetime.SCOPED,
    )

    _ = container.resolve(SingletonResource)
    closed_before = state["closed"]
    container.close()
    print(
        f"singleton_cleanup_on_close={closed_before == 0 and state['closed'] == 1}",
    )  # => singleton_cleanup_on_close=True


if __name__ == "__main__":
    main()