Function injection

What you’ll learn

• Use Injected[T] to mark injected parameters.

• Wrap callables with @resolver_context.inject(...).

• Control scope behavior for injected wrappers.

Signature filtering

Run locally

uv run python examples/ex_06_function_injection/01_signature_filtering.py

"""Focused example: ``Injected[T]`` and public signature filtering."""

from __future__ import annotations

import inspect
from dataclasses import dataclass

from diwire import Container, Injected, resolver_context


@dataclass(slots=True)
class User:
    email: str


def main() -> None:
    container = Container()
    container.add_instance(User(email="user@example.com"))

    @resolver_context.inject
    def handler(user_email: str, user: Injected[User], user_name: str) -> str:
        return f"{user_email}|{user_name}|{user.email}"

    signature = "('" + "','".join(inspect.signature(handler).parameters) + "')"
    print(f"signature={signature}")  # => signature=('user_email','user_name')


if __name__ == "__main__":
    main()

Override injected values

Run locally

uv run python examples/ex_06_function_injection/02_override_injected.py

"""Focused example: caller override for injected parameters."""

from __future__ import annotations

from dataclasses import dataclass

from diwire import Container, Injected, resolver_context


@dataclass(slots=True)
class User:
    email: str


def main() -> None:
    container = Container()
    container.add_instance(User(email="container@example.com"))

    @resolver_context.inject
    def handler(user: Injected[User]) -> str:
        return user.email

    default_value = handler()
    override_value = handler(user=User(email="override@example.com"))

    print(f"default_injected={default_value}")  # => default_injected=container@example.com
    print(f"override_injected={override_value}")  # => override_injected=override@example.com


if __name__ == "__main__":
    main()

Auto-open scope cleanup

Run locally

uv run python examples/ex_06_function_injection/03_auto_open_scope_cleanup.py

"""Focused example: ``auto_open_scope`` with scoped cleanup."""

from __future__ import annotations

from collections.abc import Generator

from diwire import Container, Injected, Lifetime, Scope, resolver_context


class Resource:
    pass


def main() -> None:
    container = Container()
    state = {"cleaned": False}

    def provide_resource() -> Generator[Resource, None, None]:
        try:
            yield Resource()
        finally:
            state["cleaned"] = True

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

    @resolver_context.inject(scope=Scope.REQUEST, auto_open_scope=True)
    def handler(resource: Injected[Resource]) -> Resource:
        return resource

    _ = handler()
    print(f"auto_scope_cleanup={state['cleaned']}")  # => auto_scope_cleanup=True


if __name__ == "__main__":
    main()

Nested wrappers

Run locally

uv run python examples/ex_06_function_injection/04_nested_wrappers.py

"""Focused example: nested injected wrappers share one active resolver."""

from __future__ import annotations

from dataclasses import dataclass

from diwire import Container, Injected, Lifetime, Scope, resolver_context


class RequestDependency:
    pass


@dataclass(slots=True)
class InnerService:
    dependency: RequestDependency


@dataclass(slots=True)
class OuterService:
    inner: InnerService
    dependency: RequestDependency


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

    @resolver_context.inject
    def build_inner(dependency: Injected[RequestDependency]) -> InnerService:
        return InnerService(dependency=dependency)

    @resolver_context.inject
    def build_outer(
        inner: Injected[InnerService],
        dependency: Injected[RequestDependency],
    ) -> OuterService:
        return OuterService(inner=inner, dependency=dependency)

    container.add_factory(
        build_inner,
        provides=InnerService,
        scope=Scope.REQUEST,
        lifetime=Lifetime.SCOPED,
    )
    container.add_factory(
        build_outer,
        provides=OuterService,
        scope=Scope.REQUEST,
        lifetime=Lifetime.TRANSIENT,
    )

    with container.enter_scope() as request_scope:
        resolved = request_scope.resolve(OuterService)

    print(
        f"nested_scope_identity={resolved.inner.dependency is resolved.dependency}",
    )  # => nested_scope_identity=True


if __name__ == "__main__":
    main()

Auto-open scope reuse

Run locally

uv run python examples/ex_06_function_injection/05_auto_open_scope_reuse.py

"""Focused example: auto-open scope reuses already-open resolvers."""

from __future__ import annotations

from collections.abc import Generator
from contextvars import ContextVar

from diwire import Container, Injected, Lifetime, Scope, resolver_context


class RequestResource:
    pass


current_value_var: ContextVar[int] = ContextVar("auto_open_scope_reuse_value", default=0)


def main() -> None:
    container = Container()
    cleanup_state = {"cleaned": False}

    def provide_request_resource() -> Generator[RequestResource, None, None]:
        try:
            yield RequestResource()
        finally:
            cleanup_state["cleaned"] = True

    def provide_current_value() -> int:
        return current_value_var.get()

    container.add_generator(
        provide_request_resource,
        provides=RequestResource,
        scope=Scope.REQUEST,
        lifetime=Lifetime.SCOPED,
    )
    container.add_factory(provide_current_value, provides=int, scope=Scope.SESSION)

    @resolver_context.inject(scope=Scope.REQUEST, auto_open_scope=True)
    def use_request_resource(resource: Injected[RequestResource]) -> RequestResource:
        return resource

    with container.enter_scope(Scope.REQUEST) as request_scope:
        resolved_resource = use_request_resource(diwire_resolver=request_scope)
        print(
            f"target_scope_reused={isinstance(resolved_resource, RequestResource) and not cleanup_state['cleaned']}",
        )  # => target_scope_reused=True

    print(
        f"cleanup_after_outer_scope={cleanup_state['cleaned']}"
    )  # => cleanup_after_outer_scope=True

    @resolver_context.inject(scope=Scope.SESSION, auto_open_scope=True)
    def read_value(value: Injected[int]) -> int:
        return value

    with (
        container.enter_scope(Scope.SESSION) as session_scope,
        session_scope.enter_scope(Scope.REQUEST) as request_scope,
    ):
        token = current_value_var.set(22)
        try:
            resolved_value = read_value(diwire_resolver=request_scope)
        finally:
            current_value_var.reset(token)
        print(
            f"deeper_scope_contextvar_reused={resolved_value}"
        )  # => deeper_scope_contextvar_reused=22


if __name__ == "__main__":
    main()

Async deep dive

Run locally

uv run python examples/ex_06_function_injection/06_function_injection_async_details.py

"""Async function-injection deep dive.

This focused script covers async callables using ``Injected[T]`` and caller
overrides for injected async parameters.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass

from diwire import Container, Injected, resolver_context


@dataclass(slots=True)
class AsyncUser:
    email: str


async def main() -> None:
    container = Container()
    container.add_instance(AsyncUser(email="async@example.com"))

    @resolver_context.inject
    async def handler(user: Injected[AsyncUser]) -> str:
        return user.email

    default_value = await handler()
    overridden_value = await handler(user=AsyncUser(email="override@example.com"))

    if default_value != "async@example.com":
        msg = "Unexpected default async injection result"
        raise TypeError(msg)
    if overridden_value != "override@example.com":
        msg = "Unexpected async override injection result"
        raise TypeError(msg)


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