ServiceProvider and Services

The mediator obtains handlers and pipeline behaviors from a ServiceProvider. PyMediate includes Services for registering existing instances and an optional adapter for Dependency Injector.

ServiceProvider (protocol)

ServiceProvider is a structural protocol. It is read-only and describes the two operations the mediator uses to resolve handlers and behaviors:

from typing import Protocol

class ServiceProvider(Protocol):
    def __getitem__[ServiceT](self, service_type: type[ServiceT]) -> ServiceT: ...
    def __contains__(self, service_type: type) -> bool: ...

The mediator accepts custom implementations of this protocol through its services constructor parameter. Because it is a structural protocol, a class conforms just by defining these two operations — inheriting from ServiceProvider is optional. The built-in providers do inherit it so a static type checker verifies conformance at their definition. They also offer len() as a convenience for inspection and testing; it is not part of the protocol the mediator requires.

Services

from pymediate import ServiceProvider

class Services(ServiceProvider):
    def __init__(self, *instances: object) -> None: ...
    def __getitem__[ServiceT](self, service_type: type[ServiceT]) -> ServiceT: ...
    def __contains__(self, service_type: type) -> bool: ...
    def __or__(self, other: Services) -> Services: ...
    def __len__(self) -> int: ...

Services is the built-in ServiceProvider. It is constructed from its instances and is immutable afterwards, so there is no separate build step — pass it straight to a Mediator:

from pymediate import Mediator, Services

services = Services(PlaceOrderHandler())
mediator = Mediator(services=services)

Services(*instances)

Each instance is registered under its concrete type, type(instance). Passing None raises ValueError. Two instances of the same type raise ServiceAlreadyRegisteredError: only one instance per type is resolvable, so a repeat is a wiring mistake rather than a preference. Build the argument list first when the set of services is computed:

services = Services(*handlers, *behaviors)

services[Type] and Type in services

Resolution is by exact type. A request for a base class does not match a registered subclass, and a missing type raises ServiceNotFoundError. len(services) is the number of registered instances.

left | right

Combining two collections returns a new one holding both operands' services, with the right operand winning on any shared type. Neither operand is modified. Unlike the constructor, a shared type is allowed here — replacing a service is this operator's purpose, which makes swapping a fake into an existing wiring explicit:

from pymediate import Mediator, Services

# The application's own wiring, imported rather than duplicated in the test.
from myapp.wiring import build_services

for_tests = build_services() | Services(PlaceOrderHandler(FakeOrderStore()))
mediator = Mediator(for_tests)

Combining with anything that is not a Services raises TypeError. There is no in-place variant: services |= other rebinds the name to a new collection, as it does for a tuple.

DependencyInjectorServiceProvider

from typing import Any

from dependency_injector import containers

from pymediate import ServiceProvider

class DependencyInjectorServiceProvider(ServiceProvider):
    def __init__(self, container: containers.Container) -> None: ...
    def __getitem__[ServiceT](self, service_type: type[ServiceT]) -> ServiceT: ...
    def __contains__(self, service_type: type[Any]) -> bool: ...
    def __len__(self) -> int: ...

Install the di extra to use the adapter:

pip install 'pymediate[di]'

from dependency_injector import containers, providers

from pymediate import Mediator
from pymediate.providers import DependencyInjectorServiceProvider


class AppContainer(containers.DeclarativeContainer):
    place_order_handler = providers.Factory(PlaceOrderHandler)


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

The adapter indexes the container's providers without resolving them, then delegates each resolution to the original Dependency Injector provider. It implements the two ServiceProvider operations and, like the built-in provider, offers len() — len(services) is the number of indexed providers.

It can infer service types from class-backed factories and singletons, object providers, list and dictionary providers, and callables with a concrete return annotation. The whole provider graph is walked with Container.traverse() — nested providers.Container children and providers reachable only through injection are all visited. A provider whose output type cannot be inferred without resolving it (an unannotated factory, Selector, Resource, or coroutine provider) is skipped, not indexed, so infrastructure providers need not be PyMediate-resolvable. provider[Type] returns the first provider found for an exact type.

The indexed provider graph and output types form a construction-time snapshot. Overrides that preserve output types continue to resolve through the original provider.

ServiceNotFoundError

class ServiceNotFoundError(KeyError):
    def __init__(
        self,
        service_type: type,
        available_types: list[type],
    ) -> None: ...

provider[Type] raises ServiceNotFoundError when the exact requested type is absent. Its service_type and available_types attributes retain the constructor arguments. This exception inherits from KeyError — the miss surfaces through the provider[Type] subscript, so except KeyError catches it — not from PyMediateError.

See also

• Mediator — consumes a service provider
• Dependency injection — composition and container lifetimes
• Errors — service-resolution failures during dispatch