Scope context values

What you’ll learn

• Use contextvars.ContextVar for task-local request values.

• Register normal providers (add_factory / add) that call ContextVar.get().

• Override values in nested work with token = var.set(...) and var.reset(token).

Provider with ContextVar

Run locally

uv run python examples/ex_17_scope_context_values/01_provider_contextvar.py

"""Focused example: providers can read values from ``contextvars.ContextVar``."""

from __future__ import annotations

from contextvars import ContextVar

from diwire import Container, Lifetime, Scope

request_value_var: ContextVar[int] = ContextVar("request_value", default=0)


class RequestValue:
    def __init__(self, value: int) -> None:
        self.value = value


def build_request_value() -> RequestValue:
    return RequestValue(value=request_value_var.get())


def main() -> None:
    container = Container()
    container.add_factory(
        build_request_value,
        provides=RequestValue,
        scope=Scope.REQUEST,
        lifetime=Lifetime.TRANSIENT,
    )

    with container.enter_scope(Scope.REQUEST) as request_scope:
        token = request_value_var.set(7)
        try:
            resolved = request_scope.resolve(RequestValue)
        finally:
            request_value_var.reset(token)

    print(f"provider_contextvar={resolved.value}")  # => provider_contextvar=7


if __name__ == "__main__":
    main()

Nested override with tokens

Run locally

uv run python examples/ex_17_scope_context_values/02_nested_scope_inheritance.py

"""Focused example: nested scope work can use ContextVar override/reset tokens."""

from __future__ import annotations

from contextvars import ContextVar

from diwire import Container, Lifetime, Scope

request_id_var: ContextVar[int] = ContextVar("request_id", default=-1)
tenant_var: ContextVar[str] = ContextVar("tenant", default="unset")


def read_request_id() -> int:
    return request_id_var.get()


def read_tenant() -> str:
    return tenant_var.get()


def main() -> None:
    container = Container()
    container.add_factory(
        read_request_id,
        provides=int,
        scope=Scope.REQUEST,
        lifetime=Lifetime.TRANSIENT,
    )
    container.add_factory(
        read_tenant,
        provides=str,
        scope=Scope.REQUEST,
        lifetime=Lifetime.TRANSIENT,
    )

    with container.enter_scope(Scope.REQUEST) as request_scope:
        request_id_token = request_id_var.set(1)
        tenant_token = tenant_var.set("parent")
        try:
            with request_scope.enter_scope(Scope.ACTION) as action_scope:
                inherited_value = action_scope.resolve(int)

                override_token = request_id_var.set(2)
                try:
                    with action_scope.enter_scope(Scope.STEP) as step_scope:
                        overridden_value = step_scope.resolve(int)
                        inherited_parent_key = step_scope.resolve(str)
                finally:
                    request_id_var.reset(override_token)
        finally:
            tenant_var.reset(tenant_token)
            request_id_var.reset(request_id_token)

    print(f"action_inherits_parent={inherited_value}")  # => action_inherits_parent=1
    print(f"step_overrides_parent={overridden_value}")  # => step_overrides_parent=2
    print(
        f"step_inherits_other_parent_key={inherited_parent_key}"
    )  # => step_inherits_other_parent_key=parent


if __name__ == "__main__":
    main()

Injected callable using ContextVar-backed dependency

Run locally

uv run python examples/ex_17_scope_context_values/03_injected_callable_context.py

"""Injected callables can consume dependencies built from ContextVar values."""

from __future__ import annotations

from contextvars import ContextVar

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

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


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


def main() -> None:
    container = Container()
    container.add_factory(
        read_current_value,
        provides=int,
        scope=Scope.REQUEST,
        lifetime=Lifetime.TRANSIENT,
    )

    @resolver_context.inject(scope=Scope.REQUEST)
    def handler(value: Injected[int]) -> int:
        return value

    with container.enter_scope(Scope.REQUEST) as request_scope:
        token = current_value_var.set(7)
        try:
            from_contextvar = handler(diwire_resolver=request_scope)
            overridden = handler(diwire_resolver=request_scope, value=8)
        finally:
            current_value_var.reset(token)

    print(f"from_contextvar={from_contextvar}")  # => from_contextvar=7
    print(f"overridden={overridden}")  # => overridden=8


if __name__ == "__main__":
    main()

Annotated provider keys with ContextVar

Run locally

uv run python examples/ex_17_scope_context_values/04_annotated_context_keys.py

"""Annotated provider keys can be backed by ContextVar values."""

from __future__ import annotations

from contextvars import ContextVar
from typing import Annotated, TypeAlias

from diwire import Component, Container, Lifetime, Scope

ReplicaNumber: TypeAlias = Annotated[int, Component("replica")]

replica_number_var: ContextVar[int] = ContextVar("replica_number", default=0)


class ReplicaConsumer:
    def __init__(self, value: ReplicaNumber) -> None:
        self.value = value


def read_replica_number() -> ReplicaNumber:
    return replica_number_var.get()


def build_consumer(value: ReplicaNumber) -> ReplicaConsumer:
    return ReplicaConsumer(value=value)


def main() -> None:
    container = Container()
    container.add_factory(
        read_replica_number,
        provides=ReplicaNumber,
        scope=Scope.REQUEST,
        lifetime=Lifetime.TRANSIENT,
    )
    container.add_factory(
        build_consumer,
        provides=ReplicaConsumer,
        scope=Scope.REQUEST,
        lifetime=Lifetime.TRANSIENT,
    )

    with container.enter_scope(Scope.REQUEST) as request_scope:
        token = replica_number_var.set(42)
        try:
            resolved = request_scope.resolve(ReplicaConsumer)
            direct = request_scope.resolve(ReplicaNumber)
        finally:
            replica_number_var.reset(token)

    print(f"consumer_value={resolved.value}")  # => consumer_value=42
    print(f"direct_value={direct}")  # => direct_value=42


if __name__ == "__main__":
    main()

ContextVar defaults without opening scopes

Run locally

uv run python examples/ex_17_scope_context_values/05_context_without_scope_open.py

"""ContextVar-backed dependencies work with default values without opening scopes."""

from __future__ import annotations

from contextvars import ContextVar

from diwire import Container, Injected, Lifetime, resolver_context

current_value_var: ContextVar[int] = ContextVar("current_value_no_scope", default=5)


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


def main() -> None:
    container = Container()
    container.add_factory(read_current_value, provides=int, lifetime=Lifetime.TRANSIENT)

    @resolver_context.inject(auto_open_scope=False)
    def handler(value: Injected[int]) -> int:
        return value

    default_value = handler()

    token = current_value_var.set(7)
    try:
        overridden_value = handler()
    finally:
        current_value_var.reset(token)

    print(f"default_value={default_value}")  # => default_value=5
    print(f"overridden_value={overridden_value}")  # => overridden_value=7


if __name__ == "__main__":
    main()