Skip to content

Dependency Injection

Depends

Depends

Marker for dependency injection.

Usage

async def get_db() -> Session: return Session()

@broker.subscriber(queue="orders") async def handle( order: Order, db: Annotated[Session, Depends(get_db)], ) -> None: ...

Per-message lifetime: a fresh dependency graph is resolved for each message. A generator (or async generator) dependency is supported too — yield the value; the code after yield runs as teardown once the handler completes (see :class:~rabbitkit.di.resolver.DependencyScope).

Source code in src/rabbitkit/di/depends.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class Depends:
    """Marker for dependency injection.

    Usage:
        async def get_db() -> Session:
            return Session()

        @broker.subscriber(queue="orders")
        async def handle(
            order: Order,
            db: Annotated[Session, Depends(get_db)],
        ) -> None:
            ...

    Per-message lifetime: a fresh dependency graph is resolved for each
    message. A generator (or async generator) dependency is supported too —
    yield the value; the code after ``yield`` runs as teardown once the
    handler completes (see :class:`~rabbitkit.di.resolver.DependencyScope`).
    """

    def __init__(self, dependency: Callable[..., Any], *, use_cache: bool = True) -> None:
        self.dependency = dependency
        self.use_cache = use_cache

    def __repr__(self) -> str:
        return f"Depends({self.dependency.__qualname__}, use_cache={self.use_cache})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Depends):
            return NotImplemented
        return self.dependency is other.dependency and self.use_cache == other.use_cache

    def __hash__(self) -> int:
        return hash((id(self.dependency), self.use_cache))

Header / Path / Context

Header

Marker for AMQP header extraction.

Usage

@broker.subscriber(queue="orders") async def handle( order: Order, tenant: Annotated[str, Header("x-tenant")], ) -> None: ...

H10 — optional values: pass default= to make a missing header resolve to that value instead of raising, or simply give the parameter itself a Python default (tenant: Annotated[str | None, Header("x-tenant")] = None) — the resolver falls back to the parameter default when the marker has none. With neither, a missing header raises MissingDependencyError (PERMANENT — settles straight to the DLQ, matching the KeyError this replaces).

Source code in src/rabbitkit/di/context.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class Header:
    """Marker for AMQP header extraction.

    Usage:
        @broker.subscriber(queue="orders")
        async def handle(
            order: Order,
            tenant: Annotated[str, Header("x-tenant")],
        ) -> None:
            ...

    H10 — optional values: pass ``default=`` to make a missing header
    resolve to that value instead of raising, or simply give the parameter
    itself a Python default (``tenant: Annotated[str | None,
    Header("x-tenant")] = None``) — the resolver falls back to the parameter
    default when the marker has none. With neither, a missing header raises
    ``MissingDependencyError`` (PERMANENT — settles straight to the DLQ,
    matching the ``KeyError`` this replaces).
    """

    def __init__(self, name: str, *, default: Any = _MISSING) -> None:
        self.name = name
        self.default = default

    @property
    def has_default(self) -> bool:
        return self.default is not _MISSING

    def __repr__(self) -> str:
        return f"Header({self.name!r})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Header):
            return NotImplemented
        return self.name == other.name and self.default == other.default

    def __hash__(self) -> int:
        return hash(("Header", self.name))

Path

Marker for topic wildcard segment extraction.

Usage

@broker.subscriber(queue="events", routing_key="events.*.#") async def handle( event: Event, level: Annotated[str, Path("level")], ) -> None: ...

H10 — optional values: pass default= to make a missing path segment resolve to that value instead of raising, or simply give the parameter itself a Python default (level: Annotated[str | None, Path("level")] = None) — the resolver falls back to the parameter default when the marker has none. With neither, a missing segment raises MissingDependencyError (PERMANENT — settles straight to the DLQ, matching the KeyError this replaces).

Source code in src/rabbitkit/di/context.py 
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
class Path:
    """Marker for topic wildcard segment extraction.

    Usage:
        @broker.subscriber(queue="events", routing_key="events.*.#")
        async def handle(
            event: Event,
            level: Annotated[str, Path("level")],
        ) -> None:
            ...

    H10 — optional values: pass ``default=`` to make a missing path segment
    resolve to that value instead of raising, or simply give the parameter
    itself a Python default (``level: Annotated[str | None, Path("level")]
    = None``) — the resolver falls back to the parameter default when the
    marker has none. With neither, a missing segment raises
    ``MissingDependencyError`` (PERMANENT — settles straight to the DLQ,
    matching the ``KeyError`` this replaces).
    """

    def __init__(self, segment: str, *, default: Any = _MISSING) -> None:
        self.segment = segment
        self.default = default

    @property
    def has_default(self) -> bool:
        return self.default is not _MISSING

    def __repr__(self) -> str:
        return f"Path({self.segment!r})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Path):
            return NotImplemented
        return self.segment == other.segment and self.default == other.default

    def __hash__(self) -> int:
        return hash(("Path", self.segment))

Context

Marker for context value injection.

Usage

@broker.subscriber(queue="orders") async def handle( order: Order, app_name: Annotated[str, Context("app")], ) -> None: ...

H10 — optional values: pass default= to make a missing context key resolve to that value instead of raising, or simply give the parameter itself a Python default (app_name: Annotated[str | None, Context("app")] = None) — the resolver falls back to the parameter default when the marker has none. With neither, a missing key raises MissingDependencyError (PERMANENT — settles straight to the DLQ, matching the KeyError this replaces).

Source code in src/rabbitkit/di/context.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class Context:
    """Marker for context value injection.

    Usage:
        @broker.subscriber(queue="orders")
        async def handle(
            order: Order,
            app_name: Annotated[str, Context("app")],
        ) -> None:
            ...

    H10 — optional values: pass ``default=`` to make a missing context key
    resolve to that value instead of raising, or simply give the parameter
    itself a Python default (``app_name: Annotated[str | None, Context("app")]
    = None``) — the resolver falls back to the parameter default when the
    marker has none. With neither, a missing key raises
    ``MissingDependencyError`` (PERMANENT — settles straight to the DLQ,
    matching the ``KeyError`` this replaces).
    """

    def __init__(self, key: str, *, default: Any = _MISSING) -> None:
        self.key = key
        self.default = default

    @property
    def has_default(self) -> bool:
        return self.default is not _MISSING

    def __repr__(self) -> str:
        return f"Context({self.key!r})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Context):
            return NotImplemented
        return self.key == other.key and self.default == other.default

    def __hash__(self) -> int:
        return hash(("Context", self.key))

ContextRepo

ContextRepo

Context repository for global and per-request values.

Global values are shared across all messages (thread-safe via a lock). Local values use contextvars.ContextVar for correct isolation across both sync threads AND async coroutines on the same event loop — threading.local() would bleed context between concurrent coroutines sharing one OS thread in an async transport.

Source code in src/rabbitkit/di/context.py 
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
class ContextRepo:
    """Context repository for global and per-request values.

    Global values are shared across all messages (thread-safe via a lock).
    Local values use ``contextvars.ContextVar`` for correct isolation across
    both sync threads AND async coroutines on the same event loop —
    ``threading.local()`` would bleed context between concurrent coroutines
    sharing one OS thread in an async transport.
    """

    def __init__(self) -> None:
        self._global: dict[str, Any] = {}
        self._local: contextvars.ContextVar[dict[str, Any]] = contextvars.ContextVar("rabbitkit_local_ctx")

    def set_global(self, key: str, value: Any) -> None:
        """Set a global context value (shared across all messages)."""
        self._global[key] = value

    def set_local(self, key: str, value: Any) -> None:
        """Set a per-request context value.

        Uses ``ContextVar.set`` with an immutable copy so that each
        coroutine/task gets its own isolated snapshot (contextvars
        copy-on-write semantics).
        """
        try:
            current = self._local.get()
        except LookupError:
            current = {}
        self._local.set({**current, key: value})

    def get(self, key: str, default: Any = None) -> Any:
        """Get a context value. Local overrides global."""
        try:
            local = self._local.get()
        except LookupError:
            local = {}
        if key in local:
            return local[key]
        return self._global.get(key, default)

    def clear_local(self) -> None:
        """Clear per-request context (called after each message)."""
        self._local.set({})

    def has(self, key: str) -> bool:
        """Check if a key exists in either local or global context."""
        try:
            return key in self._local.get() or key in self._global
        except LookupError:
            return key in self._global

Methods:

set_global(key: str, value: Any) -> None

Set a global context value (shared across all messages).

Source code in src/rabbitkit/di/context.py 
157
158
159
def set_global(self, key: str, value: Any) -> None:
    """Set a global context value (shared across all messages)."""
    self._global[key] = value

set_local(key: str, value: Any) -> None

Set a per-request context value.

Uses ContextVar.set with an immutable copy so that each coroutine/task gets its own isolated snapshot (contextvars copy-on-write semantics).

Source code in src/rabbitkit/di/context.py 
161
162
163
164
165
166
167
168
169
170
171
172
def set_local(self, key: str, value: Any) -> None:
    """Set a per-request context value.

    Uses ``ContextVar.set`` with an immutable copy so that each
    coroutine/task gets its own isolated snapshot (contextvars
    copy-on-write semantics).
    """
    try:
        current = self._local.get()
    except LookupError:
        current = {}
    self._local.set({**current, key: value})

get(key: str, default: Any = None) -> Any

Get a context value. Local overrides global.

Source code in src/rabbitkit/di/context.py 
174
175
176
177
178
179
180
181
182
def get(self, key: str, default: Any = None) -> Any:
    """Get a context value. Local overrides global."""
    try:
        local = self._local.get()
    except LookupError:
        local = {}
    if key in local:
        return local[key]
    return self._global.get(key, default)

clear_local() -> None

Clear per-request context (called after each message).

Source code in src/rabbitkit/di/context.py 
184
185
186
def clear_local(self) -> None:
    """Clear per-request context (called after each message)."""
    self._local.set({})

has(key: str) -> bool

Check if a key exists in either local or global context.

Source code in src/rabbitkit/di/context.py 
188
189
190
191
192
193
def has(self, key: str) -> bool:
    """Check if a key exists in either local or global context."""
    try:
        return key in self._local.get() or key in self._global
    except LookupError:
        return key in self._global

DIResolver

DIResolver

Resolves handler parameters.

Resolution rules (Contract 4): 1. Annotated with DI marker (Depends/Header/Path/Context) → resolve via marker 2. Type is RabbitMessage → inject the message object 3. Remaining unannotated parameters → ONE body-bound parameter allowed 4. Multiple body-like parameters → ConfigurationError at registration time 5. Parameters with defaults → use default if no other resolution applies

Constraint: At most one body-bound parameter per handler.

Source code in src/rabbitkit/di/resolver.py 
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
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
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
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
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
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
493
494
495
496
497
498
499
500
501
502
503
class DIResolver:
    """Resolves handler parameters.

    Resolution rules (Contract 4):
    1. Annotated with DI marker (Depends/Header/Path/Context) → resolve via marker
    2. Type is RabbitMessage → inject the message object
    3. Remaining unannotated parameters → ONE body-bound parameter allowed
    4. Multiple body-like parameters → ConfigurationError at registration time
    5. Parameters with defaults → use default if no other resolution applies

    Constraint: At most one body-bound parameter per handler.
    """

    def __init__(self) -> None:
        # Per-handler cache of (signature, type-hints). Reflection — especially
        # typing.get_type_hints() — is expensive and STATIC per handler, so it is
        # computed once and reused on the per-message hot path.
        self._sig_hints_cache: dict[Any, tuple[inspect.Signature, dict[str, Any]]] = {}

    def _sig_and_hints(self, handler: Callable[..., Any]) -> tuple[inspect.Signature, dict[str, Any]]:
        cached = self._sig_hints_cache.get(handler)
        if cached is None:
            cached = (inspect.signature(handler), self._get_type_hints(handler))
            self._sig_hints_cache[handler] = cached
        return cached

    def _get_type_hints(self, handler: Callable[..., Any]) -> dict[str, Any]:
        """Get resolved type hints for handler. See :func:`get_type_hints_with_fallback`."""
        return get_type_hints_with_fallback(handler)

    def validate_handler(self, handler: Callable[..., Any]) -> None:
        """Validate handler signature at registration time.

        Raises ConfigurationError for:
        - *args or **kwargs
        - Multiple body-like parameters
        - An annotation that looks like an unresolved DI marker (L11) — see
          ``_looks_like_unresolved_di_marker``
        """
        sig, hints = self._sig_and_hints(handler)
        body_params: list[str] = []

        for param_name, param in sig.parameters.items():
            # Reject *args and **kwargs
            if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
                raise ConfigurationError(
                    f"Handler '{handler.__qualname__}': *args/**kwargs not supported. "
                    "Use explicit parameters with type annotations."
                )

            ann = hints.get(param_name, inspect.Parameter.empty)

            # L11: get_type_hints_with_fallback() could not resolve this
            # annotation to a real type (it's still a raw PEP 563 string),
            # and it textually looks like a DI marker call. Silently
            # continuing would bind this parameter to the message body
            # instead of resolving it via the marker — fail fast instead.
            if _looks_like_unresolved_di_marker(ann):
                raise ConfigurationError(
                    f"Handler '{handler.__qualname__}': parameter '{param_name}' has an "
                    f"annotation ({ann!r}) that looks like it uses a DI marker "
                    "(Depends/Header/Path/Context) but rabbitkit could not resolve it to "
                    "a real type. This usually means the annotated type is not reachable "
                    "from the handler's module globals or enclosing closure -- e.g. it is "
                    "only imported under `if TYPE_CHECKING:`, or defined somewhere "
                    "typing.get_type_hints() can't see. Left unresolved, this parameter "
                    "would silently bind to the message body instead of the marker. Make "
                    "the annotated type resolvable (e.g. import it unconditionally) to fix."
                )

            # Check if it's a DI-annotated parameter
            if self._has_di_marker(ann):
                continue

            # Check if it's RabbitMessage type (class or string form)
            if _is_rabbit_message(ann):
                continue

            # Check if it has a default (non-body)
            if param.default is not inspect.Parameter.empty:
                continue

            # Unannotated parameters are NOT body-like candidates: the fallback
            # resolver binds the first unannotated param to the body and subsequent
            # unannotated params to the message (a documented pattern, e.g.
            # ``handle(body, msg)``). Only flag ANNOTATED params that look like body
            # types (a clear intent signal that multiple body bindings are expected,
            # which the resolver does not support) — e.g. ``handle(a: Order, b: Customer)``.
            if ann is inspect.Parameter.empty:
                continue

            # This is an annotated body-like parameter
            body_params.append(param_name)

        if len(body_params) > 1:
            raise ConfigurationError(
                f"Handler '{handler.__qualname__}': multiple body-like parameters "
                f"({', '.join(body_params)}). At most one body parameter allowed. "
                "Annotate extra parameters with Depends(), Header(), Path(), or Context()."
            )

    def resolve(
        self,
        handler: Callable[..., Any],
        message: RabbitMessage,
        context_repo: ContextRepo | None,
        body: Any,
        scope: DependencyScope | None = None,
    ) -> dict[str, Any]:
        """Resolve all handler parameters at message processing time."""
        sig, hints = self._sig_and_hints(handler)
        kwargs: dict[str, Any] = {}
        depends_cache: dict[int, Any] = {}
        body_injected = False

        for param_name, param in sig.parameters.items():
            ann = hints.get(param_name, inspect.Parameter.empty)

            # Rule 1: DI-annotated parameters
            marker = self._extract_di_marker(ann)
            if marker is not None:
                if isinstance(marker, Depends):
                    kwargs[param_name] = self._resolve_depends(marker, depends_cache, scope)
                else:
                    kwargs[param_name] = self._resolve_marker_with_fallback(
                        marker, param, param_name, message, context_repo
                    )
                continue

            # Rule 2: RabbitMessage type (class or string form)
            if _is_rabbit_message(ann):
                kwargs[param_name] = message
                continue

            # Rule 3: Body-bound parameter (first one)
            if not body_injected and param.default is inspect.Parameter.empty:
                kwargs[param_name] = body
                body_injected = True
                continue

            # Rule 5: Parameters with defaults — omit (use default)

        return kwargs

    async def resolve_async(
        self,
        handler: Callable[..., Any],
        message: RabbitMessage,
        context_repo: ContextRepo | None,
        body: Any,
        scope: DependencyScope | None = None,
    ) -> dict[str, Any]:
        """Resolve all handler parameters, supporting async generator dependencies."""
        sig, hints = self._sig_and_hints(handler)
        kwargs: dict[str, Any] = {}
        depends_cache: dict[int, Any] = {}
        body_injected = False

        for param_name, param in sig.parameters.items():
            ann = hints.get(param_name, inspect.Parameter.empty)

            # Rule 1: DI-annotated parameters
            marker = self._extract_di_marker(ann)
            if marker is not None:
                if isinstance(marker, Depends):
                    kwargs[param_name] = await self._resolve_depends_async(marker, depends_cache, scope)
                else:
                    kwargs[param_name] = self._resolve_marker_with_fallback(
                        marker, param, param_name, message, context_repo
                    )
                continue

            # Rule 2: RabbitMessage type (class or string form)
            if _is_rabbit_message(ann):
                kwargs[param_name] = message
                continue

            # Rule 3: Body-bound parameter (first one)
            if not body_injected and param.default is inspect.Parameter.empty:
                kwargs[param_name] = body
                body_injected = True
                continue

            # Rule 5: Parameters with defaults — omit (use default)

        return kwargs

    # ── Internal helpers ─────────────────────────────────────────────────

    def _has_di_marker(self, ann: Any) -> bool:
        """Check if annotation has a DI marker."""
        return self._extract_di_marker(ann) is not None

    def _extract_di_marker(self, ann: Any) -> Depends | Header | Path | Context | None:
        """Extract DI marker from annotation (supports Annotated)."""
        if ann is inspect.Parameter.empty:
            return None

        # Check Annotated[Type, Marker]
        if get_origin(ann) is Annotated:
            args = get_args(ann)
            for arg in args[1:]:
                if isinstance(arg, (Depends, Header, Path, Context)):
                    return arg
        return None

    def _resolve_marker_with_fallback(
        self,
        marker: Header | Path | Context,
        param: inspect.Parameter,
        param_name: str,
        message: RabbitMessage,
        context_repo: ContextRepo | None,
    ) -> Any:
        """Resolve a Header()/Path()/Context() marker (H10).

        Fallback order when the value is absent from the message:
        1. The marker's own ``default=`` (checked first, inside
           ``_resolve_header``/``_resolve_path``/``_resolve_context``).
        2. The handler parameter's own Python default (checked here).
        3. Neither present → ``MissingDependencyError`` (PERMANENT).

        ``Depends()`` markers never reach here — both ``resolve()`` and
        ``resolve_async()`` special-case them before calling this.
        """
        try:
            if isinstance(marker, Header):
                return self._resolve_header(marker, message, param_name)
            if isinstance(marker, Path):
                return self._resolve_path(marker, message, param_name)
            if isinstance(marker, Context):
                return self._resolve_context(marker, context_repo, param_name)
            raise ConfigurationError(f"Unknown DI marker: {marker!r}")  # pragma: no cover - defensive
        except MissingDependencyError:
            if param.default is not inspect.Parameter.empty:
                return param.default
            raise

    def _resolve_depends(self, marker: Depends, cache: dict[int, Any], scope: DependencyScope | None = None) -> Any:
        """Resolve a Depends() marker, with support for sync generator dependencies."""
        dep_id = id(marker.dependency)
        if marker.use_cache and dep_id in cache:
            return cache[dep_id]

        if inspect.isgeneratorfunction(marker.dependency):
            gen = marker.dependency()
            result = next(gen)
            if scope is not None:
                scope.add_sync_generator(gen)
        elif inspect.isasyncgenfunction(marker.dependency):
            raise ConfigurationError(
                f"Async generator dependency '{marker.dependency.__qualname__}' "
                "requires async pipeline. Use resolve_async() or ensure handler is async."
            )
        else:
            result = marker.dependency()

        if marker.use_cache:
            cache[dep_id] = result
        return result

    async def _resolve_depends_async(
        self, marker: Depends, cache: dict[int, Any], scope: DependencyScope | None = None
    ) -> Any:
        """Resolve a Depends() marker (async), with support for async/sync generator dependencies."""
        dep_id = id(marker.dependency)
        if marker.use_cache and dep_id in cache:
            return cache[dep_id]

        if inspect.isasyncgenfunction(marker.dependency):
            gen = marker.dependency()
            result = await gen.__anext__()
            if scope is not None:
                scope.add_async_generator(gen)
        elif inspect.isgeneratorfunction(marker.dependency):
            gen = marker.dependency()
            result = next(gen)
            if scope is not None:
                scope.add_sync_generator(gen)
        else:
            result = marker.dependency()
            if hasattr(result, "__await__"):
                result = await result

        if marker.use_cache:
            cache[dep_id] = result
        return result

    def _resolve_header(self, marker: Header, message: RabbitMessage, param_name: str) -> Any:
        """Resolve a Header() marker (H10: marker default checked first)."""
        if marker.name in message.headers:
            return message.headers[marker.name]
        if marker.has_default:
            return marker.default
        raise MissingDependencyError(repr(marker), param_name)

    def _resolve_path(self, marker: Path, message: RabbitMessage, param_name: str) -> Any:
        """Resolve a Path() marker (H10: marker default checked first)."""
        if marker.segment in message.path:
            return message.path[marker.segment]
        if marker.has_default:
            return marker.default
        raise MissingDependencyError(repr(marker), param_name)

    def _resolve_context(self, marker: Context, context_repo: ContextRepo | None, param_name: str) -> Any:
        """Resolve a Context() marker (H10: marker default checked first)."""
        if context_repo is None:
            raise ConfigurationError("ContextRepo not available for Context() resolution")
        if context_repo.has(marker.key):
            return context_repo.get(marker.key)
        if marker.has_default:
            return marker.default
        raise MissingDependencyError(repr(marker), param_name)

Methods:

validate_handler(handler: Callable[..., Any]) -> None

Validate handler signature at registration time.

Raises ConfigurationError for: - args or *kwargs - Multiple body-like parameters - An annotation that looks like an unresolved DI marker (L11) — see _looks_like_unresolved_di_marker

Source code in src/rabbitkit/di/resolver.py 
221
222
223
224
225
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
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def validate_handler(self, handler: Callable[..., Any]) -> None:
    """Validate handler signature at registration time.

    Raises ConfigurationError for:
    - *args or **kwargs
    - Multiple body-like parameters
    - An annotation that looks like an unresolved DI marker (L11) — see
      ``_looks_like_unresolved_di_marker``
    """
    sig, hints = self._sig_and_hints(handler)
    body_params: list[str] = []

    for param_name, param in sig.parameters.items():
        # Reject *args and **kwargs
        if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
            raise ConfigurationError(
                f"Handler '{handler.__qualname__}': *args/**kwargs not supported. "
                "Use explicit parameters with type annotations."
            )

        ann = hints.get(param_name, inspect.Parameter.empty)

        # L11: get_type_hints_with_fallback() could not resolve this
        # annotation to a real type (it's still a raw PEP 563 string),
        # and it textually looks like a DI marker call. Silently
        # continuing would bind this parameter to the message body
        # instead of resolving it via the marker — fail fast instead.
        if _looks_like_unresolved_di_marker(ann):
            raise ConfigurationError(
                f"Handler '{handler.__qualname__}': parameter '{param_name}' has an "
                f"annotation ({ann!r}) that looks like it uses a DI marker "
                "(Depends/Header/Path/Context) but rabbitkit could not resolve it to "
                "a real type. This usually means the annotated type is not reachable "
                "from the handler's module globals or enclosing closure -- e.g. it is "
                "only imported under `if TYPE_CHECKING:`, or defined somewhere "
                "typing.get_type_hints() can't see. Left unresolved, this parameter "
                "would silently bind to the message body instead of the marker. Make "
                "the annotated type resolvable (e.g. import it unconditionally) to fix."
            )

        # Check if it's a DI-annotated parameter
        if self._has_di_marker(ann):
            continue

        # Check if it's RabbitMessage type (class or string form)
        if _is_rabbit_message(ann):
            continue

        # Check if it has a default (non-body)
        if param.default is not inspect.Parameter.empty:
            continue

        # Unannotated parameters are NOT body-like candidates: the fallback
        # resolver binds the first unannotated param to the body and subsequent
        # unannotated params to the message (a documented pattern, e.g.
        # ``handle(body, msg)``). Only flag ANNOTATED params that look like body
        # types (a clear intent signal that multiple body bindings are expected,
        # which the resolver does not support) — e.g. ``handle(a: Order, b: Customer)``.
        if ann is inspect.Parameter.empty:
            continue

        # This is an annotated body-like parameter
        body_params.append(param_name)

    if len(body_params) > 1:
        raise ConfigurationError(
            f"Handler '{handler.__qualname__}': multiple body-like parameters "
            f"({', '.join(body_params)}). At most one body parameter allowed. "
            "Annotate extra parameters with Depends(), Header(), Path(), or Context()."
        )

resolve(handler: Callable[..., Any], message: RabbitMessage, context_repo: ContextRepo | None, body: Any, scope: DependencyScope | None = None) -> dict[str, Any]

Resolve all handler parameters at message processing time.

Source code in src/rabbitkit/di/resolver.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
def resolve(
    self,
    handler: Callable[..., Any],
    message: RabbitMessage,
    context_repo: ContextRepo | None,
    body: Any,
    scope: DependencyScope | None = None,
) -> dict[str, Any]:
    """Resolve all handler parameters at message processing time."""
    sig, hints = self._sig_and_hints(handler)
    kwargs: dict[str, Any] = {}
    depends_cache: dict[int, Any] = {}
    body_injected = False

    for param_name, param in sig.parameters.items():
        ann = hints.get(param_name, inspect.Parameter.empty)

        # Rule 1: DI-annotated parameters
        marker = self._extract_di_marker(ann)
        if marker is not None:
            if isinstance(marker, Depends):
                kwargs[param_name] = self._resolve_depends(marker, depends_cache, scope)
            else:
                kwargs[param_name] = self._resolve_marker_with_fallback(
                    marker, param, param_name, message, context_repo
                )
            continue

        # Rule 2: RabbitMessage type (class or string form)
        if _is_rabbit_message(ann):
            kwargs[param_name] = message
            continue

        # Rule 3: Body-bound parameter (first one)
        if not body_injected and param.default is inspect.Parameter.empty:
            kwargs[param_name] = body
            body_injected = True
            continue

        # Rule 5: Parameters with defaults — omit (use default)

    return kwargs

resolve_async(handler: Callable[..., Any], message: RabbitMessage, context_repo: ContextRepo | None, body: Any, scope: DependencyScope | None = None) -> dict[str, Any] async

Resolve all handler parameters, supporting async generator dependencies.

Source code in src/rabbitkit/di/resolver.py 
335
336
337
338
339
340
341
342
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
async def resolve_async(
    self,
    handler: Callable[..., Any],
    message: RabbitMessage,
    context_repo: ContextRepo | None,
    body: Any,
    scope: DependencyScope | None = None,
) -> dict[str, Any]:
    """Resolve all handler parameters, supporting async generator dependencies."""
    sig, hints = self._sig_and_hints(handler)
    kwargs: dict[str, Any] = {}
    depends_cache: dict[int, Any] = {}
    body_injected = False

    for param_name, param in sig.parameters.items():
        ann = hints.get(param_name, inspect.Parameter.empty)

        # Rule 1: DI-annotated parameters
        marker = self._extract_di_marker(ann)
        if marker is not None:
            if isinstance(marker, Depends):
                kwargs[param_name] = await self._resolve_depends_async(marker, depends_cache, scope)
            else:
                kwargs[param_name] = self._resolve_marker_with_fallback(
                    marker, param, param_name, message, context_repo
                )
            continue

        # Rule 2: RabbitMessage type (class or string form)
        if _is_rabbit_message(ann):
            kwargs[param_name] = message
            continue

        # Rule 3: Body-bound parameter (first one)
        if not body_injected and param.default is inspect.Parameter.empty:
            kwargs[param_name] = body
            body_injected = True
            continue

        # Rule 5: Parameters with defaults — omit (use default)

    return kwargs

DependencyScope

DependencyScope

Tracks generator dependencies for cleanup after handler completes.

Source code in src/rabbitkit/di/resolver.py 
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
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
class DependencyScope:
    """Tracks generator dependencies for cleanup after handler completes."""

    def __init__(self) -> None:
        self._sync_generators: list[Generator[Any, None, None]] = []
        self._async_generators: list[AsyncGenerator[Any, None]] = []

    def add_sync_generator(self, gen: Generator[Any, None, None]) -> None:
        self._sync_generators.append(gen)

    def add_async_generator(self, gen: AsyncGenerator[Any, None]) -> None:
        self._async_generators.append(gen)

    def cleanup(self) -> None:
        """Close all sync generators (in reverse order).

        Each generator's teardown is isolated: a raising teardown is logged
        and skipped so one misbehaving generator does not leak the rest or
        prevent ``clear()`` from running.
        """
        for gen in reversed(self._sync_generators):
            try:
                next(gen, None)
            except StopIteration:  # pragma: no cover
                pass  # pragma: no cover
            except Exception:
                _logger.warning("sync generator teardown raised", exc_info=True)
            finally:
                try:
                    gen.close()
                except Exception:
                    _logger.warning("sync generator close() raised", exc_info=True)
        self._sync_generators.clear()

    async def cleanup_async(self) -> None:
        """Close all generators (async generators + sync generators, in reverse order).

        Each generator's teardown is isolated: a raising async teardown is
        logged and skipped, and the sync-generator pass always runs even if an
        async generator raised. ``clear()`` always runs.
        """
        for gen in reversed(self._async_generators):
            try:
                await gen.__anext__()
            except StopAsyncIteration:
                pass
            except Exception:
                _logger.warning("async generator teardown raised", exc_info=True)
            finally:
                try:
                    await gen.aclose()
                except Exception:
                    _logger.warning("async generator aclose() raised", exc_info=True)
        self._async_generators.clear()

        for sync_gen in reversed(self._sync_generators):
            try:
                next(sync_gen, None)
            except StopIteration:  # pragma: no cover
                pass  # pragma: no cover
            except Exception:
                _logger.warning("sync generator teardown raised", exc_info=True)
            finally:
                try:
                    sync_gen.close()
                except Exception:
                    _logger.warning("sync generator close() raised", exc_info=True)
        self._sync_generators.clear()

Methods:

cleanup() -> None

Close all sync generators (in reverse order).

Each generator's teardown is isolated: a raising teardown is logged and skipped so one misbehaving generator does not leak the rest or prevent clear() from running.

Source code in src/rabbitkit/di/resolver.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def cleanup(self) -> None:
    """Close all sync generators (in reverse order).

    Each generator's teardown is isolated: a raising teardown is logged
    and skipped so one misbehaving generator does not leak the rest or
    prevent ``clear()`` from running.
    """
    for gen in reversed(self._sync_generators):
        try:
            next(gen, None)
        except StopIteration:  # pragma: no cover
            pass  # pragma: no cover
        except Exception:
            _logger.warning("sync generator teardown raised", exc_info=True)
        finally:
            try:
                gen.close()
            except Exception:
                _logger.warning("sync generator close() raised", exc_info=True)
    self._sync_generators.clear()

cleanup_async() -> None async

Close all generators (async generators + sync generators, in reverse order).

Each generator's teardown is isolated: a raising async teardown is logged and skipped, and the sync-generator pass always runs even if an async generator raised. clear() always runs.

Source code in src/rabbitkit/di/resolver.py 
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
async def cleanup_async(self) -> None:
    """Close all generators (async generators + sync generators, in reverse order).

    Each generator's teardown is isolated: a raising async teardown is
    logged and skipped, and the sync-generator pass always runs even if an
    async generator raised. ``clear()`` always runs.
    """
    for gen in reversed(self._async_generators):
        try:
            await gen.__anext__()
        except StopAsyncIteration:
            pass
        except Exception:
            _logger.warning("async generator teardown raised", exc_info=True)
        finally:
            try:
                await gen.aclose()
            except Exception:
                _logger.warning("async generator aclose() raised", exc_info=True)
    self._async_generators.clear()

    for sync_gen in reversed(self._sync_generators):
        try:
            next(sync_gen, None)
        except StopIteration:  # pragma: no cover
            pass  # pragma: no cover
        except Exception:
            _logger.warning("sync generator teardown raised", exc_info=True)
        finally:
            try:
                sync_gen.close()
            except Exception:
                _logger.warning("sync generator close() raised", exc_info=True)
    self._sync_generators.clear()