Dependency injection

PyMediate can use a dependency-injector container as its service provider. Install the optional integration first:

pip install 'pymediate[di]'

The core package does not require this integration. For direct instance registration, use Services.

Connect a container

DependencyInjectorServiceProvider implements PyMediate's ServiceProvider protocol. It indexes the container's declared providers by their inferred output type, then delegates each resolution back to the original Dependency Injector provider:

from dependency_injector import containers, providers

from pymediate import Mediator
from pymediate.providers import DependencyInjectorServiceProvider


class ApplicationContainer(containers.DeclarativeContainer):
    order_store = providers.Singleton(PostgresOrderStore)
    place_order = providers.Factory(
        PlaceOrderHandler,
        orders=order_store,
    )


container = ApplicationContainer()
services = DependencyInjectorServiceProvider(container)
mediator = Mediator(services)

receipt = await mediator.send(
    PlaceOrder(customer_id=7, item="tea", quantity=2),
)

The provider name place_order has no routing meaning. The adapter infers PlaceOrderHandler from providers.Factory(PlaceOrderHandler, ...), and the mediator requests that type when it dispatches PlaceOrder.

Indexing does not construct the handler or its dependencies. Construction happens when the mediator asks the provider for an instance.

Keep service construction synchronous

Service resolution is synchronous for both mediator variants. An asynchronous request handler is constructed synchronously, then its async def __call__ runs asynchronously during send().

The adapter rejects a providers.Coroutine while indexing. It also raises TypeError if any provider returns an awaitable during resolution, including a provider placed in Dependency Injector's asynchronous mode.

This distinction is valid:

# Synchronous construction; asynchronous work remains in the handler method.
place_order = providers.Factory(PlaceOrderHandler, orders=order_store)

An asynchronous factory for PlaceOrderHandler is not a supported service source.

Understand which providers can be indexed

The adapter must determine each service type without calling its provider. It can index:

• class-backed Factory, singleton, and Callable providers;
• function-backed providers with a concrete return annotation;
• Object, List, and Dict providers, which have intrinsic output types;
• declared child Container providers, which it visits recursively.

The constructor expects a DeclarativeContainer or DynamicContainer instance. Passing another object raises TypeError before indexing begins.

A function-backed factory therefore needs a concrete return annotation:

def build_place_order_handler(
    orders: OrderStore,
) -> PlaceOrderHandler:
    return PlaceOrderHandler(orders)


place_order = providers.Factory(
    build_place_order_handler,
    orders=order_store,
)

An unannotated factory, Selector, Resource, or another provider whose output type is opaque is skipped — it never becomes a service, and the provider is not called to discover its type. That lets a container mix PyMediate handlers with infrastructure providers (a database Resource, a dynamic Selector) freely; only the providers whose type can be inferred are indexed.

Configuration, dependency placeholders, dependency containers, and providers.Self() are composition inputs rather than exposed services, so the adapter skips them too.

Use nested containers

The adapter walks the whole provider graph with Container.traverse(), so services in nested providers.Container children are discovered alongside the root container's. Pipeline-behavior order does not depend on this - it is fixed by the mediator's behaviors= sequence, not by how the container resolves them:

class OrdersContainer(containers.DeclarativeContainer):
    order_store = providers.Dependency()
    place_order = providers.Factory(
        PlaceOrderHandler,
        orders=order_store,
    )


class ApplicationContainer(containers.DeclarativeContainer):
    request_logging = providers.Singleton(RequestLogging)
    order_store = providers.Singleton(PostgresOrderStore)
    orders = providers.Container(
        OrdersContainer,
        order_store=order_store,
    )

traverse() is cycle-safe, so a cycle between child containers is walked without error. Because it follows the whole graph, a provider reachable only as an injected argument is indexed as a service too - resolvable by its exact type like any other.

Choose provider lifetimes

Dependency Injector retains control of each registered lifetime:

• Factory constructs a handler or behavior each time the mediator resolves it.
• Singleton reuses one instance from that container.
• ContextLocalSingleton follows Dependency Injector's context-local lifetime.

The mediator resolves a request or stream handler for each dispatch. It resolves applicable pipeline behaviors for each send(), and resolves notification subscribers for each publish(). Choose a lifetime that matches the state and concurrency rules of that service rather than relying on a handler-wide default.

Compose the mediator inside the container

Dependency Injector's providers.Self() pattern lets the container construct its own service provider and mediator without recursive indexing:

class ApplicationContainer(containers.DeclarativeContainer):
    __self__ = providers.Self()

    order_store = providers.Singleton(PostgresOrderStore)
    place_order = providers.Factory(
        PlaceOrderHandler,
        orders=order_store,
    )
    services = providers.Singleton(
        DependencyInjectorServiceProvider,
        __self__,
    )
    mediator = providers.Singleton(Mediator, services=services)


mediator = ApplicationContainer().mediator()

Constructing the mediator outside the container is also valid. Both arrangements pass the same ServiceProvider interface to Mediator.

Rebuild the index after type changes

The adapter's provider graph and inferred types are a construction-time snapshot. Overrides that preserve a provider's output type continue to work because resolution remains delegated to the original provider. If an override changes the output type, construct a new DependencyInjectorServiceProvider after applying it.

If a provider produces a value that no longer matches its indexed type, resolution raises TypeError and asks for the adapter to be rebuilt.

Tests can use ordinary Dependency Injector overrides:

container = ApplicationContainer()
container.order_store.override(providers.Singleton(InMemoryOrderStore))

mediator = Mediator(DependencyInjectorServiceProvider(container))
receipt = await mediator.send(
    PlaceOrder(customer_id=7, item="tea", quantity=2),
)

For a unit test that does not exercise container wiring, construct the handler directly. See the testing guide.

Continue

• Review handler state and dependencies in request handlers.
• Check resolution semantics in the ServiceProvider reference.
• Diagnose construction and resolution failures with error handling.