diff --git a/docs/integrations/aiogram.md b/docs/integrations/aiogram.md index 9b103b61..fc74ac05 100644 --- a/docs/integrations/aiogram.md +++ b/docs/integrations/aiogram.md @@ -1,10 +1,10 @@ # Usage with `aiogram` -aiogram has no dependency-injection system of its own, so `modern-di-aiogram` -uses the `@inject` decorator with `FromDI` markers (or `auto_inject=True` to -skip the decorator entirely). `setup_di` opens the root container on -dispatcher startup, closes it on shutdown, and installs middleware that opens -a per-update child container automatically. +aiogram passes handler arguments by name from its middleware data, but it has no typed +provider graph. `modern-di-aiogram` adds one: mark handler parameters with `FromDI` and +decorate the handler with `@inject`, or pass `auto_inject=True` to skip the decorator. +`setup_di` builds a `Scope.REQUEST` child container per update and opens and closes the root +container with the dispatcher. ## How to use @@ -61,7 +61,7 @@ class AppGroup(Group): dispatcher = Dispatcher() container = Container(groups=[AppGroup]) setup_di(dispatcher, container) -container.validate() # after setup_di — its connection providers are now registered +container.validate() # after setup_di: its context providers are now registered @dispatcher.message() @@ -73,91 +73,34 @@ async def greet( await message.answer(str(report.as_dict())) ``` -`setup_di(dispatcher, container)` stores the container on the dispatcher, -registers `dispatcher.startup`/`dispatcher.shutdown` handlers that open/close -it, and installs an update-level outer middleware that builds a per-update -child container. - -## Auto-injecting handlers - -Passing `auto_inject=True` to `setup_di` wraps every handler already -registered on the dispatcher with `@inject` automatically, so individual -handlers don't need the decorator: - -```python -import typing - -from aiogram import Dispatcher, Router -from aiogram.types import Message -from modern_di import Container, Group, Scope, providers -from modern_di_aiogram import FromDI, setup_di - - -class Settings: - def __init__(self) -> None: - self.greeting = "hello" - - -class AppGroup(Group): - settings = providers.Factory(Settings, scope=Scope.APP, cache=True) - - -router = Router() - - -@router.message() -async def greet( - message: Message, - settings: typing.Annotated[Settings, FromDI(AppGroup.settings)], -) -> None: - await message.answer(f"{settings.greeting}, {message.from_user.first_name}") - - -dispatcher = Dispatcher() -dispatcher.include_router(router) -container = Container(groups=[AppGroup]) -setup_di(dispatcher, container, auto_inject=True) -container.validate() # after setup_di — its connection providers are now registered -``` - -!!! warning "Register handlers before startup" - `auto_inject` wraps handlers on `dispatcher.startup`, which fires from - `dispatcher.emit_startup()`, the call `start_polling()`/`start_webhook()` - makes before serving updates. Only handlers registered (via - `dispatcher.include_router()` or the decorators directly) **before** - `emit_startup()` runs are wrapped; a handler added afterward is invoked - without injection and any `FromDI` parameter on it is left unresolved. + ## Scopes -The integration creates one `Scope.REQUEST` child container per update. -The middleware is installed on `dispatcher.update` as an -[outer middleware](https://docs.aiogram.dev/en/latest/dispatcher/middlewares.html), -so it wraps every update regardless of which router or handler ultimately -processes it. The child container is closed after the handler runs, -including when it raises. +The integration creates one `Scope.REQUEST` child container per update. Its middleware is an +[outer middleware](https://docs.aiogram.dev/en/latest/dispatcher/middlewares.html) on +`dispatcher.update`, so it wraps every update whichever router or handler processes it. The +child is closed with `close_async()` after the handler returns or raises, so REQUEST-scoped +finalizers may be async or sync. Resolution itself is synchronous, as everywhere in modern-di. There is no `Scope.SESSION` for aiogram: each Telegram update is handled independently; there's no persistent per-chat/per-user connection comparable to a WebSocket. See [the scope hierarchy](../providers/scopes.md#what-each-scope-is-for). -## Sync resolution, async cleanup +## Root container lifecycle -`FromDI` resolves its dependency with `Container.resolve_dependency(...)`, -which is synchronous; modern-di's resolution is always sync, regardless of -the framework. The per-update `Scope.REQUEST` child container that resolution -runs against is nevertheless torn down asynchronously: after the handler -finishes (or raises), the integration awaits `child_container.close_async()`. -So async finalizers on REQUEST-scoped providers run correctly, while the -factories themselves must build synchronously. +`setup_di` registers `container.open` on `dispatcher.startup` and `container.close_async` on +`dispatcher.shutdown`, and closing runs APP-scoped finalizers. `start_polling()` emits both +events, and so does an aiohttp app wired with aiogram's `setup_application()`. ## Framework context objects -`aiogram.types.Update` and the concrete event it carries (`Message`, -`CallbackQuery`, etc.) are automatically made available by the integration, -so factories can declare them as parameters. See -[Framework context objects](../providers/context.md#framework-context-objects) -for how implicit and explicit resolution work. +Two types resolve by annotation: `aiogram.types.Update` and `aiogram.types.TelegramObject`, +the concrete event unwrapped from the update. See +[Framework context objects](../providers/context.md#framework-context-objects) for how +implicit and explicit resolution work. The following context providers are also available for explicit import: @@ -166,6 +109,10 @@ The following context providers are also available for explicit import: the concrete event unwrapped from the `Update` (e.g. a `Message` or `CallbackQuery` instance). +A concrete event type such as `Message` has no provider of its own, so a factory that +declares `message: Message` fails `container.validate()`. Annotate the parameter as +`TelegramObject`, or wire it to `aiogram_event_provider` with `kwargs`. + ### Implicit (type-based) usage ```python @@ -190,9 +137,27 @@ class AppGroup(Group): ### Explicit (provider-based) usage -`aiogram_event_provider` is bound to the base `TelegramObject` type, so -narrowing a parameter to a concrete event type (like `Message`) requires -wiring it explicitly with `FromDI`: +`aiogram_event_provider` is bound to the base `TelegramObject` type. To get the event +typed as `Message`, wire the provider explicitly. In a factory that is `kwargs`: + +```python +from aiogram.types import Message +from modern_di_aiogram import aiogram_event_provider + + +def message_text(message: Message) -> str: + return message.text or "" + + +class AppGroup(Group): + text = providers.Factory( + message_text, + scope=Scope.REQUEST, + kwargs={"message": aiogram_event_provider}, + ) +``` + +In a handler it is `FromDI`: ```python import typing @@ -209,22 +174,58 @@ async def log_message( assert message is same_message ``` -## See also +## Auto-injecting handlers -- [Testing with overrides](../recipes/testing-overrides.md): swap providers in your tests. -- [Lifecycle](../providers/lifecycle.md): finalizers and container teardown. -- [Scopes](../providers/scopes.md): the APP → REQUEST lifetime model. +Passing `auto_inject=True` to `setup_di` wraps every handler already +registered on the dispatcher with `@inject` automatically, so individual +handlers don't need the decorator: -## API +```python +import typing -| Symbol | Description | -|---|---| -| `setup_di(dispatcher, container, *, auto_inject=False)` | Stores the container on the dispatcher, registers `aiogram_update_provider`/`aiogram_event_provider`, wires `dispatcher.startup`/`dispatcher.shutdown` to open/close the container, and installs the per-update middleware. With `auto_inject=True`, also wraps every handler already registered on the dispatcher at startup. | -| `FromDI(dependency)` | Marker (used with `@inject`) that resolves a provider or type from the per-update child container. | -| `inject` | Decorator for an aiogram handler; resolves its `FromDI`-annotated parameters. Not needed when `setup_di(..., auto_inject=True)` is used. Raises `RuntimeError` naming `setup_di` when an update reaches it without the middleware installed. | -| `fetch_di_container(dispatcher)` | Returns the root `Container` stored on the dispatcher. | -| `aiogram_update_provider` | `ContextProvider` for the current `aiogram.types.Update` (REQUEST scope). | -| `aiogram_event_provider` | `ContextProvider` for the current `aiogram.types.TelegramObject` (REQUEST scope), the concrete event unwrapped from the `Update`. | +from aiogram import Dispatcher, Router +from aiogram.types import Message +from modern_di import Container, Group, Scope, providers +from modern_di_aiogram import FromDI, setup_di + + +class Settings: + def __init__(self) -> None: + self.greeting = "hello" + + +class AppGroup(Group): + settings = providers.Factory(Settings, scope=Scope.APP, cache=True) + + +router = Router() + + +@router.message() +async def greet( + message: Message, + settings: typing.Annotated[Settings, FromDI(AppGroup.settings)], +) -> None: + await message.answer(f"{settings.greeting}, {message.from_user.first_name}") + + +dispatcher = Dispatcher() +dispatcher.include_router(router) +container = Container(groups=[AppGroup]) +setup_di(dispatcher, container, auto_inject=True) +container.validate() # after setup_di: its context providers are now registered +``` + +!!! warning "Register handlers before startup" + `auto_inject` wraps handlers on `dispatcher.startup`, which fires from + `dispatcher.emit_startup()`, the call `start_polling()` makes before serving + updates. Only handlers registered (via `dispatcher.include_router()` or the + decorators directly) **before** `emit_startup()` runs are wrapped. A handler + added afterward is called without injection, so aiogram raises `TypeError` + for its missing `FromDI` argument. + +Handlers on `update` observers (`@dispatcher.update()` or `@router.update()`) are never +wrapped. Decorate those with `@inject` yourself. ## Usage with `aiogram-dialog` @@ -284,3 +285,71 @@ the per-update container. - An `@inject` getter must still declare `**kwargs` (aiogram-dialog always calls getters with the full `middleware_data`), and a `FromDI` getter parameter must not share a name with a `middleware_data` key (e.g. `bot`, `event`). + +## Testing + +Drive a real `Dispatcher` with a bot that has a fake token: emit startup, feed it an +`Update`, and emit shutdown. `feed_update` returns what the handler returned, so a handler +that returns a value is easy to assert on. + + + +```python +import datetime + +from aiogram import Bot, Dispatcher +from aiogram.types import Chat, Message, Update +from modern_di_aiogram import FromDI, inject, setup_di + + +async def test_report() -> None: + dispatcher = Dispatcher() + container = Container(groups=[AppGroup]) + setup_di(dispatcher, container) + + @dispatcher.message() + @inject + async def report_handler( + message: Message, + report: typing.Annotated[Report, FromDI(Report)], + ) -> dict[str, str]: + return report.as_dict() + + bot = Bot("123456:" + "A" * 35) + update = Update( + update_id=1, + message=Message( + message_id=1, + date=datetime.datetime.now(tz=datetime.UTC), + chat=Chat(id=1, type="private"), + text="report", + ), + ) + await dispatcher.emit_startup(bot=bot) + assert await dispatcher.feed_update(bot, update) == {"service": "catalog"} + await dispatcher.emit_shutdown(bot=bot) + await bot.session.close() +``` + + + +## See also + +- [Testing with overrides](../recipes/testing-overrides.md): swap providers in your tests. +- [Lifecycle](../providers/lifecycle.md): finalizers and container teardown. +- [Scopes](../providers/scopes.md): the APP → REQUEST lifetime model. + +## API + +| Symbol | Description | +|---|---| +| `setup_di(dispatcher, container, *, auto_inject=False)` | Stores the container on the dispatcher, registers `aiogram_update_provider`/`aiogram_event_provider`, wires `dispatcher.startup`/`dispatcher.shutdown` to open/close the container, and installs the per-update middleware. With `auto_inject=True`, also wraps every handler already registered on the dispatcher at startup. | +| `FromDI(dependency)` | Marker (used with `@inject`) that resolves a provider or type from the per-update child container. | +| `inject` | Decorator for an aiogram handler; resolves its `FromDI`-annotated parameters. Not needed when `setup_di(..., auto_inject=True)` is used. Raises `RuntimeError` naming `setup_di` when an update reaches it without the middleware installed. | +| `fetch_di_container(dispatcher)` | Returns the root `Container` stored on the dispatcher. | +| `aiogram_update_provider` | `ContextProvider` for the current `aiogram.types.Update` (REQUEST scope). | +| `aiogram_event_provider` | `ContextProvider` for the current `aiogram.types.TelegramObject` (REQUEST scope), the concrete event unwrapped from the `Update`. | diff --git a/docs/integrations/arq.md b/docs/integrations/arq.md index 3f02533d..7ade92e5 100644 --- a/docs/integrations/arq.md +++ b/docs/integrations/arq.md @@ -82,70 +82,81 @@ async def main() -> None: await pool.enqueue_job("run_report") ``` -`setup_di(worker_settings, container)` seeds the container into arq's `ctx` dict -(arq's per-worker state store) and wraps four of arq's lifecycle hooks: -`on_startup`/`on_shutdown` open and close the root container, and -`on_job_start`/`on_job_end` build and, as a safety net, close a `Scope.REQUEST` -child container around each job. Any hook you already defined still runs: yours -runs *after* ours on startup/job-start and *before* ours on shutdown/job-end. The -root container is live in all of them, but for an `@inject` task the child is -already closed by the time your `on_job_end` runs. It accepts a `WorkerSettings` class (the common -case) or a plain settings `dict`, and returns the container. - -`@inject` resolves each `FromDI`-annotated parameter from the per-job child -container and forwards it to your task. Your task **must** declare arq's `ctx` -dict as its first parameter (arq calls every task as `task(ctx, *args)`). -Injection is parameter-order-insensitive, so a `FromDI` parameter may sit -anywhere in the signature, and a task with no `FromDI` parameter is returned -unchanged. +`setup_di(worker_settings, container)` puts the container in arq's `ctx` dict and +wraps four of arq's hooks: `on_startup` and `on_shutdown` open and close the root +container, and `on_job_start` and `on_job_end` build and close a `Scope.REQUEST` +child around each job. It accepts a `WorkerSettings` class or a plain settings +`dict`, and returns the container. + +`@inject` resolves each `FromDI`-annotated parameter from the job's child container +and passes it to your task. arq calls every task as `task(ctx, *args)`, so `ctx` +must be the first parameter apart from the `FromDI` ones, which may sit anywhere in +the signature. A task with no `FromDI` parameter is returned unchanged. ## Scopes -The integration builds one `Scope.REQUEST` child container per job in -`on_job_start`. For an `@inject` task, the wrapper closes it with -`close_async()` when the task body exits, whether it returned or raised. Nested -or concurrent `@inject` calls in the same job share the child, and the last one -to exit closes it. `on_job_end` closes the child only if it is still open, which -covers jobs that ran no `@inject` wrapper. Either way REQUEST-scoped providers -(and their finalizers) never leak on the error path. APP-scoped providers persist for the whole worker: `setup_di` opens -the root container on `on_startup` and closes it on `on_shutdown`, running -APP-scoped finalizers once when the worker stops. +`on_job_start` builds one `Scope.REQUEST` child container per job. For an `@inject` +task, the wrapper closes it when the task body exits, whether it returned or +raised. Nested or concurrent `@inject` calls in the same job share the child, and +the last one to exit closes it. `on_job_end` closes the child only if it is still +open, which covers a job that ran no `@inject` wrapper. + +Resolution is synchronous, as everywhere in modern-di. The child is closed with +`close_async()`, so REQUEST-scoped finalizers may be async or sync. There is no `Scope.SESSION` for arq: a job queue has no session concept comparable to a websocket connection. -## Async resolution, no connection object +## Root container lifecycle + +`on_startup` opens the root container and `on_shutdown` closes it with +`close_async()`, which runs APP-scoped finalizers once when the worker stops. A +worker that starts again on the same container, after a restart or in a test that +runs the worker twice, reopens it on startup. -`FromDI` resolves its dependency with `Container.resolve_dependency(...)`, which -is synchronous: modern-di's resolution is always sync, regardless of the -framework. Container *lifecycle* here is async, matching arq: the root and each -per-job child are closed with `close_async()`, so REQUEST- and APP-scoped -finalizers may be async (or sync). +A hook you already defined on `WorkerSettings` still runs. Yours runs after ours +on startup and job start, and before ours on shutdown and job end. The root +container is open in all of them, but for an `@inject` task the job's child is +already closed by the time your `on_job_end` runs. -arq's per-job `ctx` is a plain `dict` (`job_id`, `job_try`, `redis`, ...), not a -dedicated request/message type, so `modern_di_arq` registers no context -provider, as with Celery and Typer. A task that needs job metadata reads it from the -`ctx` argument arq already passes. If you need the root container elsewhere (for -example in your own `on_job_start`), `fetch_di_container(ctx)` returns it. +Calling `setup_di` twice on the same `worker_settings` raises a `TypeError`, +because stacking the hook wrappers would leak a per-job child container. -## Restart safety +## Framework context objects -`setup_di` wires `container.open()` onto `on_startup`, and calling `open()` again -on an already-open container is a no-op: it unconditionally clears `closed` and -runs no validation, so it costs nothing regardless of graph state. A worker that -starts, stops (closing the container), and starts again (a restart, or a test -that runs the worker twice) reopens the same container cleanly. Calling `setup_di` twice on the same `worker_settings` is rejected with -a `TypeError`, since stacking the hook wrappers would leak a per-job child -container. +arq's per-job `ctx` is a plain `dict` (`job_id`, `job_try`, `redis` and so on), +so `modern_di_arq` registers no context provider. A task that needs job metadata +reads it from the `ctx` argument. To reach the root container elsewhere, for +example in your own `on_job_start`, call `fetch_di_container(ctx)`. ## Tasks with `*args`/`**kwargs` `@inject` resolves dependencies by binding the task signature by name, which is what makes injection order-insensitive. A task that mixes a `FromDI` parameter with `*args` or `**kwargs` cannot be bound unambiguously, so `@inject` raises a -`TypeError` at decoration time rather than silently misrouting arguments. -Give an `@inject` task explicit named parameters. (A task with no `FromDI` -parameter is untouched and may use `*args`/`**kwargs` freely.) +`TypeError` at decoration time. Give an `@inject` task explicit named parameters. +A task with no `FromDI` parameter is left untouched and may use `*args`/`**kwargs` +freely. + +## Testing + +A burst worker runs the queued jobs and stops, so it drives the whole lifecycle +in a test: startup, one child per job, and shutdown. It needs a running Redis. + + + +```python +from arq.worker import create_worker + + +async def test_run_report() -> None: + pool = await create_pool(RedisSettings(host="localhost")) + job = await pool.enqueue_job("run_report") + worker = create_worker(WorkerSettings, burst=True, handle_signals=False) + await worker.main() + await worker.close() + assert await job.result() == "service=catalog" +``` ## See also diff --git a/docs/integrations/celery.md b/docs/integrations/celery.md index cc865624..e45a4dc9 100644 --- a/docs/integrations/celery.md +++ b/docs/integrations/celery.md @@ -63,122 +63,107 @@ def run_report(report: typing.Annotated[Report, FromDI(Report)]) -> str: return report.render() ``` -`setup_di(app, container)` stores the container on `app.conf` and registers `worker_process_init`/`worker_process_shutdown` and `worker_init`/`worker_shutdown` signal handlers that open/close it. Those fire when a real `celery worker` process starts and stops, so a script or test that calls tasks without spinning one up (e.g. with `task_always_eager = True`) must drive the container lifecycle itself; see [Worker-process lifecycle](#worker-process-lifecycle) below. +`setup_di(app, container)` stores the container on `app.conf` and connects Celery's +worker signals that open and close it. `@inject` builds a `Scope.REQUEST` child container +for each task call and resolves `FromDI`-annotated parameters from it. It looks the +container up through Celery's `current_app` at call time, so it resolves against +whichever app is active. -`@inject` builds a `Scope.REQUEST` child container per call and resolves `FromDI`-annotated parameters from it. It looks the container up through Celery's `current_app` proxy at call time, not the `app` object captured at decoration time, so it always resolves against whichever app is currently active. +Celery closes containers with `close_sync()`, both the per-task child and the root at +worker shutdown. Finalizers must therefore be sync: an async finalizer makes the close +raise `FinalizerError`. ## Scopes -The integration creates a `Scope.REQUEST` child container for each task invocation, whether wired via `@inject` or [`DITask`](#the-ditask-base-class). REQUEST-scoped providers (and their finalizers) live for the duration of that one call; the child container is closed with `close_sync()` once the task returns, including when it raises. APP-scoped providers persist for the whole worker process: `setup_di` opens the APP container on `worker_process_init` (or `worker_init`) and closes it with `close_sync()` on `worker_process_shutdown` (or `worker_shutdown`). +The integration creates a `Scope.REQUEST` child container for each task call, whether +the task is wired with `@inject` or with [`DITask`](#the-ditask-base-class). +REQUEST-scoped providers and their finalizers live for that one call. The child is +closed with `close_sync()` once the task returns or raises. An async finalizer on a +REQUEST-scoped provider makes the task itself fail with `FinalizerError`. -There is no `Scope.SESSION` for Celery: a task queue doesn't have a session concept comparable to websockets. +Resolution is synchronous, as everywhere in modern-di, and Celery tasks are sync +callables, so there is no async teardown path. -## Sync resolution, no connection object +There is no `Scope.SESSION` for Celery: a task queue has no session concept +comparable to websockets. -`FromDI` resolves its dependency with `Container.resolve_dependency(...)`, which is synchronous; modern-di's resolution is always sync, regardless of the framework. Celery tasks are themselves sync callables, so the per-task `Scope.REQUEST` container is torn down the same way it's built: `@inject` calls `close_sync()` in a `finally` block after the task returns. There is no async counterpart. REQUEST-scoped finalizers must be sync (or `close_sync`-compatible) for Celery tasks. +## Root container lifecycle -Unlike aiohttp, FastAPI, or taskiq, a Celery task has no framework request or message object comparable to an HTTP request or a broker message. `modern_di_celery` does not register a context provider, and there is no implicit/explicit "framework context object" for this integration. Pass whatever per-call data a task needs through its own arguments (or `self.request` on a bound task, which is Celery's own mechanism, unrelated to modern-di). +A constructed container is already open, so tasks resolve without any worker signal +having fired. The signals matter for closing: `setup_di` closes the root with +`close_sync()` when the worker stops, which runs APP-scoped finalizers. Which signal +does the work depends on the pool: -## The `DITask` base class +| Pool | Opens the root on | Closes the root on | +|---|---|---| +| `prefork` | `worker_process_init`, in each child process | `worker_process_shutdown`, in each child process | +| `solo` | `worker_process_init` | `worker_shutdown` | +| `threads`, `gevent`, `eventlet` | `worker_init` | `worker_shutdown` | -`DITask` applies `@inject` to a task's `run` method automatically, so individual tasks don't need their own `@inject` decorator. Apply it to every task on an app with `task_cls=DITask`: +Under prefork each child process gets its own APP-scoped instances. `worker_init` and +`worker_shutdown` fire in the main worker process under every pool; opening an open +container and closing one with nothing cached are both no-ops, so the overlap is +harmless. -```python -import typing - -from celery import Celery -from modern_di import Container, Group, Scope, providers -from modern_di_celery import DITask, FromDI, setup_di +!!! warning "The prefork pool fails on macOS" + On macOS, billiard starts prefork children with the spawn method, which pickles + the Celery app. The container that `setup_di` stores on `app.conf` cannot be + pickled, so the worker stops at startup with + `TypeError: cannot pickle '_thread._local' object`. Run the worker with + `--pool=solo` or `--pool=threads` there. +Code that calls tasks without a worker, such as a script or a test in eager mode, +fires none of these signals. Close the root yourself when it is done, for example +with `with container:`, as in [Testing](#testing). -class Settings: - def __init__(self) -> None: - self.greeting = "hello" +## Framework context objects +A Celery task has no framework request or message object, so `modern_di_celery` +registers no context provider. Pass the per-call data a task needs through its +arguments, or read `self.request` on a bound task. -class AppGroup(Group): - settings = providers.Factory(Settings, scope=Scope.APP, cache=True) - - -app = Celery("myapp", broker="redis://localhost", task_cls=DITask) -container = Container(groups=[AppGroup]) -setup_di(app, container) -container.validate() # fails fast on a broken graph before the worker runs - +## The `DITask` base class -@app.task -def greet(name: str, settings: typing.Annotated[Settings, FromDI(Settings)]) -> str: - return f"{settings.greeting}, {name}" -``` +`DITask` applies `@inject` to a task's `run` method, so individual tasks don't need +the decorator. Pass `task_cls=DITask` to apply it to every task on the app: -Or apply it to a single task instead of the whole app: + ```python -import typing - -from celery import Celery -from modern_di import Container, Group, Scope, providers -from modern_di_celery import DITask, FromDI, setup_di - - -class Settings: - def __init__(self) -> None: - self.greeting = "hello" +from modern_di_celery import DITask - -class AppGroup(Group): - settings = providers.Factory(Settings, scope=Scope.APP, cache=True) - - -app = Celery("myapp", broker="redis://localhost") +app = Celery("myapp", broker="redis://localhost", task_cls=DITask) container = Container(groups=[AppGroup]) setup_di(app, container) -container.validate() # fails fast on a broken graph before the worker runs -@app.task(base=DITask) -def greet(name: str, settings: typing.Annotated[Settings, FromDI(Settings)]) -> str: - return f"{settings.greeting}, {name}" +@app.task +def labelled_report(label: str, report: typing.Annotated[Report, FromDI(Report)]) -> str: + return f"{label}: {report.render()}" ``` -`DITask.__init__` wraps `self.run` with `inject` the first time the task class is instantiated (Celery instantiates each task class once per app) and resets `self.__header__` via `head_from_fun` so Celery still binds call arguments against the *visible*, non-DI signature. It skips re-wrapping if `run` is already injected, so stacking an explicit `@inject` under `@app.task(base=DITask)` is safe and only wraps once. +To apply it to a single task instead, pass `base=DITask` to that task's decorator: +`@app.task(base=DITask)`. Celery still checks call arguments against the signature +without the `FromDI` parameters, so `labelled_report.delay()` with no `label` raises +`TypeError`. A task that also carries an explicit `@inject` is wrapped only once. -## Worker-process lifecycle +## Testing -`setup_di` connects to Celery's `worker_process_init`/`worker_process_shutdown` and `worker_init`/`worker_shutdown` signals with `weak=False`. Celery signals default to weak references, which would otherwise let the handlers be garbage-collected before a worker process ever fires them. `container.open()` runs on `worker_process_init` and `worker_init`, and `container.close_sync()` runs on `worker_process_shutdown` and `worker_shutdown`. The prefork and solo pools send `worker_process_init`/`worker_process_shutdown` once per worker process, so each forked process gets its own APP-scoped providers. The threads, gevent, and eventlet pools never fork and send only `worker_init`/`worker_shutdown`, once in the main worker process. Either way the signals fire per worker, not per task. Opening an open container and closing one with nothing cached are both no-ops, so the overlap is harmless. +With `task_always_eager`, a task runs in the calling process, so no worker signal +fires. Wrap the calls in `with container:` to close the root and run APP-scoped +finalizers at the end: -A real `celery worker` invocation fires both signals automatically. Code that calls tasks without a running worker (a script, or a test using `task_always_eager`) must trigger the same signals (or drive the container directly) itself: + ```python -from celery import Celery, signals -from modern_di import Container, Group, Scope, providers -from modern_di_celery import setup_di - - -class Settings: - def __init__(self) -> None: - self.greeting = "hello" - - -class AppGroup(Group): - settings = providers.Factory(Settings, scope=Scope.APP, cache=True) - - -app = Celery("myapp", broker="memory://", backend="cache+memory://") app.conf.task_always_eager = True -app.conf.task_store_eager_result = True -container = Container(groups=[AppGroup]) -setup_di(app, container) - - -@app.task -def ping() -> str: - return "pong" - -signals.worker_process_init.send(sender=None) # a real worker fires this on startup -print(ping.delay().get()) # -> "pong" -signals.worker_process_shutdown.send(sender=None) # a real worker fires this on shutdown +with container: + assert run_report.delay().get() == "service=catalog" ``` ## See also diff --git a/docs/integrations/fastmcp.md b/docs/integrations/fastmcp.md index d7f447a7..41e89855 100644 --- a/docs/integrations/fastmcp.md +++ b/docs/integrations/fastmcp.md @@ -84,13 +84,22 @@ extend-immutable-calls = ["modern_di_fastmcp.FromDI"] ## Scopes -`setup_di` opens the APP container when the server's lifespan starts and closes it with -`close_async()` when the lifespan ends. A middleware opens a `Scope.REQUEST` child for every MCP -request: a tool call, a resource read, a prompt render or a list call. Notifications open nothing. -There is no `Scope.SESSION` child: an MCP session has no close hook, so nothing could close it. +A middleware opens a `Scope.REQUEST` child container for every MCP request: a tool call, a +resource read, a prompt render or a list call. Notifications open nothing. The child is +closed with `close_async()`, so REQUEST-scoped finalizers may be async or sync. There is no +`Scope.SESSION` child: an MCP session has no close hook, so nothing could close it. -A tool of a mounted server resolves through the parent's middleware, so call `setup_di` on the -server that clients connect to. +`setup_di` on the server that clients connect to covers its own components and those of +every server mounted on it. `setup_di` on a mounted server covers only that server's +components, and a `FromDI` parameter on the parent's own tools then fails with a +`ToolError` that names `setup_di`. + +## Root container lifecycle + +`setup_di` opens the root container when the server's lifespan starts and closes it with +`close_async()` when the lifespan ends, which runs APP-scoped finalizers. With +`manage_lifespan=False` it leaves both to another app; see +[Sharing a container with FastAPI](#sharing-a-container-with-fastapi). ## Framework context objects @@ -140,8 +149,11 @@ app.mount("/mcp", mcp_app) ## Background tasks -A tool registered with `task=True` runs in FastMCP's task worker, outside middleware, so there is no -request container for it. `FromDI` raises a `RuntimeError` there that names this case. +A tool registered with `task=True` needs `fastmcp[tasks]` and a `TasksExtension` added with +`mcp.add_extension(...)`. Such a tool runs in FastMCP's task worker, outside middleware, so +there is no request container for it and `FromDI` cannot resolve. The client gets a +`ToolError` with `Failed to resolve dependencies for parameter(s): ...`, and the server logs a +`RuntimeError` that names background tasks. ## Testing @@ -158,6 +170,10 @@ async with fastmcp.Client(mcp) as client: assert result.data == "Hello, world!" ``` +A `FromDI` inside `Annotated` that the server rejects at startup reaches the test as +`RuntimeError: Client failed to connect: ...`, followed by the server's message naming the +parameter. + ## See also - [Testing with overrides](../recipes/testing-overrides.md): swap providers in your tests. diff --git a/docs/integrations/faststream.md b/docs/integrations/faststream.md index 099b41d3..6d20e2cb 100644 --- a/docs/integrations/faststream.md +++ b/docs/integrations/faststream.md @@ -56,7 +56,7 @@ broker = NatsBroker() app = faststream.FastStream(broker) container = Container(groups=[AppGroup]) modern_di_faststream.setup_di(app, container) -container.validate() # after setup_di — its connection providers are now registered +container.validate() # after setup_di: its context provider is now registered @broker.subscriber("orders.in") @@ -66,69 +66,27 @@ async def handle_order( return report.as_dict() ``` -### Several brokers +`setup_di` takes a `faststream.FastStream` or an `AsgiFastStream` app. It registers the +message context provider, installs a middleware on every broker of the app at startup that +builds a `Scope.REQUEST` child container per message, and closes the root container at +shutdown. -`setup_di` covers every broker of the app, not only the first. The DI middleware is installed -by an `on_startup` hook that walks `app.brokers`, so a broker passed to `FastStream(...)` and -one added with `app.add_broker` after `setup_di` are treated the same: - - - - - -```python -app = faststream.FastStream(nats_broker) -modern_di_faststream.setup_di(app, container) -app.add_broker(kafka_broker) # also gets the DI middleware at startup -``` - -A broker created inside an `on_startup` hook is covered as well, since `setup_di` doesn't need a -broker at call time. Hooks run in registration order, so register that hook **before** -calling `setup_di`; otherwise the install step runs first and does not see the broker. If the -app still has no broker when the install step runs, it raises a `RuntimeError` naming both -remedies. - - - - - -```python -app = faststream.FastStream() - - -@app.on_startup -async def attach_broker() -> None: - app.add_broker(NatsBroker(settings.nats_url)) - - -modern_di_faststream.setup_di(app, container) # after the hook, so startup sees the broker -``` +## Scopes -Between `setup_di` and startup no broker carries the middleware yet; see -[Testing](#testing) for the one place that shows. +The integration creates a `Scope.REQUEST` child container for each message a subscriber +receives. REQUEST-scoped providers and their finalizers live for that one message. The +child is closed with `close_async()`, so finalizers may be async or sync. Resolution itself +is synchronous, as everywhere in modern-di. -## Scopes +There is no `Scope.SESSION` for FastStream: message brokers don't have a session concept +comparable to websockets. -The integration creates a `Scope.REQUEST` child container for each message the subscriber receives. REQUEST-scoped providers (and their finalizers) live for the duration of that one message; APP-scoped providers persist for the whole process. At app shutdown, the integration runs `await container.close_async()` on the APP container. +## Root container lifecycle -There is no `Scope.SESSION` for FastStream: message brokers don't have a session concept comparable to websockets. +`setup_di` registers two `on_startup` hooks: one opens the root container and the next +installs the middleware on every broker. An `after_shutdown` hook closes the root with +`close_async()`, which runs APP-scoped finalizers. Because startup reopens the container, +an app that stops and starts again on the same container works. ## Framework context objects @@ -182,22 +140,91 @@ class AppGroup(Group): ## Testing -!!! warning "Pair the test broker with `TestApp` in the same `with` statement" - When testing DI-using subscribers, pair the test broker (`TestNatsBroker`, - etc.) with `TestApp` in the **same** `with` / `async with` statement. - FastStream's `TestBroker` decides whether to run app `on_startup` hooks by - inspecting that statement; `async with TestNatsBroker(broker):` alone - starts the broker without running `on_startup`. The integration installs - its middleware and reopens the container in those hooks, so a `FromDI` - subscriber reached this way has no request container and raises a - `RuntimeError` that names `setup_di` and `TestApp` as the fix. - - - - ```python - async with TestNatsBroker(broker) as br, TestApp(app): - await br.publish(...) - ``` +The middleware is installed and the container reopened by the app's `on_startup` hooks, +and FastStream's `TestApp` is what runs them. Enter the test broker first and `TestApp` +after it, either in one `async with` statement or with `TestApp` nested inside: + +```python +from faststream import TestApp +from faststream.nats import TestNatsBroker + + + +async def test_handle_order() -> None: + async with TestNatsBroker(broker) as test_broker, TestApp(app): + response = await test_broker.request({"order_id": 1}, "orders.in") + assert await response.decode() == {"service": "catalog"} +``` + + + +A test broker without `TestApp` runs no startup hook, so a `FromDI` subscriber finds no +request container and raises a `RuntimeError` that names `setup_di` and `TestApp`. +Entering `TestApp` before the test broker makes the app connect to the real broker. + +## Several brokers + +`setup_di` covers every broker of the app, not only the first. The DI middleware is installed +by an `on_startup` hook that walks `app.brokers`, so a broker passed to `FastStream(...)` and +one added with `app.add_broker` after `setup_di` are treated the same: + + + + + +```python +app = faststream.FastStream(nats_broker) +modern_di_faststream.setup_di(app, container) +app.add_broker(kafka_broker) # also gets the DI middleware at startup +``` + +A broker created inside an `on_startup` hook is covered as well, since `setup_di` doesn't need a +broker at call time. Hooks run in registration order, so register that hook **before** +calling `setup_di`; otherwise the install step runs first and does not see the broker. If the +app still has no broker when the install step runs, it raises a `RuntimeError` naming both +remedies. + + + + + +```python +app = faststream.FastStream() + + +@app.on_startup +async def attach_broker() -> None: + app.add_broker(NatsBroker()) + + +modern_di_faststream.setup_di(app, container) # after the hook, so startup sees the broker +``` + +Between `setup_di` and startup no broker carries the middleware yet; see +[Testing](#testing) for the one place that shows. ## See also diff --git a/docs/integrations/grpc.md b/docs/integrations/grpc.md index 92489f6c..0208344a 100644 --- a/docs/integrations/grpc.md +++ b/docs/integrations/grpc.md @@ -149,17 +149,12 @@ described in [Writing an integration](writing-integrations.md#lifecycle-rules). ### 3. Async server (`grpc.aio`) `DIAioInterceptor` is the async twin: pass it to `grpc.aio.server(...)` and write -`async def` servicer methods (server-streaming methods as `async` generators): +`async def` servicer methods (server-streaming methods as `async` generators). The +`AppGroup` and `container` from the sync example work unchanged: @@ -174,9 +169,9 @@ class GreeterService(greeter_pb2_grpc.GreeterServicer): self, request: greeter_pb2.HelloRequest, context: grpc.aio.ServicerContext, - greeter: typing.Annotated[Greeter, FromDI(Greeter)], + report: typing.Annotated[RpcReport, FromDI(RpcReport)], ) -> greeter_pb2.HelloReply: - return greeter_pb2.HelloReply(message=greeter.greet(request.name)) + return greeter_pb2.HelloReply(message=report.line()) server = grpc.aio.server(interceptors=[DIAioInterceptor(container)]) @@ -201,6 +196,10 @@ including on the error and client-cancellation paths. REQUEST-scoped providers (and their finalizers) live for exactly one RPC. APP-scoped providers persist for the life of the container. +The sync server closes each RPC's child with `close_sync()`, so an async finalizer on a +REQUEST-scoped provider fails the RPC with `StatusCode.UNKNOWN` and a `FinalizerError` in the +details. The aio server closes it with `close_async()`, where async finalizers run. + There is no `Scope.SESSION` for gRPC: a streaming RPC is one method invocation, modelled as a single REQUEST-scoped unit of work. @@ -224,6 +223,10 @@ class AppGroup(Group): caller = providers.Factory(make_caller, scope=Scope.REQUEST) ``` +Annotate the parameter as `grpc.ServicerContext` on both servers. The aio interceptor stores +its context under that type too, and `grpc.aio.ServicerContext` is not a subclass of it, so a +factory parameter annotated `grpc.aio.ServicerContext` does not resolve. + The interceptor's provider has no default, so the context is required for a direct resolve and for a required parameter. Outside an RPC no context is set, and resolving `caller` raises `ContextValueNotSetError`. @@ -247,8 +250,8 @@ argument. ## Root container lifecycle gRPC has no server startup/shutdown hook, so the root container's lifecycle is -yours to own (as with Flask). Create the container open, pass it to the -interceptor, and close it after the server stops to run APP-scoped finalizers: +yours to own (as with Flask). Build the container, pass it to the interceptor, and +close it after the server stops to run APP-scoped finalizers: