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 
53
54
55
56
57
58
59
60
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
    self._shadow_derived_cache: dict[frozenset[type], frozenset[type]] = {}

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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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 
 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
132
133
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 
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
201
202
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 
204
205
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 
207
208
209
210
211
212
213
214
215
216
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

shadow_derived_providers 

shadow_derived_providers(
    shadow_types: frozenset[type],
) -> frozenset[type]

Every non-value provider whose dependency closure reaches one of shadow_types (see :func:taskq._di._shadow.shadow_derived_providers).

Memoized per shadow_types once the registry is sealed: the closure is a pure function of the (then immutable) provider graph, and the per-slot dispatch path asks for it on every job. Before the seal a registration can still change the answer, so the walk runs uncached. The memo is keyed by the worker's own slot wiring (one key per distinct shadowed-type set), never by user input, so it needs no bound.

Source code in src/taskq/_di/registry.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def shadow_derived_providers(self, shadow_types: frozenset[type]) -> frozenset[type]:
    """Every non-value provider whose dependency closure reaches one of
    *shadow_types* (see :func:`taskq._di._shadow.shadow_derived_providers`).

    Memoized per *shadow_types* once the registry is sealed: the
    closure is a pure function of the (then immutable) provider graph,
    and the per-slot dispatch path asks for it on every job. Before
    the seal a registration can still change the answer, so the walk
    runs uncached. The memo is keyed by the worker's own slot wiring
    (one key per distinct shadowed-type set), never by user input, so
    it needs no bound.
    """
    if not self._sealed:
        return shadow_derived_providers(self._providers, shadow_types)
    derived = self._shadow_derived_cache.get(shadow_types)
    if derived is None:
        derived = shadow_derived_providers(self._providers, shadow_types)
        self._shadow_derived_cache[shadow_types] = derived
    return derived

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 
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
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
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 
53
54
55
56
57
58
59
60
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
    self._shadow_derived_cache: dict[frozenset[type], frozenset[type]] = {}

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 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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 
 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
132
133
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 
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
201
202
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 
204
205
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 
207
208
209
210
211
212
213
214
215
216
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

shadow_derived_providers 

shadow_derived_providers(
    shadow_types: frozenset[type],
) -> frozenset[type]

Every non-value provider whose dependency closure reaches one of shadow_types (see :func:taskq._di._shadow.shadow_derived_providers).

Memoized per shadow_types once the registry is sealed: the closure is a pure function of the (then immutable) provider graph, and the per-slot dispatch path asks for it on every job. Before the seal a registration can still change the answer, so the walk runs uncached. The memo is keyed by the worker's own slot wiring (one key per distinct shadowed-type set), never by user input, so it needs no bound.

Source code in src/taskq/_di/registry.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def shadow_derived_providers(self, shadow_types: frozenset[type]) -> frozenset[type]:
    """Every non-value provider whose dependency closure reaches one of
    *shadow_types* (see :func:`taskq._di._shadow.shadow_derived_providers`).

    Memoized per *shadow_types* once the registry is sealed: the
    closure is a pure function of the (then immutable) provider graph,
    and the per-slot dispatch path asks for it on every job. Before
    the seal a registration can still change the answer, so the walk
    runs uncached. The memo is keyed by the worker's own slot wiring
    (one key per distinct shadowed-type set), never by user input, so
    it needs no bound.
    """
    if not self._sealed:
        return shadow_derived_providers(self._providers, shadow_types)
    derived = self._shadow_derived_cache.get(shadow_types)
    if derived is None:
        derived = shadow_derived_providers(self._providers, shadow_types)
        self._shadow_derived_cache[shadow_types] = derived
    return derived

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 
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
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
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. Teardowns are kept in the container's own LIFO list rather than an AsyncExitStack, whose aclose() re-raises the first exception and swallows the rest.

logger module-attribute 

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

ScopeContainer 

ScopeContainer(
    *,
    scope: Scope,
    resolver: _Resolver,
    factory_timeout: float | None = None,
)

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

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

Source code in src/taskq/_di/scopes.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def __init__(
    self,
    *,
    scope: Scope,
    resolver: _Resolver,
    factory_timeout: float | None = None,
) -> None:
    self._scope: Scope = scope
    self._cache: dict[type, object] = {}
    self._teardowns: list[Callable[[], Any]] = []
    self._resolver: _Resolver = resolver
    self._sync_gen_executor: ThreadPoolExecutor | None = None
    self._last_cache_hit: bool = False
    self._factory_timeout: float | None = factory_timeout

last_cache_hit property 

last_cache_hit: bool

Whether the most recent get_or_create returned a cached value.

has_teardown_work property 

has_teardown_work: bool

Whether :meth:aclose has anything left to run.

False once every registered teardown has run and the pinned SYNC_GENERATOR executor (if one was ever opened) is shut down , i.e. exactly when aclose() would return without awaiting anything, which lets a per-job caller skip scheduling it.

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 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
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."""
    try:
        return await self._get_or_create_user_code(type_, entry)
    except (KeyboardInterrupt, asyncio.CancelledError):
        # The carve-out, exactly as the consumer's attempt boundary
        # applies it: interpreter/operator intent (KeyboardInterrupt)
        # and shutdown cancellation (CancelledError) are never a
        # factory outcome; both propagate raw.
        raise
    except SystemExit as exc:
        # The single conversion seam every provider factory crosses:
        # see _ProviderSystemExitError for why the carrier, not the
        # raw SystemExit, is what may leave this frame.
        raise _ProviderSystemExitError(exc) from exc

aclose async 

aclose() -> None

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

Source code in src/taskq/_di/scopes.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
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 .
    #
    # The wait itself is bounded by ``factory_timeout`` when the
    # container has one: a sync-gen ``__enter__``/``__exit__`` parked
    # in unkillable user code would otherwise park scope teardown for
    # as long as the user code hangs. The bound trips loudly and lets
    # the close complete, the shutdown already initiated finishes in
    # the background whenever the user code returns, and the residue
    # is that one thread per hung container until then (a leak the
    # stdlib makes unrecoverable by construction; what teardown owns
    # is never parking the REST of the close on it).
    if self._sync_gen_executor is not None:
        executor = self._sync_gen_executor
        self._sync_gen_executor = None
        try:
            if self._factory_timeout is None:
                await asyncio.to_thread(executor.shutdown, True)
            else:
                async with asyncio.timeout(self._factory_timeout):
                    await asyncio.to_thread(executor.shutdown, True)
        except TimeoutError:
            logger.error(
                "sync-generator-executor-shutdown-timeout",
                kind="sync_generator_executor_shutdown_timeout",
                scope=self._scope.name,
                timeout=self._factory_timeout,
            )
        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,
    factory_timeout: float | None = None,
)

Bases: ScopeContainer

PROCESS-lifetime scope, worker process startup → exit.

Source code in src/taskq/_di/scopes.py 
415
416
417
418
419
420
421
def __init__(
    self,
    *,
    resolver: _Resolver,
    factory_timeout: float | None = None,
) -> None:
    super().__init__(scope=Scope.PROCESS, resolver=resolver, factory_timeout=factory_timeout)

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 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 
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
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 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 
445
446
447
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 
449
450
def get(self, type_: type) -> object | None:
    return self._cache.get(type_)

ThreadScope 

ThreadScope(
    *,
    resolver: _Resolver,
    factory_timeout: float | None = None,
)

Bases: ScopeContainer

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

Source code in src/taskq/_di/scopes.py 
456
457
458
459
460
461
462
def __init__(
    self,
    *,
    resolver: _Resolver,
    factory_timeout: float | None = None,
) -> None:
    super().__init__(scope=Scope.THREAD, resolver=resolver, factory_timeout=factory_timeout)

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 
464
465
466
467
468
469
470
471
472
473
474
475
476
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 
478
479
480
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 
482
483
def get(self, type_: type) -> object | None:
    return self._cache.get(type_)

LoopScope 

LoopScope(
    *,
    resolver: _Resolver,
    factory_timeout: float | None = None,
)

Bases: ScopeContainer

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

Source code in src/taskq/_di/scopes.py 
489
490
491
492
493
494
495
496
def __init__(
    self,
    *,
    resolver: _Resolver,
    factory_timeout: float | None = None,
) -> None:
    super().__init__(scope=Scope.LOOP, resolver=resolver, factory_timeout=factory_timeout)
    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 
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
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 
519
520
521
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 
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
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 
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
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 
571
572
573
574
575
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 
577
578
def get(self, type_: type) -> object | None:
    return self._cache.get(type_)

LoopScopeSlotView 

LoopScopeSlotView(
    inner: LoopScope,
    slot_values: Mapping[type, object],
    *,
    invocation_runner: ScopeContainer,
    shadow_derived: frozenset[type],
)

Per-actor-invocation view of the LOOP scope for one consumer slot.

The worker's per-slot path acquires a dedicated connection per job (a LOOP-scope asyncpg.Connection registered plus max_concurrency > 1); the transaction that wraps the actor runs on that connection. This view shadows the LOOP-shared cache for exactly the types its mapping carries, for exactly one build_actor_scope body, so the actor, and every nested LOOP-scoped resolution its dependencies trigger through the same scope-containers map, resolves the slot's instance instead of the one object registered at Scope.LOOP. Without the shadow, every concurrent slot's actor received the SAME registered connection and interleaved operations on it: asyncpg permits one operation per connection, so healthy actors raised InterfaceError (misattributed to the actor, retry budget burned), and the actor's own writes sat outside its slot's transaction (each job's work must run in isolation within its transaction; sharing a connection across concurrent jobs violates that isolation and causes writes to land outside any job's boundary).

The shadow reaches one level further than the mapping itself: a LOOP-scoped FACTORY whose dependency closure reaches a shadowed type (a helper that injects the connection, "any object derived from a shared connection is shared the same way", the same rule one level removed) is re-resolved PER INVOCATION through the invocation's TRANSIENT runner instead of the LOOP cache. The scope's bootstrap eagerly resolved that factory through the REAL containers, so the cached singleton holds the ONE registered connection; handing it to concurrent slots' actors would rebuild the exact sharing the view exists to prevent. Re-resolution runs the factory's own parameters through the same shadowed scope-containers map (nested LOOP-scoped dependencies included), produces an instance this invocation alone owns, and lands any lifecycle teardown on the invocation's TRANSIENT teardown, the derived object lives and dies with the actor call. Providers whose closure never touches a shadowed type keep the LOOP singleton: the view does not per-invocation-ize unrelated loop-lifetime resources.

The view is a pure read-through otherwise: it never caches, never invokes a factory on its own account, and never registers a teardown, a mapped instance's lifecycle belongs to the wiring that supplied it (the dispatch acquire/release exit stack that owns the slot connection), and a re-resolved instance's to the invocation runner. Every type NOT in the mapping and NOT shadow-derived resolves through the real LOOP container unchanged, and nothing outside this actor invocation can observe the shadow: the LOOP cache itself is never touched.

Mapped values must be live instances; None is not a meaningful mapping (it is treated as absent, exactly like the LOOP cache's own miss shape).

Source code in src/taskq/_di/scopes.py 
634
635
636
637
638
639
640
641
642
643
644
645
646
def __init__(
    self,
    inner: LoopScope,
    slot_values: Mapping[type, object],
    *,
    invocation_runner: ScopeContainer,
    shadow_derived: frozenset[type],
) -> None:
    self._inner = inner
    self._slot_values = slot_values
    self._invocation_runner = invocation_runner
    self._shadow_derived = shadow_derived
    self._last_cache_hit = False

last_cache_hit property 

last_cache_hit: bool

Whether the most recent get_or_create served a mapped or cached value.

get_or_create async 

get_or_create(type_: type[T], entry: ProviderEntry[T]) -> T
Source code in src/taskq/_di/scopes.py 
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
async def get_or_create[T](self, type_: type[T], entry: ProviderEntry[T]) -> T:
    value = self._slot_values.get(type_)
    if value is not None:
        self._last_cache_hit = True
        return cast("T", value)  # pyright: ignore[reportReturnType]  # Why: the DI erasure boundary, the mapping's value is the live instance the caller's wiring owns for type_ (dispatch maps the slot connection under asyncpg.Connection); the same value-shape trust ScopeContainer's cache-hit branch applies.
    if type_ in self._shadow_derived:
        # A LOOP-scoped factory derived (transitively) from a
        # shadowed type: the LOOP cache's bootstrap singleton baked
        # in the registered instance, so resolve this invocation's
        # own through the TRANSIENT runner, fresh parameters off
        # the shadowed containers map, no caching, teardown owned
        # by the invocation.
        self._last_cache_hit = False
        return await self._invocation_runner.get_or_create(type_, entry)  # pyright: ignore[reportReturnType]  # Why: same erasure boundary as the mapped branch, the runner's ProviderEntry[T] is the caller's entry; ScopeContainer's return-site coercion covers T.
    self._last_cache_hit = False
    return await self._inner.get_or_create(type_, entry)

aclose async 

aclose() -> None
Source code in src/taskq/_di/scopes.py 
670
671
672
673
async def aclose(self) -> None:
    # The view owns no resources; the LOOP container's lifecycle is
    # unchanged by one actor invocation having looked through it.
    await self._inner.aclose()

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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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],
    loop_slot_values: Mapping[type, object] | None = None,
) -> 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.

loop_slot_values maps LOOP-registered types to the instances THIS consumer slot owns for this one actor invocation (the dispatch path maps asyncpg.Connection to the job's slot connection). Each mapped type resolves to the slot's instance, through a :class:LoopScopeSlotView that shadows the LOOP cache for this invocation only, actor parameters and nested LOOP-scoped dependencies alike, so a LOOP-registered connection never reaches two concurrent slots' actors, and a LOOP-scoped factory DERIVED from a shadowed type (a helper holding the connection) resolves per invocation instead of serving the bootstrap singleton that baked the registered connection in. None /empty keeps the plain LOOP container: the single-slot worker, whose transaction connection IS the registered connection.

Source code in src/taskq/_di/scopes.py 
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
@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],
    loop_slot_values: Mapping[type, object] | None = None,
) -> 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.

    loop_slot_values maps LOOP-registered types to the instances THIS
    consumer slot owns for this one actor invocation (the dispatch path
    maps ``asyncpg.Connection`` to the job's slot connection). Each
    mapped type resolves to the slot's instance, through a
    :class:`LoopScopeSlotView` that shadows the LOOP cache for this
    invocation only, actor parameters and nested LOOP-scoped
    dependencies alike, so a LOOP-registered connection never reaches
    two concurrent slots' actors, and a LOOP-scoped
    factory DERIVED from a shadowed type (a helper holding the
    connection) resolves per invocation instead of serving the
    bootstrap singleton that baked the registered connection in. ``None``
    /empty keeps the plain LOOP container: the single-slot worker, whose
    transaction connection IS the registered connection.
    """
    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)

    # Why a per-invocation view instead of mutating the LOOP cache: the
    # LOOP cache is shared by every concurrent slot, a swap would race
    # sibling dispatches and leak the slot's connection to later jobs.
    # The view is visible only to the solves that run inside THIS
    # build_actor_scope body, which is exactly the audience the slot's
    # connection is safe for. The invocation runner is the body's own
    # TRANSIENT container: a shadow-derived LOOP factory resolves its
    # parameters through the same containers map (LOOP is this view, so
    # its connection-typed parameters see the slot's instance) and its
    # lifecycle teardowns land on the invocation's teardown, the
    # derived object lives and dies with this one actor call, never
    # cached where a sibling slot could reach it.
    loop_container: LoopScope | LoopScopeSlotView = loop_scope
    if loop_slot_values:
        loop_container = LoopScopeSlotView(
            loop_scope,
            loop_slot_values,
            invocation_runner=transient_scope,
            shadow_derived=registry.shadow_derived_providers(frozenset(loop_slot_values)),
        )

    scope_containers = {
        Scope.PROCESS: process_scope,
        Scope.THREAD: thread_scope,
        Scope.LOOP: loop_container,
        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

    # DEBUG, not INFO: this pair fires once per job carrying only the
    # actor name, so at the default level it is a per-job rendering cost
    # for a line nothing consumes.
    logger.debug("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:
            # shield_with_retrieval, not plain asyncio.shield: a detached
            # teardown that fails under a double cancel must have its
            # outcome retrieved and logged, not lost (see taskq._shield).
            # Skipped outright when the container has nothing to close:
            # most jobs resolve no teardown-bearing provider, and the
            # shield's task creation would be the whole cost.
            if transient_scope.has_teardown_work:
                await shield_with_retrieval(transient_scope.aclose())
        except asyncio.CancelledError:
            raise
        finally:
            logger.debug("transient-scope-closed", actor_name=actor_name)