docs: Add the python-injection integration guide and Pump.drain

README.md

@@ -20,4 +20,4 @@ Requires Python 3.12 or higher.
 pip install "python-cq[injection]"
 ```
 
-The `[injection]` extra installs [python-injection](https://github.com/100nm/python-injection) as the default DI backend (recommended). To bring your own DI framework, install `python-cq` without the extra and see the [Custom DI adapter](https://python-cq.remimd.dev/routing/di/) guide.
+The `[injection]` extra installs [python-injection](https://github.com/100nm/python-injection) as the default DI backend (recommended). To bring your own DI framework, install `python-cq` without the extra and see the [Custom DI adapter](https://python-cq.remimd.dev/routing/di/) guide.

docs/integrations/injection.md

@@ -0,0 +1,110 @@
+# python-injection
+
+[python-injection](https://github.com/100nm/python-injection) is the default DI backend of **python-cq**. Install it with the `[injection]` extra:
+
+```bash
+pip install "python-cq[injection]"
+```
+
+With the extra installed, the default router uses `cq.ext.injection.InjectionAdapter`, so there is nothing to wire. This page describes what the adapter sets up, and how to tune it.
+
+## What is wired
+
+* **Handlers** are built through the container on every dispatch: each constructor parameter is resolved from its type annotation.
+* **Buses**: `CommandBus`, `EventBus`, and `QueryBus` are registered as fallbacks, so they can be injected right away, and any factory you register for them takes precedence (see [Configuring a bus](../routing/configuring.md)).
+* **Command scope**: each command dispatch opens a python-injection scope named `CQScope.COMMAND_DISPATCH`. `RelatedEvents` is registered in this scope, which is what makes it injectable in command handlers.
+* **Pipelines** resolve the buses from the container each time they dispatch a message.
+
+Everything is registered in the default python-injection module, the one used by the `@injectable`, `@inject`, and `@scoped` decorators imported from `injection`.
+
+## Dependencies scoped to a command
+
+The command scope is a regular python-injection scope, so your own dependencies can live in it too. A dependency declared with `@scoped(CQScope.COMMAND_DISPATCH)` is created once per command dispatch, and shared by everything resolved during that dispatch. A typical use is a transaction shared by several repositories:
+
+```python
+from collections.abc import AsyncIterator
+from cq.ext.injection import CQScope
+from injection import scoped
+
+
+@scoped(CQScope.COMMAND_DISPATCH)
+async def transaction_factory(database: Database) -> AsyncIterator[Transaction]:
+    async with database.transaction() as transaction:
+        yield transaction
+```
+
+Every repository that depends on `Transaction` receives the same one during a command dispatch, so everything the command writes is committed together when its dispatch ends, or rolled back if it raises. The next command gets a new transaction. `Database` and `Transaction` stand for whatever your storage library provides.
+
+Resolving a dependency of the command scope outside a command dispatch raises a python-injection `ScopeUndefinedError`.
+
+!!! warning
+    The command scope belongs to the command handler only. An event handler must not depend on anything scoped to `CQScope.COMMAND_DISPATCH`, even for an event published through `RelatedEvents`. Where and when related events are handled is a configuration choice: they may be [sent to a queue](#sending-related-events-to-a-queue) and handled after the command, or in another process. Even when they are dispatched in-process, they can run while the scope is closing, and resolving the dependency then raises a python-injection `ScopeError`. Put what an event handler needs in the event itself, or give it its own dependencies.
+
+## Sending related events to a queue
+
+By default, related events are dispatched on the `EventBus` before the command dispatch returns. To handle them later instead, register your own `RelatedEvents` in the command scope. It takes precedence over the default one:
+
+```python
+from collections.abc import AsyncIterator
+from cq import AnyIORelatedEvents, RelatedEvents
+from cq.ext.injection import CQScope
+from injection import scoped
+
+
+@scoped(CQScope.COMMAND_DISPATCH)
+async def related_events_factory(
+    queue: EventQueue,
+) -> AsyncIterator[RelatedEvents]:
+    async with AnyIORelatedEvents(queue) as events:
+        yield events
+```
+
+`AnyIORelatedEvents` takes the function that emits each event. A queue's producer is callable, so it fits here. The command dispatch now only waits for the events to be sent, and a [`Pump`](../queuing/pump.md) drains the queue into the `EventBus`. `EventQueue` stands for the queue registered in your container, for example a [`MemoryQueue`](../queuing/queues.md#memoryqueue) subclass or your own transport.
+
+## Custom module
+
+To register everything in another python-injection module than the default one, build a `Router` with your own `InjectionAdapter`, as described in [Custom DI adapter](../routing/di.md#the-router-class):
+
+```python
+from cq import Router
+from cq.ext.injection import InjectionAdapter
+from enum import StrEnum
+from injection import mod
+
+
+class Profile(StrEnum):
+    APP = "app"
+
+
+router = Router(InjectionAdapter(mod(Profile.APP))).register_defaults()
+```
+
+`InjectionAdapter` accepts two parameters:
+
+* `module`: the python-injection module where handlers are resolved and buses are registered. Defaults to the default module.
+* `threadsafe`: forwarded to python-injection whenever the adapter resolves a dependency or opens the command scope. With `True`, the resolution is wrapped in a lock. Defaults to `None`, which leaves the choice to python-injection.
+
+## Opening a scope around other dispatches
+
+`cq.ext.injection.InjectionScopeMiddleware` is the middleware behind the command scope. Add it to any bus, pipeline, or pump to open a python-injection scope around each message:
+
+```python
+from cq import Pump
+from cq.ext.injection import InjectionScopeMiddleware
+from enum import StrEnum
+
+
+class DIScope(StrEnum):
+    MESSAGE = "message"
+
+
+pump = Pump(queue, event_bus).add_middlewares(InjectionScopeMiddleware(DIScope.MESSAGE))
+```
+
+Opening a scope that is already open raises a python-injection `ScopeAlreadyDefinedError`. Pass `exist_ok=True` to reuse the open scope instead, for example when the same middleware can be nested in another dispatch:
+
+```python
+InjectionScopeMiddleware(DIScope.MESSAGE, exist_ok=True)
+```
+
+It also accepts `threadsafe`, with the same meaning as on `InjectionAdapter`.

docs/queuing/pump.md

@@ -30,6 +30,18 @@ async with Pump(queue, event_bus).draining(graceful=True):
     await queue.close()
 ```
 
+## Draining without a context
+
+`draining` suits a pump that runs alongside other code. When consuming the queue is all a process does, as in a dedicated worker, await `drain` directly:
+
+```python
+await Pump(queue, event_bus).drain()
+```
+
+`drain` returns once the consumer's iteration ends, that is when the queue is closed, and runs forever otherwise. `fail_silently` and the pump's middlewares apply as with `draining`. Since there is no task group, an exception comes out of `drain` as is, not inside an `ExceptionGroup`.
+
+`drain` runs a single drain loop. For concurrency, start several of them in your own task group, against a consumer that supports concurrent iteration (see [Concurrency](#concurrency)).
+
 ## `MemoryQueue.draining`
 
 For a `MemoryQueue`, `draining` combines both steps: it opens a graceful pump, yields the queue, and closes it on exit, so the queue cannot be reused afterwards. When the block ends, every message sent inside it has been processed:

docs/routing/di.md

@@ -1,7 +1,7 @@
 # Custom DI adapter
 
 !!! note
-    If you installed `python-cq[injection]`, you can skip this page. The integration with [python-injection](https://github.com/100nm/python-injection) is already wired up. This page is only useful to plug in a different DI framework.
+    If you installed `python-cq[injection]`, you can skip this page. The integration with [python-injection](https://github.com/100nm/python-injection) is already wired up (see [python-injection](../integrations/injection.md)). This page is only useful to plug in a different DI framework.
 
 **python-cq** does not depend on a specific dependency injection container. It talks to DI through the `DIAdapter` protocol: implement it once and the library works with the container you already use.
 
@@ -27,6 +27,8 @@ new_query_bus = router.new_query_bus
 
 `register_defaults()` hands the bus factories to the adapter so that `CommandBus`, `EventBus`, and `QueryBus` can be resolved from the container. Call it once per router.
 
+A router also lists the message types its handlers are registered for, through `command_types`, `event_types`, and `query_types`.
+
 Pipelines follow the same rule. Build them from your router so they resolve the buses through your adapter:
 
 ```python
@@ -63,8 +65,9 @@ class MyDIAdapter(DIAdapter):
     def lazy[T](self, tp: type[T]) -> Callable[[], Awaitable[T]]:
         """
         Return a callable that, when called and awaited, resolves `tp` from
-        the container. Pipelines use it to resolve the buses on first use,
-        so that buses registered later in the import graph are picked up.
+        the container. Pipelines call the returned callable on every
+        message they dispatch, so that buses registered later in the import
+        graph are picked up.
         """
         ...
 

docs/routing/multiple_commands.md

@@ -71,7 +71,7 @@ class PaymentContext:
 
 Annotate the pipeline with `ClassVar`: if the context is a dataclass, a plain annotation would turn the pipeline into a field.
 
-The pipeline resolves `CommandBus` and `QueryBus` from the DI container the first time it dispatches, so the buses you configure in your container are the ones it uses. If you manage your own `Router` (see [Custom DI adapter](di.md)), build the pipeline with `router.command_pipeline()` instead.
+The pipeline resolves `CommandBus` and `QueryBus` from the DI container each time it dispatches a message, not when it is created, so the buses you configure in your container are the ones it uses. With a bus factory registered as `@injectable`, each message is therefore dispatched by a new bus. If you manage your own `Router` (see [Custom DI adapter](di.md)), build the pipeline with `router.command_pipeline()` instead.
 
 ### Steps
 

mkdocs.yml

@@ -27,6 +27,8 @@ nav:
   - Queuing:
       - Queues: queuing/queues.md
       - Pump: queuing/pump.md
+  - Integrations:
+      - python-injection: integrations/injection.md
 
 plugins:
   - search