Skip to content

Dependency Injection

ProviderRegistry, Scope, and lifecycle types.

di

Public DI declaration surface.

__all__ module-attribute 

__all__ = [
    "LifecycleDetectionWarning",
    "ProviderLifecycle",
    "ProviderRegistry",
    "Scope",
]

ProviderRegistry 

ProviderRegistry()

Mutable provider registry with seal guard and edge capture.

Structurally satisfies the taskq._di.types.ProviderRegistry Protocol without explicit inheritance — the Protocol is runtime_checkable for isinstance checks, but the concrete class does not inherit from it to avoid forcing runtime_checkable constraints on the implementation.

Source code in src/taskq/_di/registry.py 
52
53
54
55
56
57
58
def __init__(self) -> None:
    self._providers: dict[type, ProviderEntry[object]] = {}
    self._dep_edges: list[tuple[type, type, Scope | None]] = []
    self._validated: bool = False
    self._sealed: bool = False
    self._plan_cache: dict[tuple[str, Scope], list[type]] = {}
    self._validating: bool = False

providers property 

providers: dict[type, ProviderEntry[object]]

Shallow copy preventing external mutation.

Why: the internal map is dict[type, ProviderEntry[object]] per the erasure boundary. Returning a copy prevents callers from mutating registry state.

register_value 

register_value(
    type_: type[T], scope: Scope, value: T
) -> None
Source code in src/taskq/_di/registry.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def register_value[T](self, type_: type[T], scope: Scope, value: T) -> None:
    if self._sealed:
        raise RuntimeError("registry is sealed after validate()")
    if type_ in self._providers:
        raise ValueError(f"{type_!r} is already registered")
    entry = ProviderEntry(
        type_=type_,
        scope=scope,
        kind="value",
        impl=value,
        factory_shape=FactoryShape.VALUE,
    )
    self._providers[type_] = entry
    logger.debug(
        "provider-registered",
        type_name=type_.__qualname__,
        scope=scope.name,
        kind="value",
    )

register_factory 

register_factory(
    type_: type[T], scope: Scope, factory: Factory[T]
) -> None
Source code in src/taskq/_di/registry.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def register_factory[T](self, type_: type[T], scope: Scope, factory: Factory[T]) -> None:
    if self._sealed:
        raise RuntimeError("registry is sealed after validate()")
    if type_ in self._providers:
        raise ValueError(f"{type_!r} is already registered")
    if not callable(factory):
        raise TypeError(f"factory must be callable, got {type(factory).__qualname__}")
    if inspect.isasyncgenfunction(factory):
        shape = FactoryShape.ASYNC_GENERATOR
    elif inspect.isgeneratorfunction(factory):
        shape = FactoryShape.SYNC_GENERATOR
    elif inspect.iscoroutinefunction(factory):
        shape = FactoryShape.ASYNC_CALLABLE
    else:
        shape = FactoryShape.SYNC_CALLABLE

    lifecycle = detect_factory_lifecycle(factory)

    if lifecycle is ProviderLifecycle.SyncGenerator:
        warnings.warn(
            LifecycleDetectionWarning(
                f"{getattr(factory, '__qualname__', repr(factory))} is a sync "
                f"generator factory; its cleanup will run synchronously and "
                f"cannot be awaited by the DI engine."
            ),
            stacklevel=2,
        )
        logger.warning(
            "sync_generator_registered",
            factory_name=getattr(factory, "__qualname__", repr(factory)),
            scope=scope.name,
        )

    edges = _collect_dep_edges(factory, type_)
    self._dep_edges.extend(edges)

    entry = ProviderEntry(
        type_=type_,
        scope=scope,
        kind="factory",
        impl=factory,
        factory_shape=shape,
        lifecycle=lifecycle,
    )
    self._providers[type_] = entry
    logger.debug(
        "provider-registered",
        type_name=type_.__qualname__,
        scope=scope.name,
        kind="factory",
        factory_shape=shape.name,
    )

register_class 

register_class(
    type_: type[T],
    scope: Scope,
    *,
    lifecycle: ProviderLifecycle | None = None,
) -> None
Source code in src/taskq/_di/registry.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def register_class[T](
    self,
    type_: type[T],
    scope: Scope,
    *,
    lifecycle: ProviderLifecycle | None = None,
) -> None:
    if self._sealed:
        raise RuntimeError("registry is sealed after validate()")
    if type_ in self._providers:
        raise ValueError(f"{type_!r} is already registered")

    resolved_lifecycle = lifecycle if lifecycle is not None else detect_lifecycle(type_)

    if lifecycle is None:
        if resolved_lifecycle is ProviderLifecycle.SyncCloseable:
            warnings.warn(
                LifecycleDetectionWarning(
                    f"{type_.__qualname__} has close() but no aclose(); "
                    f"the DI engine will call close() via asyncio.to_thread "
                    f"at teardown. Prefer an async close (aclose) for "
                    f"async-native code."
                ),
                stacklevel=2,
            )
            logger.warning(
                "sync_close_registered",
                class_name=type_.__qualname__,
                scope=scope.name,
            )
        elif (
            resolved_lifecycle is ProviderLifecycle.Plain
            and hasattr(type_, "__enter__")
            and not hasattr(type_, "__aenter__")
        ):
            warnings.warn(
                LifecycleDetectionWarning(
                    f"{type_.__qualname__} has __enter__/__exit__ but no "
                    f"__aenter__/__aexit__; sync context managers are not "
                    f"supported by the async DI engine. The class will be "
                    f"managed as Plain (no teardown)."
                ),
                stacklevel=2,
            )
            logger.warning(
                "sync_context_manager_not_supported",
                class_name=type_.__qualname__,
                scope=scope.name,
            )

    edges = _collect_dep_edges(type_.__init__, type_)
    self._dep_edges.extend(edges)

    entry = ProviderEntry(
        type_=type_,
        scope=scope,
        kind="class",
        impl=type_,
        factory_shape=FactoryShape.CLASS,
        lifecycle=resolved_lifecycle,
    )
    self._providers[type_] = entry
    logger.debug(
        "provider-registered",
        type_name=type_.__qualname__,
        scope=scope.name,
        kind="class",
    )

has_provider 

has_provider(type_: type) -> bool
Source code in src/taskq/_di/registry.py 
202
203
def has_provider(self, type_: type) -> bool:
    return type_ in self._providers

get 

get(type_: type[T]) -> ProviderEntry[T]
Source code in src/taskq/_di/registry.py 
205
206
207
208
209
210
211
212
213
214
def get[T](self, type_: type[T]) -> ProviderEntry[T]:
    entry = self._providers.get(type_)
    if entry is None:
        raise MissingProvider(
            type_name=type_.__qualname__,
            required_by="<unknown>",
        )
    return cast(
        ProviderEntry[T], entry
    )  # Why: internal map is dict[type, ProviderEntry[object]] (erasure boundary); get[T] recovers the type parameter

validate 

validate(
    actors: list[ActorRef[Any, Any]] | None = None,
    rate_limit_registry: RateLimitRegistry | None = None,
) -> None

Walk all providers and actors; raise on first error; seal on success.

Pure graph-walk over registration-time metadata. Never invokes a factory, calls a resolver, or performs await — the entire algorithm is synchronous introspection on _providers and _dep_edges.

actors: explicit list of ActorRef instances whose DI parameter annotations are walked for MissingProvider checks and plan-cache population. Defaults to None (no actor walk — only provider→provider edges are validated).

rate_limit_registry: when provided, each actor's rate_limits and reservations lists are checked against the registry's dicts. Unknown names raise MissingProvider at startup. When None (default), the name-check phase is skipped entirely.

Why: ActorRef[Any, Any] is the sanctioned erasure
boundary for the heterogeneous actor registry — each ActorRef has
different P and R type parameters, and validate() does not need
per-actor narrowing. The same erasure is used at the existing API
boundary in worker/run.py:283 and cli.py:98.
Source code in src/taskq/_di/registry.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def validate(
    self,
    actors: list["ActorRef[Any, Any]"] | None = None,
    rate_limit_registry: "RateLimitRegistry | None" = None,
) -> None:
    """Walk all providers and actors; raise on first error; seal on success.

    Pure graph-walk over registration-time metadata. Never invokes a
    factory, calls a resolver, or performs await — the entire algorithm
    is synchronous introspection on ``_providers`` and ``_dep_edges``.

    actors: explicit list of ActorRef instances whose DI parameter
    annotations are walked for MissingProvider checks and plan-cache
    population. Defaults to None (no actor walk — only
    provider→provider edges are validated).

    rate_limit_registry: when provided, each actor's ``rate_limits``
    and ``reservations`` lists are checked against the registry's
    dicts. Unknown names raise ``MissingProvider`` at startup.
    When ``None`` (default), the name-check phase is
    skipped entirely.

    # Why: ActorRef[Any, Any] is the sanctioned erasure
    # boundary for the heterogeneous actor registry — each ActorRef has
    # different P and R type parameters, and validate() does not need
    # per-actor narrowing. The same erasure is used at the existing API
    # boundary in worker/run.py:283 and cli.py:98.
    """
    if self._validated:
        return
    if self._validating:
        raise RuntimeError("validate() called recursively or concurrently")
    self._validating = True
    try:
        self._plan_cache = run_validation(
            self._providers,
            self._dep_edges,
            actors,
            rate_limit_registry,
        )
        if actors is not None:
            _emit_redundant_override_warnings(actors, self._providers)
        self._validated = True
        self._sealed = True
        logger.info(
            "registry-validated",
            provider_count=len(self._providers),
            actor_count=len(actors) if actors else 0,
        )
    finally:
        self._validating = False

LifecycleDetectionWarning

Bases: UserWarning

Emitted for DI usage hazards detected during DI operations.

Specific timing varies per emission site and is documented at each emission. Fires for redundant call-site Scope overrides , sync close detection on registered classes, and other drift-prone API patterns. Subclass of UserWarning so pytest captures it under default filters via pytest.warns(LifecycleDetectionWarning).

Scope

Bases: IntEnum

DI scope lifetime. Higher value = narrower scope.

PROCESS: worker process startup -> exit (singletons). THREAD: worker thread spawn -> thread close (placeholder for multi-thread worker). LOOP: worker loop start -> loop close (pools, long-lived clients). TRANSIENT: fresh per injection point - no cache.

PROCESS class-attribute instance-attribute 

PROCESS = 0

THREAD class-attribute instance-attribute 

THREAD = 1

LOOP class-attribute instance-attribute 

LOOP = 2

TRANSIENT class-attribute instance-attribute 

TRANSIENT = 3

ProviderLifecycle

Bases: Enum

Provider lifecycle shapes for resolution-time dispatch.

Plain class-attribute instance-attribute 

Plain = 'plain'

AsyncContextManager class-attribute instance-attribute 

AsyncContextManager = 'acm'

AsyncCloseable class-attribute instance-attribute 

AsyncCloseable = 'acloseable'

SyncCloseable class-attribute instance-attribute 

SyncCloseable = 'scloseable'

AsyncGenerator class-attribute instance-attribute 

AsyncGenerator = 'asyncgen'

SyncGenerator class-attribute instance-attribute 

SyncGenerator = 'syncgen'

PlainFactory class-attribute instance-attribute 

PlainFactory = 'plainfactory'

registry

Concrete ProviderRegistry — registration API and seal mechanics.

The public surface (register_value / register_factory / register_class, has_provider, get[T], providers, validate) matches the published block. The validate() method delegates the five-phase startup validation algorithm to _di._validate.run_validation.

logger module-attribute 

logger = structlog.get_logger('taskq._di.registry')

ProviderRegistry 

ProviderRegistry()

Mutable provider registry with seal guard and edge capture.

Structurally satisfies the taskq._di.types.ProviderRegistry Protocol without explicit inheritance — the Protocol is runtime_checkable for isinstance checks, but the concrete class does not inherit from it to avoid forcing runtime_checkable constraints on the implementation.

Source code in src/taskq/_di/registry.py 
52
53
54
55
56
57
58
def __init__(self) -> None:
    self._providers: dict[type, ProviderEntry[object]] = {}
    self._dep_edges: list[tuple[type, type, Scope | None]] = []
    self._validated: bool = False
    self._sealed: bool = False
    self._plan_cache: dict[tuple[str, Scope], list[type]] = {}
    self._validating: bool = False

providers property 

providers: dict[type, ProviderEntry[object]]

Shallow copy preventing external mutation.

Why: the internal map is dict[type, ProviderEntry[object]] per the erasure boundary. Returning a copy prevents callers from mutating registry state.

register_value 

register_value(
    type_: type[T], scope: Scope, value: T
) -> None
Source code in src/taskq/_di/registry.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def register_value[T](self, type_: type[T], scope: Scope, value: T) -> None:
    if self._sealed:
        raise RuntimeError("registry is sealed after validate()")
    if type_ in self._providers:
        raise ValueError(f"{type_!r} is already registered")
    entry = ProviderEntry(
        type_=type_,
        scope=scope,
        kind="value",
        impl=value,
        factory_shape=FactoryShape.VALUE,
    )
    self._providers[type_] = entry
    logger.debug(
        "provider-registered",
        type_name=type_.__qualname__,
        scope=scope.name,
        kind="value",
    )

register_factory 

register_factory(
    type_: type[T], scope: Scope, factory: Factory[T]
) -> None
Source code in src/taskq/_di/registry.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def register_factory[T](self, type_: type[T], scope: Scope, factory: Factory[T]) -> None:
    if self._sealed:
        raise RuntimeError("registry is sealed after validate()")
    if type_ in self._providers:
        raise ValueError(f"{type_!r} is already registered")
    if not callable(factory):
        raise TypeError(f"factory must be callable, got {type(factory).__qualname__}")
    if inspect.isasyncgenfunction(factory):
        shape = FactoryShape.ASYNC_GENERATOR
    elif inspect.isgeneratorfunction(factory):
        shape = FactoryShape.SYNC_GENERATOR
    elif inspect.iscoroutinefunction(factory):
        shape = FactoryShape.ASYNC_CALLABLE
    else:
        shape = FactoryShape.SYNC_CALLABLE

    lifecycle = detect_factory_lifecycle(factory)

    if lifecycle is ProviderLifecycle.SyncGenerator:
        warnings.warn(
            LifecycleDetectionWarning(
                f"{getattr(factory, '__qualname__', repr(factory))} is a sync "
                f"generator factory; its cleanup will run synchronously and "
                f"cannot be awaited by the DI engine."
            ),
            stacklevel=2,
        )
        logger.warning(
            "sync_generator_registered",
            factory_name=getattr(factory, "__qualname__", repr(factory)),
            scope=scope.name,
        )

    edges = _collect_dep_edges(factory, type_)
    self._dep_edges.extend(edges)

    entry = ProviderEntry(
        type_=type_,
        scope=scope,
        kind="factory",
        impl=factory,
        factory_shape=shape,
        lifecycle=lifecycle,
    )
    self._providers[type_] = entry
    logger.debug(
        "provider-registered",
        type_name=type_.__qualname__,
        scope=scope.name,
        kind="factory",
        factory_shape=shape.name,
    )

register_class 

register_class(
    type_: type[T],
    scope: Scope,
    *,
    lifecycle: ProviderLifecycle | None = None,
) -> None
Source code in src/taskq/_di/registry.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def register_class[T](
    self,
    type_: type[T],
    scope: Scope,
    *,
    lifecycle: ProviderLifecycle | None = None,
) -> None:
    if self._sealed:
        raise RuntimeError("registry is sealed after validate()")
    if type_ in self._providers:
        raise ValueError(f"{type_!r} is already registered")

    resolved_lifecycle = lifecycle if lifecycle is not None else detect_lifecycle(type_)

    if lifecycle is None:
        if resolved_lifecycle is ProviderLifecycle.SyncCloseable:
            warnings.warn(
                LifecycleDetectionWarning(
                    f"{type_.__qualname__} has close() but no aclose(); "
                    f"the DI engine will call close() via asyncio.to_thread "
                    f"at teardown. Prefer an async close (aclose) for "
                    f"async-native code."
                ),
                stacklevel=2,
            )
            logger.warning(
                "sync_close_registered",
                class_name=type_.__qualname__,
                scope=scope.name,
            )
        elif (
            resolved_lifecycle is ProviderLifecycle.Plain
            and hasattr(type_, "__enter__")
            and not hasattr(type_, "__aenter__")
        ):
            warnings.warn(
                LifecycleDetectionWarning(
                    f"{type_.__qualname__} has __enter__/__exit__ but no "
                    f"__aenter__/__aexit__; sync context managers are not "
                    f"supported by the async DI engine. The class will be "
                    f"managed as Plain (no teardown)."
                ),
                stacklevel=2,
            )
            logger.warning(
                "sync_context_manager_not_supported",
                class_name=type_.__qualname__,
                scope=scope.name,
            )

    edges = _collect_dep_edges(type_.__init__, type_)
    self._dep_edges.extend(edges)

    entry = ProviderEntry(
        type_=type_,
        scope=scope,
        kind="class",
        impl=type_,
        factory_shape=FactoryShape.CLASS,
        lifecycle=resolved_lifecycle,
    )
    self._providers[type_] = entry
    logger.debug(
        "provider-registered",
        type_name=type_.__qualname__,
        scope=scope.name,
        kind="class",
    )

has_provider 

has_provider(type_: type) -> bool
Source code in src/taskq/_di/registry.py 
202
203
def has_provider(self, type_: type) -> bool:
    return type_ in self._providers

get 

get(type_: type[T]) -> ProviderEntry[T]
Source code in src/taskq/_di/registry.py 
205
206
207
208
209
210
211
212
213
214
def get[T](self, type_: type[T]) -> ProviderEntry[T]:
    entry = self._providers.get(type_)
    if entry is None:
        raise MissingProvider(
            type_name=type_.__qualname__,
            required_by="<unknown>",
        )
    return cast(
        ProviderEntry[T], entry
    )  # Why: internal map is dict[type, ProviderEntry[object]] (erasure boundary); get[T] recovers the type parameter

validate 

validate(
    actors: list[ActorRef[Any, Any]] | None = None,
    rate_limit_registry: RateLimitRegistry | None = None,
) -> None

Walk all providers and actors; raise on first error; seal on success.

Pure graph-walk over registration-time metadata. Never invokes a factory, calls a resolver, or performs await — the entire algorithm is synchronous introspection on _providers and _dep_edges.

actors: explicit list of ActorRef instances whose DI parameter annotations are walked for MissingProvider checks and plan-cache population. Defaults to None (no actor walk — only provider→provider edges are validated).

rate_limit_registry: when provided, each actor's rate_limits and reservations lists are checked against the registry's dicts. Unknown names raise MissingProvider at startup. When None (default), the name-check phase is skipped entirely.

Why: ActorRef[Any, Any] is the sanctioned erasure
boundary for the heterogeneous actor registry — each ActorRef has
different P and R type parameters, and validate() does not need
per-actor narrowing. The same erasure is used at the existing API
boundary in worker/run.py:283 and cli.py:98.
Source code in src/taskq/_di/registry.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def validate(
    self,
    actors: list["ActorRef[Any, Any]"] | None = None,
    rate_limit_registry: "RateLimitRegistry | None" = None,
) -> None:
    """Walk all providers and actors; raise on first error; seal on success.

    Pure graph-walk over registration-time metadata. Never invokes a
    factory, calls a resolver, or performs await — the entire algorithm
    is synchronous introspection on ``_providers`` and ``_dep_edges``.

    actors: explicit list of ActorRef instances whose DI parameter
    annotations are walked for MissingProvider checks and plan-cache
    population. Defaults to None (no actor walk — only
    provider→provider edges are validated).

    rate_limit_registry: when provided, each actor's ``rate_limits``
    and ``reservations`` lists are checked against the registry's
    dicts. Unknown names raise ``MissingProvider`` at startup.
    When ``None`` (default), the name-check phase is
    skipped entirely.

    # Why: ActorRef[Any, Any] is the sanctioned erasure
    # boundary for the heterogeneous actor registry — each ActorRef has
    # different P and R type parameters, and validate() does not need
    # per-actor narrowing. The same erasure is used at the existing API
    # boundary in worker/run.py:283 and cli.py:98.
    """
    if self._validated:
        return
    if self._validating:
        raise RuntimeError("validate() called recursively or concurrently")
    self._validating = True
    try:
        self._plan_cache = run_validation(
            self._providers,
            self._dep_edges,
            actors,
            rate_limit_registry,
        )
        if actors is not None:
            _emit_redundant_override_warnings(actors, self._providers)
        self._validated = True
        self._sealed = True
        logger.info(
            "registry-validated",
            provider_count=len(self._providers),
            actor_count=len(actors) if actors else 0,
        )
    finally:
        self._validating = False

scopes

Concrete ScopeContainer with LIFO teardown and factory-shape dispatch.

Implements the log-and-continue teardown policy: each teardown callback runs in its own try/except; failures are logged at ERROR; remaining teardowns always fire. A parallel _teardowns list replaces AsyncExitStack.aclose() which re-raises the first exception and swallows the rest (research line 743-758).

logger module-attribute 

logger = structlog.get_logger('taskq._di.scopes')

ScopeContainer 

ScopeContainer(*, scope: Scope, resolver: _Resolver)

Concrete scope-lifetime container owning a cache, teardown list, and AsyncExitStack.

The container is responsible for ALL factory invocation, caching, and teardown registration. The solver engine NEVER calls a factory directly and NEVER touches an AsyncExitStack.

Source code in src/taskq/_di/scopes.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def __init__(
    self,
    *,
    scope: Scope,
    resolver: _Resolver,
) -> None:
    self._scope: Scope = scope
    self._cache: dict[type, object] = {}
    self._stack: AsyncExitStack = AsyncExitStack()
    self._teardowns: list[Callable[[], Any]] = []
    self._resolver: _Resolver = resolver
    self._sync_gen_executor: ThreadPoolExecutor | None = None
    self._last_cache_hit: bool = False

last_cache_hit property 

last_cache_hit: bool

Whether the most recent get_or_create returned a cached value.

get_or_create async 

get_or_create(type_: type[T], entry: ProviderEntry[T]) -> T

Resolve type_ via entry, creating and caching the instance if needed.

Source code in src/taskq/_di/scopes.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def get_or_create[T](self, type_: type[T], entry: ProviderEntry[T]) -> T:
    """Resolve *type_* via *entry*, creating and caching the instance if needed."""
    if self._scope is not Scope.TRANSIENT:
        cached = self._cache.get(type_)
        if cached is not None:
            self._last_cache_hit = True
            return cached  # type: ignore[return-value]  # Why: _cache is dict[type, object]; caller's T is recovered via the ProviderEntry[T] that selected this branch
    self._last_cache_hit = False

    match entry.factory_shape:
        case FactoryShape.VALUE:
            result: object = entry.impl
        case FactoryShape.SYNC_CALLABLE:
            kwargs = await self._resolver(entry.impl)
            result = cast(Callable[..., Any], entry.impl)(**kwargs)
        case FactoryShape.ASYNC_CALLABLE:
            kwargs = await self._resolver(entry.impl)
            result = await cast(Callable[..., Any], entry.impl)(**kwargs)
        case FactoryShape.SYNC_GENERATOR:
            result = await self._resolve_sync_generator(entry)
        case FactoryShape.ASYNC_GENERATOR:
            result = await self._resolve_async_generator(entry)
        case FactoryShape.CLASS:
            kwargs = await self._resolver(cast(type[Any], entry.impl).__init__)
            instance = cast(type[Any], entry.impl)(**kwargs)
            match entry.lifecycle:
                case ProviderLifecycle.AsyncContextManager:
                    value = await instance.__aenter__()

                    async def _acm_teardown() -> None:
                        await instance.__aexit__(None, None, None)

                    self._teardowns.append(_acm_teardown)
                    result = value
                case ProviderLifecycle.AsyncCloseable:
                    self._teardowns.append(instance.aclose)
                    result = instance
                case ProviderLifecycle.SyncCloseable:
                    self._teardowns.append(lambda inst=instance: asyncio.to_thread(inst.close))
                    result = instance
                case ProviderLifecycle.Plain | None:
                    result = instance
                case (
                    ProviderLifecycle.AsyncGenerator
                    | ProviderLifecycle.SyncGenerator
                    | ProviderLifecycle.PlainFactory
                ):
                    msg = f"factory lifecycle {entry.lifecycle!r} reached CLASS arm"
                    raise RuntimeError(msg)
                case _:
                    assert_never(entry.lifecycle)
        case _:
            assert_never(entry.factory_shape)

    if self._scope is not Scope.TRANSIENT:
        self._cache[type_] = result

    return result  # type: ignore[return-value]  # Why: same recovery as cache-hit branch — erasure boundary documented

aclose async 

aclose() -> None

Close the container with the log-and-continue teardown policy.

Source code in src/taskq/_di/scopes.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
async def aclose(self) -> None:
    """Close the container with the log-and-continue teardown policy."""
    pending_cancel: BaseException | None = None
    while self._teardowns:
        cb = self._teardowns.pop()
        try:
            await cb()
        except BaseException as exc:
            logger.error(
                "provider-teardown-error",
                scope=self._scope.name,
                exc_info=True,
            )
            # Why: remember a CancelledError so we can re-raise it after
            # all remaining teardowns have run; non-cancel exceptions
            # are logged and dropped  log-and-continue.
            if isinstance(exc, asyncio.CancelledError):
                pending_cancel = exc

    # Why: shut down the pinned SYNC_GENERATOR executor AFTER all
    # per-provider teardown callbacks have run. shutdown(wait=True) is
    # blocking; running it inline would block the loop .
    if self._sync_gen_executor is not None:
        executor = self._sync_gen_executor
        self._sync_gen_executor = None
        try:
            await asyncio.to_thread(executor.shutdown, True)
        except BaseException as exc:
            # Why: parallels the per-callback BaseException pattern above.
            logger.error(
                "sync-generator-executor-shutdown-error",
                scope=self._scope.name,
                exc_info=True,
            )
            if isinstance(exc, asyncio.CancelledError):
                pending_cancel = exc
        else:
            logger.info(
                "sync_generator_executor_shutdown",
                scope=self._scope.name,
            )

    # Why: after every teardown attempted, re-raise any deferred
    # CancelledError so the outer task's cancellation contract is honored.
    if pending_cancel is not None:
        raise pending_cancel

ProcessScope 

ProcessScope(*, resolver: _Resolver)

Bases: ScopeContainer

PROCESS-lifetime scope — worker process startup → exit.

Source code in src/taskq/_di/scopes.py 
232
233
def __init__(self, *, resolver: _Resolver) -> None:
    super().__init__(scope=Scope.PROCESS, resolver=resolver)

bootstrap async 

bootstrap(
    registry: ProviderRegistry, settings: WorkerSettings
) -> None

Resolve all PROCESS-scoped providers.

Why: no try/except around get_or_create — earlier providers' teardowns are already registered on self._teardowns; the caller's AsyncExitStack runs aclose() on unwind, which iterates whatever teardowns were registered before the failure. Partial-bootstrap leaks are not possible: every successful get_or_create has its teardown queued before the next get_or_create starts.

Source code in src/taskq/_di/scopes.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
async def bootstrap(
    self,
    registry: ProviderRegistry,
    settings: WorkerSettings,  # Why: accepted for API symmetry with the public surface; bootstrap doesn't use it directly — registration is a separate concern owned by worker bootstrap
) -> None:
    """Resolve all PROCESS-scoped providers.

    Why: no try/except around get_or_create — earlier providers'
    teardowns are already registered on self._teardowns; the
    caller's AsyncExitStack runs aclose() on unwind, which
    iterates whatever teardowns were registered before the
    failure. Partial-bootstrap leaks are not possible: every
    successful get_or_create has its teardown queued before
    the next get_or_create starts.
    """
    process_providers = [t for t, e in registry.providers.items() if e.scope is Scope.PROCESS]
    for t in process_providers:
        await self.get_or_create(t, registry.get(t))  # type: ignore[reportUnknownArgumentType]  # Why: registry.get() returns ProviderEntry[Unknown] when called with plain type; runtime guarantee from providers dict iteration
    logger.info(
        "process-scope-opened",
        provider_count=len(process_providers),
    )

shutdown async 

shutdown() -> None
Source code in src/taskq/_di/scopes.py 
258
259
260
async def shutdown(self) -> None:
    await self.aclose()
    logger.info("process-scope-closed")

get 

get(type_: type) -> object | None
Source code in src/taskq/_di/scopes.py 
262
263
def get(self, type_: type) -> object | None:
    return self._cache.get(type_)

ThreadScope 

ThreadScope(*, resolver: _Resolver)

Bases: ScopeContainer

THREAD-lifetime scope — placeholder for multi-thread workers (trivially empty in M3).

Source code in src/taskq/_di/scopes.py 
269
270
def __init__(self, *, resolver: _Resolver) -> None:
    super().__init__(scope=Scope.THREAD, resolver=resolver)

bootstrap async 

bootstrap(
    registry: ProviderRegistry, process_scope: ProcessScope
) -> None

Resolve all THREAD-scoped providers (empty in M3 single-thread deployment).

Source code in src/taskq/_di/scopes.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
async def bootstrap(
    self,
    registry: ProviderRegistry,
    process_scope: ProcessScope,
) -> None:
    """Resolve all THREAD-scoped providers (empty in M3 single-thread deployment)."""
    thread_providers = [t for t, e in registry.providers.items() if e.scope is Scope.THREAD]
    for t in thread_providers:
        await self.get_or_create(t, registry.get(t))  # type: ignore[reportUnknownArgumentType]  # Why: registry.get() returns ProviderEntry[Unknown] when called with plain type; runtime guarantee from providers dict iteration
    logger.info(
        "thread-scope-opened",
        provider_count=len(thread_providers),
    )

shutdown async 

shutdown() -> None
Source code in src/taskq/_di/scopes.py 
286
287
288
async def shutdown(self) -> None:
    await self.aclose()
    logger.info("thread-scope-closed")

get 

get(type_: type) -> object | None
Source code in src/taskq/_di/scopes.py 
290
291
def get(self, type_: type) -> object | None:
    return self._cache.get(type_)

LoopScope 

LoopScope(*, resolver: _Resolver)

Bases: ScopeContainer

LOOP-lifetime scope — worker loop start → loop close.

Source code in src/taskq/_di/scopes.py 
297
298
299
def __init__(self, *, resolver: _Resolver) -> None:
    super().__init__(scope=Scope.LOOP, resolver=resolver)
    self._loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()

bootstrap async 

bootstrap(
    registry: ProviderRegistry,
    process_scope: ProcessScope,
    thread_scope: ThreadScope | None = None,
) -> None

Resolve all LOOP-scoped providers.

Why: process_scope / thread_scope parameters are accepted for API completeness even though bootstrap doesn't read them directly — wider scopes' instances are reachable through the _resolver callable (which holds the registry + container map).

Source code in src/taskq/_di/scopes.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
async def bootstrap(
    self,
    registry: ProviderRegistry,
    process_scope: ProcessScope,
    thread_scope: ThreadScope | None = None,
) -> None:
    """Resolve all LOOP-scoped providers.

    Why: process_scope / thread_scope parameters are accepted for
    API completeness even though bootstrap doesn't read them
    directly — wider scopes' instances are reachable through the
    _resolver callable (which holds the registry + container map).
    """
    loop_providers = [t for t, e in registry.providers.items() if e.scope is Scope.LOOP]
    for t in loop_providers:
        await self.get_or_create(t, registry.get(t))  # type: ignore[reportUnknownArgumentType]  # Why: registry.get() returns ProviderEntry[Unknown] when called with plain type; runtime guarantee from providers dict iteration
    logger.info(
        "loop-scope-opened",
        provider_count=len(loop_providers),
    )

shutdown async 

shutdown() -> None
Source code in src/taskq/_di/scopes.py 
322
323
324
async def shutdown(self) -> None:
    await self.aclose()
    logger.info("loop-scope-closed")

resolved_cache 

resolved_cache() -> Mapping[type, object]

Return a read-only view of the LOOP-scope resolved values.

The returned mapping is a live view onto the underlying cache — providers resolved after this method is called (e.g., lazy providers) become visible without re-fetching, and replacements made via :meth:replace_value become visible to subsequent .get() reads. Callers MUST NOT attempt to mutate the mapping; doing so raises TypeError.

Stability invariant: once bootstrap returns, the registered set of LOOP-scope provider types is FROZEN for the lifetime of the loop. asyncpg.Connection values are stable references — callers that read the mapping repeatedly (such as SubJobEnqueuer.enqueue) assume the connection identity does not change between dispatches, because the consumer's per-dispatch transaction lifecycle requires the same connection across both transaction-open and transaction-close.

asyncpg.Pool values are the one sanctioned exception: they may be replaced mid-loop via :meth:replace_value (credential hot-reload — see taskq.worker.deps.reload_credentials). Pools are stateless acquisition factories, so swapping the reference does not violate any transaction-identity invariant; consumers that read the mapping per dispatch (rather than capturing it once) always see the current pool.

Source code in src/taskq/_di/scopes.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
def resolved_cache(self) -> Mapping[type, object]:
    """Return a read-only view of the LOOP-scope resolved values.

    The returned mapping is a live view onto the underlying cache —
    providers resolved after this method is called (e.g., lazy
    providers) become visible without re-fetching, and replacements
    made via :meth:`replace_value` become visible to subsequent
    ``.get()`` reads. Callers MUST NOT attempt to mutate the mapping;
    doing so raises ``TypeError``.

    Stability invariant: once ``bootstrap`` returns, the registered
    set of LOOP-scope provider types is FROZEN for the lifetime of
    the loop. ``asyncpg.Connection`` values are stable references —
    callers that read the mapping repeatedly (such as
    ``SubJobEnqueuer.enqueue``) assume the connection identity does
    not change between dispatches, because the consumer's
    per-dispatch transaction lifecycle requires the same connection
    across both transaction-open and transaction-close.

    ``asyncpg.Pool`` values are the one sanctioned exception: they
    may be replaced mid-loop via :meth:`replace_value` (credential
    hot-reload — see ``taskq.worker.deps.reload_credentials``).
    Pools are stateless acquisition factories, so swapping the
    reference does not violate any transaction-identity invariant;
    consumers that read the mapping per dispatch (rather than
    capturing it once) always see the current pool.
    """
    return MappingProxyType(self._cache)

replace_value 

replace_value(type_: type, value: object) -> None

Replace the cached instance for type_.

The sanctioned mid-loop swap for VALUE providers whose resource was hot-swapped externally — currently asyncpg.Pool after a SIGHUP credential reload (taskq.worker.deps.reload_credentials drains and closes the old pool; without this, DI-injected db: asyncpg.Pool consumers would keep using a closed pool).

The old value is NOT torn down here — its lifecycle belongs to the swapper. Only the cache entry is replaced; the (sealed) registry is untouched. Raises :class:KeyError when type_ has no cached value — replacing an absent entry would silently insert a value no provider declared.

Source code in src/taskq/_di/scopes.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def replace_value(self, type_: type, value: object) -> None:
    """Replace the cached instance for *type_*.

    The sanctioned mid-loop swap for VALUE providers whose resource
    was hot-swapped externally — currently ``asyncpg.Pool`` after a
    SIGHUP credential reload (``taskq.worker.deps.reload_credentials``
    drains and closes the old pool; without this, DI-injected
    ``db: asyncpg.Pool`` consumers would keep using a closed pool).

    The old value is NOT torn down here — its lifecycle belongs to
    the swapper. Only the cache entry is replaced; the (sealed)
    registry is untouched. Raises :class:`KeyError` when *type_* has
    no cached value — replacing an absent entry would silently
    insert a value no provider declared.
    """
    if type_ not in self._cache:
        raise KeyError(f"no cached value for {type_!r} — nothing to replace")
    self._cache[type_] = value

get_or_create async 

get_or_create(type_: type[T], entry: ProviderEntry[T]) -> T
Source code in src/taskq/_di/scopes.py 
374
375
376
377
378
async def get_or_create[T](self, type_: type[T], entry: ProviderEntry[T]) -> T:
    running = asyncio.get_running_loop()
    if running is not self._loop:
        raise RuntimeError(f"LoopScope created on {self._loop!r} but accessed from {running!r}")
    return await super().get_or_create(type_, entry)

get 

get(type_: type) -> object | None
Source code in src/taskq/_di/scopes.py 
380
381
def get(self, type_: type) -> object | None:
    return self._cache.get(type_)

ResolvedActorScope dataclass 

ResolvedActorScope(
    ctx: JobContext[BaseModel], di_kwargs: dict[str, object]
)

The yielded value of build_actor_scope.

ctx — the JobContext supplied by the consumer (passthrough). di_kwargs — the DI-resolved kwargs dict for the actor function.

Usage::

async with build_actor_scope(...) as resolved:
    await run_actor(job, resolved.ctx, **resolved.di_kwargs)

ctx instance-attribute 

ctx: JobContext[BaseModel]

di_kwargs instance-attribute 

di_kwargs: dict[str, object]

make_resolver 

make_resolver(
    registry: ProviderRegistry,
    scope_containers: dict[Scope, ScopeContainer],
) -> _Resolver

Construct the resolver callable expected by ScopeContainer.init.

The resolver invokes solve_dependencies with the registry and scope containers. The closure captures scope_containers by reference so the resolver sees containers added after its construction (the dict is populated incrementally as each scope container is bootstrapped).

Source code in src/taskq/_di/scopes.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def make_resolver(
    registry: ProviderRegistry,
    scope_containers: dict[Scope, ScopeContainerProtocol],
) -> _Resolver:
    """Construct the resolver callable expected by ScopeContainer.__init__.

    The resolver invokes solve_dependencies with the registry and scope
    containers. The closure captures scope_containers by reference so the
    resolver sees containers added after its construction (the dict is
    populated incrementally as each scope container is bootstrapped).
    """

    def _resolver(func: object) -> Any:
        async def _resolve() -> dict[str, object]:
            return await solve_dependencies(
                func=func,
                registry=registry,
                scope_containers=scope_containers,
            )

        return _resolve()

    return _resolver

build_actor_scope async 

build_actor_scope(
    *,
    registry: ProviderRegistry,
    process_scope: ProcessScope,
    thread_scope: ThreadScope,
    loop_scope: LoopScope,
    actor_func: Callable[..., Awaitable[object]],
    actor_name: str,
    passthrough_kwargs: dict[str, object],
) -> AsyncGenerator[ResolvedActorScope, None]

Per-invocation actor scope: opens TRANSIENT stack, resolves DI kwargs.

Yields ResolvedActorScope(ctx, di_kwargs). On exit (regardless of outcome), closes the TRANSIENT stack in LIFO order via the log-and-continue teardown policy.

passthrough_kwargs MUST contain the JobContext (key "ctx") and the validated payload (key "payload"). The consumer constructs both per-job and supplies them as passthrough; the registry's graph walk is configured to skip these parameter names, so they are never resolved from providers.

Source code in src/taskq/_di/scopes.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
@asynccontextmanager
async def build_actor_scope(
    *,
    registry: ProviderRegistry,
    process_scope: ProcessScope,
    thread_scope: ThreadScope,
    loop_scope: LoopScope,
    actor_func: Callable[..., Awaitable[object]],
    actor_name: str,
    passthrough_kwargs: dict[str, object],
) -> AsyncGenerator[ResolvedActorScope, None]:
    """Per-invocation actor scope: opens TRANSIENT stack, resolves DI kwargs.

    Yields ResolvedActorScope(ctx, di_kwargs). On exit (regardless of
    outcome), closes the TRANSIENT stack in LIFO order via the
    log-and-continue teardown policy.

    passthrough_kwargs MUST contain the JobContext (key "ctx") and the
    validated payload (key "payload"). The consumer constructs both
    per-job and supplies them as passthrough; the registry's graph walk
    is configured to skip these parameter names, so they are never
    resolved from providers.
    """
    scope_containers: dict[Scope, ScopeContainerProtocol] = {}

    def _resolver(func: object) -> Any:
        async def _resolve() -> dict[str, object]:
            return await solve_dependencies(
                func=func,
                registry=registry,
                scope_containers=scope_containers,
            )

        return _resolve()

    transient_scope = ScopeContainer(scope=Scope.TRANSIENT, resolver=_resolver)

    scope_containers = {
        Scope.PROCESS: process_scope,
        Scope.THREAD: thread_scope,
        Scope.LOOP: loop_scope,
        Scope.TRANSIENT: transient_scope,
    }

    # Why: the resolver closure captured scope_containers before TRANSIENT
    # was added; re-bind so the resolver sees all four containers.
    def _resolver_with_all(func: object) -> Any:
        async def _resolve() -> dict[str, object]:
            return await solve_dependencies(
                func=func,
                registry=registry,
                scope_containers=scope_containers,
            )

        return _resolve()

    transient_scope._resolver = _resolver_with_all  # pyright: ignore[reportPrivateUsage]  # Why: build_actor_scope constructs the TRANSIENT container and must wire its resolver to see all four scope containers; the resolver is a closure detail owned by this call site

    logger.info("transient-scope-opened", actor_name=actor_name)
    try:
        di_kwargs = await solve_dependencies(
            func=actor_func,
            registry=registry,
            scope_containers=scope_containers,
            passthrough_kwargs=passthrough_kwargs,
        )
        # Why: solve_dependencies includes passthrough keys in its result
        # dict; build_actor_scope yields ctx separately and the consumer
        # spreads di_kwargs — so passthrough keys must not appear in
        # di_kwargs to avoid duplicate keyword arguments at the call site.
        for _key in passthrough_kwargs:
            di_kwargs.pop(_key, None)
        ctx = passthrough_kwargs["ctx"]
        if not isinstance(ctx, JobContext):
            raise TypeError(
                f"passthrough_kwargs['ctx'] must be a JobContext, got {type(ctx).__name__}"
            )
        yield ResolvedActorScope(ctx=ctx, di_kwargs=di_kwargs)  # type: ignore[reportUnknownArgumentType]  # Why: ctx is narrowed to JobContext[Any] by isinstance but pyright cannot recover the BaseModel bound from the passthrough_kwargs dict[str, object] — the consumer that built passthrough_kwargs guarantees the correct P at the call site
    finally:
        # Why: shield the TRANSIENT teardown so cancellation /
        # asyncio.wait_for timeouts in the with-body do not
        # short-circuit the teardown mid-way.
        # ("wrap terminal writes in asyncio.shield") applies to
        # scope teardown too — losing teardown of an opened
        # resource leaks it. CancelledError after the shielded
        # aclose finishes is re-raised to honor the outer cancel.
        try:
            await asyncio.shield(transient_scope.aclose())
        except asyncio.CancelledError:
            raise
        finally:
            logger.info("transient-scope-closed", actor_name=actor_name)