From ecfe48bcaa71e3f7dbdcb470b3d82d2a2cd117ac Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 9 Jul 2026 20:05:00 +0300 Subject: [PATCH 1/2] docs: add taskiq integration usage page --- docs/integrations/taskiq.md | 145 ++++++++++++++++++++++++++++++++++++ mkdocs.yml | 2 + 2 files changed, 147 insertions(+) create mode 100644 docs/integrations/taskiq.md diff --git a/docs/integrations/taskiq.md b/docs/integrations/taskiq.md new file mode 100644 index 0000000..ce2cd44 --- /dev/null +++ b/docs/integrations/taskiq.md @@ -0,0 +1,145 @@ +# Usage with `taskiq` + +## How to use + +### 1. Install `modern-di-taskiq` + +=== "uv" + + ```bash + uv add modern-di-taskiq + ``` + +=== "pip" + + ```bash + pip install modern-di-taskiq + ``` + +=== "poetry" + + ```bash + poetry add modern-di-taskiq + ``` + +### 2. Apply to your application + +```python +import dataclasses +import typing + +from modern_di import Container, Group, Scope, providers +from modern_di_taskiq import FromDI, setup_di +from taskiq import InMemoryBroker + + +@dataclasses.dataclass(kw_only=True, slots=True, frozen=True) +class Settings: + greeting: str = "hello" + + +@dataclasses.dataclass(kw_only=True, slots=True) +class Greeter: + settings: Settings # auto-injected by type + + def greet(self, name: str) -> str: + return f"{self.settings.greeting}, {name}" + + +class AppGroup(Group): + settings = providers.Factory( + Settings, + scope=Scope.APP, + cache=True, + ) + greeter = providers.Factory( + Greeter, + scope=Scope.REQUEST, + ) + + +broker = InMemoryBroker() +setup_di(broker, Container(groups=[AppGroup], validate=True)) + + +@broker.task +async def greet( + name: str, + greeter: typing.Annotated[ + Greeter, + FromDI(Greeter), # resolve by type + ], +) -> str: + return greeter.greet(name) +``` + +`setup_di(broker, container)` stores the container on `broker.state` and registers `TaskiqEvents.WORKER_STARTUP`/`WORKER_SHUTDOWN` handlers that open/close it — those fire when the broker's worker process starts and stops, so a script that just calls tasks directly (like `InMemoryBroker` in a test) must drive the container lifecycle itself, e.g. `async with broker: ...` or an explicit `container.open()` / `await container.close_async()`. + +## Scopes + +The integration creates a `Scope.REQUEST` child container **for each task** the worker executes. REQUEST-scoped providers (and their finalizers) live for the duration of that one task — the child container is closed after the task returns, including when it raises. APP-scoped providers persist for the whole worker process; `setup_di` opens the APP container on `WORKER_STARTUP` and runs `await container.close_async()` on `WORKER_SHUTDOWN`. + +There is no `Scope.SESSION` for taskiq — a task queue doesn't have a session concept comparable to websockets. + +## Sync resolution, async cleanup + +`FromDI` resolves its dependency with `Container.resolve_dependency(...)`, which is synchronous — modern-di's resolution is always sync, regardless of the framework. The per-task `Scope.REQUEST` child container that resolution runs against is nevertheless torn down asynchronously: after the task handler finishes (or raises), the integration awaits `container.close_async()` on it. So async finalizers on REQUEST-scoped providers run correctly, while the factories themselves must build synchronously. + +## Framework context objects + +`taskiq.TaskiqMessage` is automatically made available by the integration, so factories can declare it as a parameter and get the message that triggered the current task — see [Framework Context Objects](../providers/context.md#framework-context-objects) for how implicit and explicit resolution work. + +The following context provider is also available for explicit import: + +- `taskiq_message_provider` — provides the current `taskiq.TaskiqMessage` object. + +### Implicit (type-based) usage + +```python +import taskiq +from modern_di import Group, Scope, providers + + +def create_task_info(message: taskiq.TaskiqMessage) -> dict[str, str]: + return { + "task_id": message.task_id, + "task_name": message.task_name, + } + + +class AppGroup(Group): + # The message dependency is resolved by type annotation + task_info = providers.Factory( + create_task_info, + scope=Scope.REQUEST, + ) +``` + +### Explicit (provider-based) usage + +```python +import taskiq +import modern_di_taskiq +from modern_di import Group, Scope, providers + + +def create_task_info(message: taskiq.TaskiqMessage) -> dict[str, str]: + return {"task_id": message.task_id} + + +class AppGroup(Group): + task_info = providers.Factory( + create_task_info, + scope=Scope.REQUEST, + kwargs={"message": modern_di_taskiq.taskiq_message_provider}, + ) +``` + +## API + +| Symbol | Description | +|---|---| +| `setup_di(broker, container)` | Wire the APP-scope container into taskiq — creates a REQUEST child container per task and opens/closes the APP container on worker startup/shutdown. | +| `FromDI(provider_or_type)` | Marker for `Annotated[T, FromDI(...)]` in task signatures; accepts a provider instance or a plain type. | +| `fetch_di_container(broker)` | Returns the APP-scope container registered with the taskiq broker. | +| `taskiq_message_provider` | `ContextProvider` for the current `taskiq.TaskiqMessage`. | diff --git a/mkdocs.yml b/mkdocs.yml index dd2d812..cd5d8b3 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -24,6 +24,7 @@ nav: - aiohttp: integrations/aiohttp.md - FastAPI: integrations/fastapi.md - FastStream: integrations/faststream.md + - taskiq: integrations/taskiq.md - Litestar: integrations/litestar.md - Starlette: integrations/starlette.md - Typer: integrations/typer.md @@ -161,6 +162,7 @@ plugins: - integrations/aiohttp.md: Usage with aiohttp - integrations/fastapi.md: Usage with FastAPI - integrations/faststream.md: Usage with FastStream + - integrations/taskiq.md: Usage with taskiq - integrations/litestar.md: Usage with Litestar - integrations/starlette.md: Usage with Starlette - integrations/typer.md: Usage with Typer From cb90f730e76eb7aef254a19326ceb089f5b894d3 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 9 Jul 2026 20:09:58 +0300 Subject: [PATCH 2/2] docs: order taskiq nav entry alphabetically among framework integrations --- mkdocs.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/mkdocs.yml b/mkdocs.yml index cd5d8b3..b385325 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -24,9 +24,9 @@ nav: - aiohttp: integrations/aiohttp.md - FastAPI: integrations/fastapi.md - FastStream: integrations/faststream.md - - taskiq: integrations/taskiq.md - Litestar: integrations/litestar.md - Starlette: integrations/starlette.md + - taskiq: integrations/taskiq.md - Typer: integrations/typer.md - Pytest: integrations/pytest.md - Writing an integration: integrations/writing-integrations.md @@ -162,9 +162,9 @@ plugins: - integrations/aiohttp.md: Usage with aiohttp - integrations/fastapi.md: Usage with FastAPI - integrations/faststream.md: Usage with FastStream - - integrations/taskiq.md: Usage with taskiq - integrations/litestar.md: Usage with Litestar - integrations/starlette.md: Usage with Starlette + - integrations/taskiq.md: Usage with taskiq - integrations/typer.md: Usage with Typer - integrations/pytest.md: Usage with pytest, with and without modern-di-pytest - integrations/writing-integrations.md: Specification for building a new framework integration