waku ¶

WakuApplication ¶

WakuApplication(
    *,
    container: AsyncContainer,
    registry: ModuleRegistry,
    lifespan: Sequence[LifespanFunc | LifespanWrapper],
    extension_registry: ExtensionRegistry,
)

Source code in src/waku/application.py

def __init__(
    self,
    *,
    container: AsyncContainer,
    registry: ModuleRegistry,
    lifespan: Sequence[LifespanFunc | LifespanWrapper],
    extension_registry: ExtensionRegistry,
) -> None:
    self._container = container
    self._registry = registry
    self._lifespan = tuple(
        LifespanWrapper(lifespan_func) if not isinstance(lifespan_func, LifespanWrapper) else lifespan_func
        for lifespan_func in lifespan
    )
    self._extension_registry = extension_registry

    self._exit_stack = AsyncExitStack()
    self._initialized = False

container `property` ¶

container: AsyncContainer

registry `property` ¶

registry: ModuleRegistry

initialize `async` ¶

initialize() -> None

Source code in src/waku/application.py

async def initialize(self) -> None:
    if self._initialized:
        return
    await self._call_on_init_extensions()
    self._initialized = True
    await self._call_after_init_extensions()

close `async` ¶

close() -> None

Source code in src/waku/application.py

async def close(self) -> None:
    if not self._initialized:
        return
    await self._call_on_shutdown_extensions()
    self._initialized = False

WakuFactory ¶

WakuFactory(
    root_module_type: ModuleType,
    /,
    context: dict[Any, Any] | None = None,
    lifespan: Sequence[LifespanFunc] = (),
    extensions: Sequence[
        ApplicationExtension
    ] = DEFAULT_EXTENSIONS,
    container_config: ContainerConfig | None = None,
    provider_filter: IProviderFilter | None = None,
)

Source code in src/waku/factory.py

def __init__(
    self,
    root_module_type: ModuleType,
    /,
    context: dict[Any, Any] | None = None,
    lifespan: Sequence[LifespanFunc] = (),
    extensions: Sequence[ApplicationExtension] = DEFAULT_EXTENSIONS,
    container_config: ContainerConfig | None = None,
    provider_filter: IProviderFilter | None = None,
) -> None:
    self._root_module_type = root_module_type

    self._context = context
    self._lifespan = lifespan
    self._extensions = extensions
    self._container_config = container_config or ContainerConfig()
    self._provider_filter = provider_filter

create ¶

create() -> WakuApplication

Source code in src/waku/factory.py

def create(self) -> WakuApplication:
    registry = ModuleRegistryBuilder(
        self._root_module_type,
        context=self._context,
        provider_filter=self._provider_filter,
        app_extensions=self._extensions,
    ).build()

    container = self._build_container(registry.providers)
    return WakuApplication(
        container=container,
        registry=registry,
        lifespan=self._lifespan,
        extension_registry=self._build_extension_registry(registry.modules),
    )

DynamicModule `dataclass` ¶

DynamicModule(
    *,
    providers: list[ProviderSpec] = list(),
    imports: list[ModuleType | DynamicModule] = list(),
    exports: list[
        type[object] | ModuleType | DynamicModule
    ] = list(),
    extensions: list[ModuleExtension] = list(),
    is_global: bool = False,
    id: UUID = uuid4(),
    parent_module: ModuleType,
)

Bases: ModuleMetadata

providers `class-attribute` `instance-attribute` ¶

providers: list[ProviderSpec] = field(default_factory=list)

List of providers for dependency injection.

imports `class-attribute` `instance-attribute` ¶

imports: list[ModuleType | DynamicModule] = field(
    default_factory=list
)

List of modules imported by this module.

exports `class-attribute` `instance-attribute` ¶

exports: list[type[object] | ModuleType | DynamicModule] = (
    field(default_factory=list)
)

List of types or modules exported by this module.

extensions `class-attribute` `instance-attribute` ¶

extensions: list[ModuleExtension] = field(
    default_factory=list
)

List of module extensions for lifecycle hooks.

is_global `class-attribute` `instance-attribute` ¶

is_global: bool = False

Whether this module is global or not.

id `class-attribute` `instance-attribute` ¶

id: UUID = field(default_factory=uuid4)

parent_module `instance-attribute` ¶

parent_module: ModuleType

Module ¶

Module(module_type: ModuleType, metadata: ModuleMetadata)

Source code in src/waku/modules/_module.py

def __init__(self, module_type: ModuleType, metadata: ModuleMetadata) -> None:
    self.id: Final[UUID] = metadata.id
    self.target: Final[ModuleType] = module_type

    self.providers: Final[Sequence[ProviderSpec]] = metadata.providers
    self.imports: Final[Sequence[ModuleType | DynamicModule]] = metadata.imports
    self.exports: Final[Sequence[type[object] | ModuleType | DynamicModule]] = metadata.exports
    self.extensions: Final[Sequence[ModuleExtension]] = metadata.extensions
    self.is_global: Final[bool] = metadata.is_global

    self._provider: BaseProvider | None = None

id `instance-attribute` ¶

id: Final[UUID] = id

target `instance-attribute` ¶

target: Final[ModuleType] = module_type

providers `instance-attribute` ¶

providers: Final[Sequence[ProviderSpec]] = providers

imports `instance-attribute` ¶

imports: Final[Sequence[ModuleType | DynamicModule]] = (
    imports
)

exports `instance-attribute` ¶

exports: Final[
    Sequence[type[object] | ModuleType | DynamicModule]
] = exports

extensions `instance-attribute` ¶

extensions: Final[Sequence[ModuleExtension]] = extensions

is_global `instance-attribute` ¶

is_global: Final[bool] = is_global

name `property` ¶

name: str

provider `property` ¶

provider: BaseProvider

Get the aggregated provider for this module.

This property returns the provider created by create_provider(). Must be called after create_provider() has been invoked.

RAISES	DESCRIPTION
`RuntimeError`	If create_provider() has not been called yet.

create_provider ¶

create_provider(
    context: dict[Any, Any] | None,
    builder: ActivationBuilder,
    provider_filter: IProviderFilter,
) -> BaseProvider

Create aggregated provider with activation filtering applied.

PARAMETER	DESCRIPTION
`context`	Context dict for activation decisions. TYPE: `dict[Any, Any] \| None`
`builder`	Activation builder for checking if types are registered. TYPE: `ActivationBuilder`
`provider_filter`	Filter strategy for conditional provider activation. TYPE: `IProviderFilter`

RETURNS	DESCRIPTION
`BaseProvider`	BaseProvider with only active providers aggregated.

Source code in src/waku/modules/_module.py

def create_provider(
    self,
    context: dict[Any, Any] | None,
    builder: ActivationBuilder,
    provider_filter: IProviderFilter,
) -> BaseProvider:
    """Create aggregated provider with activation filtering applied.

    Args:
        context: Context dict for activation decisions.
        builder: Activation builder for checking if types are registered.
        provider_filter: Filter strategy for conditional provider activation.

    Returns:
        BaseProvider with only active providers aggregated.
    """
    active_providers = provider_filter.filter(
        list(self.providers),
        context=context,
        module_type=self.target,
        builder=builder,
    )

    cls = cast('type[_ModuleProvider]', type(f'{self.name}Provider', (_ModuleProvider,), {}))
    self._provider = cls(active_providers)
    return self._provider

module ¶

module(
    *,
    providers: Sequence[ProviderSpec] = (),
    imports: Sequence[ModuleType | DynamicModule] = (),
    exports: Sequence[
        type[object] | ModuleType | DynamicModule
    ] = (),
    extensions: Sequence[ModuleExtension] = (),
    is_global: bool = False,
) -> Callable[[type[_T]], type[_T]]

Decorator to define a module.

PARAMETER	DESCRIPTION
`providers`	Sequence of providers for dependency injection. TYPE: `Sequence[ProviderSpec]` DEFAULT: `()`
`imports`	Sequence of modules imported by this module. TYPE: `Sequence[ModuleType \| DynamicModule]` DEFAULT: `()`
`exports`	Sequence of types or modules exported by this module. TYPE: `Sequence[type[object] \| ModuleType \| DynamicModule]` DEFAULT: `()`
`extensions`	Sequence of module extensions for lifecycle hooks. TYPE: `Sequence[ModuleExtension]` DEFAULT: `()`
`is_global`	Whether this module is global or not. TYPE: `bool` DEFAULT: `False`

Source code in src/waku/modules/_metadata.py

def module(
    *,
    providers: Sequence[ProviderSpec] = (),
    imports: Sequence[ModuleType | DynamicModule] = (),
    exports: Sequence[type[object] | ModuleType | DynamicModule] = (),
    extensions: Sequence[ModuleExtension] = (),
    is_global: bool = False,
) -> Callable[[type[_T]], type[_T]]:
    """Decorator to define a module.

    Args:
        providers: Sequence of providers for dependency injection.
        imports: Sequence of modules imported by this module.
        exports: Sequence of types or modules exported by this module.
        extensions: Sequence of module extensions for lifecycle hooks.
        is_global: Whether this module is global or not.
    """

    def decorator(cls: type[_T]) -> type[_T]:
        metadata = ModuleMetadata(
            providers=list(providers),
            imports=list(imports),
            exports=list(exports),
            extensions=list(extensions),
            is_global=is_global,
        )
        for extension in metadata.extensions:
            if isinstance(extension, OnModuleConfigure):
                extension.on_module_configure(metadata)

        setattr(cls, _MODULE_METADATA_KEY, metadata)
        return cls

    return decorator

application ¶

WakuApplication ¶

WakuApplication(
    *,
    container: AsyncContainer,
    registry: ModuleRegistry,
    lifespan: Sequence[LifespanFunc | LifespanWrapper],
    extension_registry: ExtensionRegistry,
)

Source code in src/waku/application.py

def __init__(
    self,
    *,
    container: AsyncContainer,
    registry: ModuleRegistry,
    lifespan: Sequence[LifespanFunc | LifespanWrapper],
    extension_registry: ExtensionRegistry,
) -> None:
    self._container = container
    self._registry = registry
    self._lifespan = tuple(
        LifespanWrapper(lifespan_func) if not isinstance(lifespan_func, LifespanWrapper) else lifespan_func
        for lifespan_func in lifespan
    )
    self._extension_registry = extension_registry

    self._exit_stack = AsyncExitStack()
    self._initialized = False

container `property` ¶

container: AsyncContainer

registry `property` ¶

registry: ModuleRegistry

initialize `async` ¶

initialize() -> None

Source code in src/waku/application.py

async def initialize(self) -> None:
    if self._initialized:
        return
    await self._call_on_init_extensions()
    self._initialized = True
    await self._call_after_init_extensions()

close `async` ¶

close() -> None

Source code in src/waku/application.py

async def close(self) -> None:
    if not self._initialized:
        return
    await self._call_on_shutdown_extensions()
    self._initialized = False

cqrs ¶

NextHandlerType `module-attribute` ¶

NextHandlerType: TypeAlias = Callable[
    [RequestT], Awaitable[ResponseT]
]

Event `dataclass` ¶

Event(*, event_id: UUID = uuid4())

Bases: INotification

Convenience base class for events with auto-generated ID.

Use this class when you want automatic event_id generation. For custom identification strategies, implement INotification directly.

Example::

@dataclass(frozen=True, kw_only=True)
class UserCreated(Event):
    user_id: str
    email: str

event_id `class-attribute` `instance-attribute` ¶

event_id: UUID = field(default_factory=uuid4)

INotification ¶

Bases: Protocol

Marker interface for notification-type objects (events).

This is a pure marker protocol with no required attributes or methods. Implement this protocol for domain events that need custom identification strategies or no identification at all.

MediatR equivalent: INotification

Example::

@dataclass(frozen=True)
class OrderPlaced(INotification):
    order_id: str
    customer_id: str


# Or with custom identification:
@dataclass(frozen=True)
class DomainEvent(INotification):
    aggregate_id: str
    occurred_at: datetime


@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: str

IPipelineBehavior ¶

Bases: ABC, Generic[RequestT, ResponseT]

Interface for pipeline behaviors that wrap request handling.

handle `abstractmethod` `async` ¶

handle(
    request: RequestT,
    /,
    next_handler: NextHandlerType[RequestT, ResponseT],
) -> ResponseT

Handle the request and call the next handler in the pipeline.

PARAMETER	DESCRIPTION
`request`	The request to handle TYPE: `RequestT`
`next_handler`	Function to call the next handler in the pipeline TYPE: `NextHandlerType[RequestT, ResponseT]`

RETURNS	DESCRIPTION
`ResponseT`	The response from the pipeline

Source code in src/waku/cqrs/contracts/pipeline.py

@abstractmethod
async def handle(self, request: RequestT, /, next_handler: NextHandlerType[RequestT, ResponseT]) -> ResponseT:
    """Handle the request and call the next handler in the pipeline.

    Args:
        request: The request to handle
        next_handler: Function to call the next handler in the pipeline

    Returns:
        The response from the pipeline
    """
    ...

IRequest ¶

Bases: Protocol[ResponseT]

Marker interface for request-type objects (commands/queries).

This is a pure marker protocol with no required attributes or methods. Implement this protocol for requests that need custom identification strategies or no identification at all.

MediatR equivalent: IRequest

Example::

@dataclass(frozen=True)
class GetUserQuery(IRequest[UserDTO]):
    user_id: str


@dataclass(frozen=True)
class CreateOrderCommand(IRequest[OrderId]):
    customer_id: str
    items: list[OrderItem]

Request `dataclass` ¶

Request(*, request_id: UUID = uuid4())

Bases: IRequest[ResponseT], Generic[ResponseT]

Convenience base class for requests with auto-generated ID.

Use this class when you want automatic request_id generation. For custom identification strategies, implement IRequest directly.

Example::

@dataclass(frozen=True, kw_only=True)
class GetUserQuery(Request[UserDTO]):
    user_id: str

request_id `class-attribute` `instance-attribute` ¶

request_id: UUID = field(default_factory=uuid4)

Response `dataclass` ¶

Response()

Base class for response type objects.

EventHandler ¶

Bases: INotificationHandler[NotificationT], ABC, Generic[NotificationT]

Abstract base class for event handlers.

Use this class when you want explicit ABC inheritance and type checking. For structural subtyping, implement INotificationHandler directly.

Example::

class UserJoinedEventHandler(EventHandler[UserJoinedEvent]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, event: UserJoinedEvent, /) -> None:
        await self._meetings_api.notify_room(event.meeting_id, 'New user joined!')

handle `abstractmethod` `async` ¶

handle(event: NotificationT) -> None

Source code in src/waku/cqrs/events/handler.py

@abc.abstractmethod
async def handle(self, event: NotificationT, /) -> None:
    raise NotImplementedError

INotificationHandler ¶

Bases: Protocol[NotificationT]

Protocol for notification/event handlers.

MediatR equivalent: INotificationHandler

This protocol allows structural subtyping - any class with a matching handle method signature is compatible.

Example::

class OrderPlacedHandler(INotificationHandler[OrderPlaced]):
    async def handle(self, event: OrderPlaced, /) -> None:
        await self._send_confirmation_email(event.order_id)

handle `async` ¶

handle(event: NotificationT) -> None

Handle the notification/event.

Source code in src/waku/cqrs/events/handler.py

async def handle(self, event: NotificationT, /) -> None:
    """Handle the notification/event."""
    ...

Mediator ¶

Mediator(
    container: AsyncContainer,
    event_publisher: EventPublisher,
    request_map: RequestMap,
    event_map: EventMap,
    behavior_map: PipelineBehaviorMap,
)

Bases: IMediator

Default CQRS implementation.

Source code in src/waku/cqrs/impl.py

def __init__(
    self,
    container: AsyncContainer,
    event_publisher: EventPublisher,
    request_map: RequestMap,
    event_map: EventMap,
    behavior_map: PipelineBehaviorMap,
) -> None:
    self._container = container
    self._event_publisher = event_publisher
    self._request_map = request_map
    self._event_map = event_map
    self._behavior_map = behavior_map

send `async` ¶

send(request: IRequest[None]) -> None

send(request: IRequest[ResponseT]) -> ResponseT

send(request: IRequest[Any]) -> Any

Source code in src/waku/cqrs/impl.py

@override
async def send(self, request: IRequest[Any], /) -> Any:
    request_type = type(request)
    handler = await self._resolve_request_handler(request_type)  # pyrefly: ignore[bad-argument-type]
    return await self._handle_request(handler, request)

publish `async` ¶

publish(notification: INotification) -> None

Source code in src/waku/cqrs/impl.py

@override
async def publish(self, notification: INotification, /) -> None:
    event_type = type(notification)
    handlers = await self._resolve_event_handlers(event_type)
    await self._event_publisher(handlers, notification)

IMediator ¶

Bases: ISender, IPublisher, ABC

Defines a cqrs to encapsulate request/response and publishing interaction patterns.

publish `abstractmethod` `async` ¶

publish(notification: INotification) -> None

Asynchronously send notification to multiple handlers.

Source code in src/waku/cqrs/interfaces.py

@abc.abstractmethod
async def publish(self, notification: INotification, /) -> None:
    """Asynchronously send notification to multiple handlers."""

send `abstractmethod` `async` ¶

send(request: IRequest[None]) -> None

send(request: IRequest[ResponseT]) -> ResponseT

send(request: IRequest[ResponseT]) -> ResponseT

Asynchronously send a request to a single handler.

Source code in src/waku/cqrs/interfaces.py

@abc.abstractmethod
async def send(self, request: IRequest[ResponseT], /) -> ResponseT:
    """Asynchronously send a request to a single handler."""

IPublisher ¶

Bases: ABC

Publish notification through the cqrs to be handled by multiple handlers.

publish `abstractmethod` `async` ¶

publish(notification: INotification) -> None

Asynchronously send notification to multiple handlers.

Source code in src/waku/cqrs/interfaces.py

@abc.abstractmethod
async def publish(self, notification: INotification, /) -> None:
    """Asynchronously send notification to multiple handlers."""

ISender ¶

Bases: ABC

Send a request through the cqrs middleware chain to be handled by a single handler.

send `abstractmethod` `async` ¶

send(request: IRequest[None]) -> None

send(request: IRequest[ResponseT]) -> ResponseT

send(request: IRequest[ResponseT]) -> ResponseT

Asynchronously send a request to a single handler.

Source code in src/waku/cqrs/interfaces.py

@abc.abstractmethod
async def send(self, request: IRequest[ResponseT], /) -> ResponseT:
    """Asynchronously send a request to a single handler."""

MediatorConfig `dataclass` ¶

MediatorConfig(
    *,
    mediator_implementation_type: type[
        IMediator
    ] = Mediator,
    event_publisher: type[
        EventPublisher
    ] = SequentialEventPublisher,
    pipeline_behaviors: Sequence[
        type[IPipelineBehavior[Any, Any]]
    ] = (),
)

Configuration for the Mediator extension.

This class defines the configuration options for setting up the cqrs pattern implementation in the application.

ATTRIBUTE	DESCRIPTION
`mediator_implementation_type`	The concrete implementation class for the cqrs interface (IMediator). Defaults to the standard Mediator class. TYPE: `type[IMediator]`
`event_publisher`	The implementation class for publishing events. Defaults to `SequentialEventPublisher`. TYPE: `type[EventPublisher]`
`pipeline_behaviors`	A sequence of pipeline behavior configurations that will be applied to the cqrs pipeline. Behaviors are executed in the order they are defined. Defaults to an empty sequence. TYPE: `Sequence[type[IPipelineBehavior[Any, Any]]]`

Example

config = MediatorConfig(
    pipeline_behaviors=[
        LoggingBehavior,
        ValidationBehavior,
    ]
)

mediator_implementation_type `class-attribute` `instance-attribute` ¶

mediator_implementation_type: type[IMediator] = Mediator

event_publisher `class-attribute` `instance-attribute` ¶

event_publisher: type[EventPublisher] = (
    SequentialEventPublisher
)

pipeline_behaviors `class-attribute` `instance-attribute` ¶

pipeline_behaviors: Sequence[
    type[IPipelineBehavior[Any, Any]]
] = ()

MediatorExtension ¶

MediatorExtension()

Bases: OnModuleConfigure

Source code in src/waku/cqrs/modules.py

def __init__(self) -> None:
    self._request_map = RequestMap()
    self._event_map = EventMap()
    self._behavior_map = PipelineBehaviorMap()

request_map `property` ¶

request_map: RequestMap

event_map `property` ¶

event_map: EventMap

behavior_map `property` ¶

behavior_map: PipelineBehaviorMap

bind_request ¶

bind_request(
    request_type: type[RequestT],
    handler_type: type[RequestHandler[RequestT, Any]],
    *,
    behaviors: list[type[IPipelineBehavior[RequestT, Any]]]
    | None = None,
) -> Self

Source code in src/waku/cqrs/modules.py

def bind_request(
    self,
    request_type: type[RequestT],
    handler_type: type[RequestHandler[RequestT, Any]],
    *,
    behaviors: list[type[IPipelineBehavior[RequestT, Any]]] | None = None,
) -> Self:
    self._request_map.bind(request_type, handler_type)
    if behaviors:
        self._behavior_map.bind(request_type, behaviors)
    return self

bind_event ¶

bind_event(
    event_type: type[NotificationT],
    handler_types: list[type[EventHandler[NotificationT]]],
) -> Self

Source code in src/waku/cqrs/modules.py

def bind_event(
    self,
    event_type: type[NotificationT],
    handler_types: list[type[EventHandler[NotificationT]]],
) -> Self:
    self._event_map.bind(event_type, handler_types)
    return self

on_module_configure ¶

on_module_configure(metadata: ModuleMetadata) -> None

Source code in src/waku/cqrs/modules.py

def on_module_configure(self, metadata: ModuleMetadata) -> None:
    metadata.providers.extend(
        chain(
            self._create_request_handler_providers(),
            self._create_event_handler_providers(),
            self._create_pipeline_behavior_providers(),
        ),
    )

MediatorModule ¶

register `classmethod` ¶

register(
    config: MediatorConfig | None = None,
) -> DynamicModule

Application-level module for Mediator setup.

PARAMETER	DESCRIPTION
`config`	Configuration for the Mediator extension. TYPE: `MediatorConfig \| None` DEFAULT: `None`

Source code in src/waku/cqrs/modules.py

@classmethod
def register(cls, config: MediatorConfig | None = None, /) -> DynamicModule:
    """Application-level module for Mediator setup.

    Args:
        config: Configuration for the Mediator extension.
    """
    config_ = config or MediatorConfig()
    return DynamicModule(
        parent_module=cls,
        providers=[
            *cls._create_mediator_providers(config_),
            *cls._create_pipeline_behavior_providers(config_),
        ],
        extensions=[RequestMapAggregator(), EventMapAggregator(), PipelineBehaviorMapAggregator()],
        is_global=True,
    )

IRequestHandler ¶

Bases: Protocol[RequestT, ResponseT]

Protocol for request handlers (commands/queries).

MediatR equivalent: IRequestHandler

This protocol allows structural subtyping - any class with a matching handle method signature is compatible.

Example::

class GetUserQueryHandler(IRequestHandler[GetUserQuery, UserDTO]):
    async def handle(self, request: GetUserQuery, /) -> UserDTO:
        return await self._repository.get(request.user_id)

handle `async` ¶

handle(request: RequestT) -> ResponseT

Handle the request and return a response.

Source code in src/waku/cqrs/requests/handler.py

async def handle(self, request: RequestT, /) -> ResponseT:
    """Handle the request and return a response."""
    ...

RequestHandler ¶

Bases: IRequestHandler[RequestT, ResponseT], ABC, Generic[RequestT, ResponseT]

Abstract base class for request handlers.

Use this class when you want explicit ABC inheritance and type checking. For structural subtyping, implement IRequestHandler directly.

Command handler example::

class JoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, request: JoinMeetingCommand, /) -> None:
        await self._meetings_api.join_user(request.user_id, request.meeting_id)

Query handler example::

class ReadMeetingQueryHandler(RequestHandler[ReadMeetingQuery, ReadMeetingQueryResult]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, request: ReadMeetingQuery, /) -> ReadMeetingQueryResult:
        link = await self._meetings_api.get_link(request.meeting_id)
        return ReadMeetingQueryResult(link=link, meeting_id=request.meeting_id)

handle `abstractmethod` `async` ¶

handle(request: RequestT) -> ResponseT

Source code in src/waku/cqrs/requests/handler.py

@abc.abstractmethod
async def handle(self, request: RequestT, /) -> ResponseT:
    raise NotImplementedError

contracts ¶

NotificationT `module-attribute` ¶

NotificationT = TypeVar(
    'NotificationT', bound=INotification, contravariant=True
)

NextHandlerType `module-attribute` ¶

NextHandlerType: TypeAlias = Callable[
    [RequestT], Awaitable[ResponseT]
]

RequestT `module-attribute` ¶

RequestT = TypeVar(
    'RequestT', bound=IRequest[Any], contravariant=True
)

ResponseT `module-attribute` ¶

ResponseT = TypeVar(
    'ResponseT',
    bound='Response | None',
    default=None,
    covariant=True,
)

Event `dataclass` ¶

Event(*, event_id: UUID = uuid4())

Bases: INotification

Convenience base class for events with auto-generated ID.

Use this class when you want automatic event_id generation. For custom identification strategies, implement INotification directly.

Example::

@dataclass(frozen=True, kw_only=True)
class UserCreated(Event):
    user_id: str
    email: str

event_id `class-attribute` `instance-attribute` ¶

event_id: UUID = field(default_factory=uuid4)

INotification ¶

Bases: Protocol

Marker interface for notification-type objects (events).

This is a pure marker protocol with no required attributes or methods. Implement this protocol for domain events that need custom identification strategies or no identification at all.

MediatR equivalent: INotification

Example::

@dataclass(frozen=True)
class OrderPlaced(INotification):
    order_id: str
    customer_id: str


# Or with custom identification:
@dataclass(frozen=True)
class DomainEvent(INotification):
    aggregate_id: str
    occurred_at: datetime


@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: str

IPipelineBehavior ¶

Bases: ABC, Generic[RequestT, ResponseT]

Interface for pipeline behaviors that wrap request handling.

handle `abstractmethod` `async` ¶

handle(
    request: RequestT,
    /,
    next_handler: NextHandlerType[RequestT, ResponseT],
) -> ResponseT

Handle the request and call the next handler in the pipeline.

PARAMETER	DESCRIPTION
`request`	The request to handle TYPE: `RequestT`
`next_handler`	Function to call the next handler in the pipeline TYPE: `NextHandlerType[RequestT, ResponseT]`

RETURNS	DESCRIPTION
`ResponseT`	The response from the pipeline

Source code in src/waku/cqrs/contracts/pipeline.py

@abstractmethod
async def handle(self, request: RequestT, /, next_handler: NextHandlerType[RequestT, ResponseT]) -> ResponseT:
    """Handle the request and call the next handler in the pipeline.

    Args:
        request: The request to handle
        next_handler: Function to call the next handler in the pipeline

    Returns:
        The response from the pipeline
    """
    ...

IRequest ¶

Bases: Protocol[ResponseT]

Marker interface for request-type objects (commands/queries).

This is a pure marker protocol with no required attributes or methods. Implement this protocol for requests that need custom identification strategies or no identification at all.

MediatR equivalent: IRequest

Example::

@dataclass(frozen=True)
class GetUserQuery(IRequest[UserDTO]):
    user_id: str


@dataclass(frozen=True)
class CreateOrderCommand(IRequest[OrderId]):
    customer_id: str
    items: list[OrderItem]

Request `dataclass` ¶

Request(*, request_id: UUID = uuid4())

Bases: IRequest[ResponseT], Generic[ResponseT]

Convenience base class for requests with auto-generated ID.

Use this class when you want automatic request_id generation. For custom identification strategies, implement IRequest directly.

Example::

@dataclass(frozen=True, kw_only=True)
class GetUserQuery(Request[UserDTO]):
    user_id: str

request_id `class-attribute` `instance-attribute` ¶

request_id: UUID = field(default_factory=uuid4)

Response `dataclass` ¶

Response()

Base class for response type objects.

event ¶

NotificationT `module-attribute` ¶

NotificationT = TypeVar(
    'NotificationT', bound=INotification, contravariant=True
)

INotification ¶

Bases: Protocol

Marker interface for notification-type objects (events).

This is a pure marker protocol with no required attributes or methods. Implement this protocol for domain events that need custom identification strategies or no identification at all.

MediatR equivalent: INotification

Example::

@dataclass(frozen=True)
class OrderPlaced(INotification):
    order_id: str
    customer_id: str


# Or with custom identification:
@dataclass(frozen=True)
class DomainEvent(INotification):
    aggregate_id: str
    occurred_at: datetime


@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: str

Event `dataclass` ¶

Event(*, event_id: UUID = uuid4())

Bases: INotification

Convenience base class for events with auto-generated ID.

Use this class when you want automatic event_id generation. For custom identification strategies, implement INotification directly.

Example::

@dataclass(frozen=True, kw_only=True)
class UserCreated(Event):
    user_id: str
    email: str

event_id `class-attribute` `instance-attribute` ¶

event_id: UUID = field(default_factory=uuid4)

notification ¶

NotificationT `module-attribute` ¶

NotificationT = TypeVar(
    'NotificationT', bound=INotification, contravariant=True
)

INotification ¶

Bases: Protocol

Marker interface for notification-type objects (events).

This is a pure marker protocol with no required attributes or methods. Implement this protocol for domain events that need custom identification strategies or no identification at all.

MediatR equivalent: INotification

Example::

@dataclass(frozen=True)
class OrderPlaced(INotification):
    order_id: str
    customer_id: str


# Or with custom identification:
@dataclass(frozen=True)
class DomainEvent(INotification):
    aggregate_id: str
    occurred_at: datetime


@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: str

pipeline ¶

NextHandlerType `module-attribute` ¶

NextHandlerType: TypeAlias = Callable[
    [RequestT], Awaitable[ResponseT]
]

IPipelineBehavior ¶

Bases: ABC, Generic[RequestT, ResponseT]

Interface for pipeline behaviors that wrap request handling.

handle `abstractmethod` `async` ¶

handle(
    request: RequestT,
    /,
    next_handler: NextHandlerType[RequestT, ResponseT],
) -> ResponseT

Handle the request and call the next handler in the pipeline.

PARAMETER	DESCRIPTION
`request`	The request to handle TYPE: `RequestT`
`next_handler`	Function to call the next handler in the pipeline TYPE: `NextHandlerType[RequestT, ResponseT]`

RETURNS	DESCRIPTION
`ResponseT`	The response from the pipeline

Source code in src/waku/cqrs/contracts/pipeline.py

@abstractmethod
async def handle(self, request: RequestT, /, next_handler: NextHandlerType[RequestT, ResponseT]) -> ResponseT:
    """Handle the request and call the next handler in the pipeline.

    Args:
        request: The request to handle
        next_handler: Function to call the next handler in the pipeline

    Returns:
        The response from the pipeline
    """
    ...

request ¶

ResponseT `module-attribute` ¶

ResponseT = TypeVar(
    'ResponseT',
    bound='Response | None',
    default=None,
    covariant=True,
)

RequestT `module-attribute` ¶

RequestT = TypeVar(
    'RequestT', bound=IRequest[Any], contravariant=True
)

IRequest ¶

Bases: Protocol[ResponseT]

Marker interface for request-type objects (commands/queries).

This is a pure marker protocol with no required attributes or methods. Implement this protocol for requests that need custom identification strategies or no identification at all.

MediatR equivalent: IRequest

Example::

@dataclass(frozen=True)
class GetUserQuery(IRequest[UserDTO]):
    user_id: str


@dataclass(frozen=True)
class CreateOrderCommand(IRequest[OrderId]):
    customer_id: str
    items: list[OrderItem]

Request `dataclass` ¶

Request(*, request_id: UUID = uuid4())

Bases: IRequest[ResponseT], Generic[ResponseT]

Convenience base class for requests with auto-generated ID.

Use this class when you want automatic request_id generation. For custom identification strategies, implement IRequest directly.

Example::

@dataclass(frozen=True, kw_only=True)
class GetUserQuery(Request[UserDTO]):
    user_id: str

request_id `class-attribute` `instance-attribute` ¶

request_id: UUID = field(default_factory=uuid4)

Response `dataclass` ¶

Response()

Base class for response type objects.

events ¶

EventHandler ¶

Bases: INotificationHandler[NotificationT], ABC, Generic[NotificationT]

Abstract base class for event handlers.

Use this class when you want explicit ABC inheritance and type checking. For structural subtyping, implement INotificationHandler directly.

Example::

class UserJoinedEventHandler(EventHandler[UserJoinedEvent]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, event: UserJoinedEvent, /) -> None:
        await self._meetings_api.notify_room(event.meeting_id, 'New user joined!')

handle `abstractmethod` `async` ¶

handle(event: NotificationT) -> None

Source code in src/waku/cqrs/events/handler.py

@abc.abstractmethod
async def handle(self, event: NotificationT, /) -> None:
    raise NotImplementedError

EventPublisher ¶

Bases: Protocol

GroupEventPublisher ¶

Bases: EventPublisher

SequentialEventPublisher ¶

Bases: EventPublisher

handler ¶

INotificationHandler ¶

Bases: Protocol[NotificationT]

Protocol for notification/event handlers.

MediatR equivalent: INotificationHandler

This protocol allows structural subtyping - any class with a matching handle method signature is compatible.

Example::

class OrderPlacedHandler(INotificationHandler[OrderPlaced]):
    async def handle(self, event: OrderPlaced, /) -> None:
        await self._send_confirmation_email(event.order_id)

handle `async` ¶

handle(event: NotificationT) -> None

Handle the notification/event.

Source code in src/waku/cqrs/events/handler.py

async def handle(self, event: NotificationT, /) -> None:
    """Handle the notification/event."""
    ...

EventHandler ¶

Bases: INotificationHandler[NotificationT], ABC, Generic[NotificationT]

Abstract base class for event handlers.

Use this class when you want explicit ABC inheritance and type checking. For structural subtyping, implement INotificationHandler directly.

Example::

class UserJoinedEventHandler(EventHandler[UserJoinedEvent]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, event: UserJoinedEvent, /) -> None:
        await self._meetings_api.notify_room(event.meeting_id, 'New user joined!')

handle `abstractmethod` `async` ¶

handle(event: NotificationT) -> None

Source code in src/waku/cqrs/events/handler.py

@abc.abstractmethod
async def handle(self, event: NotificationT, /) -> None:
    raise NotImplementedError

map ¶

EventMapRegistry `module-attribute` ¶

EventMapRegistry: TypeAlias = MutableMapping[
    type[INotification], EventMapEntry[INotification]
]

EventMapEntry `dataclass` ¶

EventMapEntry(
    di_lookup_type: type[EventHandler[_EventT]],
    handler_types: list[
        type[EventHandler[_EventT]]
    ] = list(),
)

Bases: Generic[_EventT]

di_lookup_type `instance-attribute` ¶

di_lookup_type: type[EventHandler[_EventT]]

handler_types `class-attribute` `instance-attribute` ¶

handler_types: list[type[EventHandler[_EventT]]] = field(
    default_factory=list
)

EventMap ¶

EventMap()

Source code in src/waku/cqrs/events/map.py

def __init__(self) -> None:
    self._registry: EventMapRegistry = {}

registry `property` ¶

registry: EventMapRegistry

bind ¶

bind(
    event_type: type[NotificationT],
    handler_types: list[type[EventHandler[NotificationT]]],
) -> Self

Source code in src/waku/cqrs/events/map.py

def bind(self, event_type: type[NotificationT], handler_types: list[type[EventHandler[NotificationT]]]) -> Self:
    if event_type not in self._registry:
        di_lookup_type = EventHandler[event_type]  # type: ignore[valid-type]
        self._registry[event_type] = EventMapEntry(di_lookup_type=di_lookup_type)  # type: ignore[type-abstract]

    entry = self._registry[event_type]
    for handler_type in handler_types:
        if handler_type in entry.handler_types:
            raise EventHandlerAlreadyRegistered(event_type, handler_type)
        entry.handler_types.append(handler_type)  # type: ignore[arg-type]
    return self

merge ¶

merge(other: EventMap) -> Self

Source code in src/waku/cqrs/events/map.py

def merge(self, other: EventMap) -> Self:
    for event_type, entry in other._registry.items():
        self.bind(event_type, entry.handler_types)
    return self

has_handlers ¶

has_handlers(event_type: type[INotification]) -> bool

Source code in src/waku/cqrs/events/map.py

def has_handlers(self, event_type: type[INotification]) -> bool:
    return event_type in self._registry and len(self._registry[event_type].handler_types) > 0

get_handler_type ¶

get_handler_type(
    event_type: type[INotification],
) -> type[EventHandler[INotification]]

Source code in src/waku/cqrs/events/map.py

def get_handler_type(self, event_type: type[INotification]) -> type[EventHandler[INotification]]:
    return self._registry[event_type].di_lookup_type

publish ¶

EventPublisher ¶

Bases: Protocol

SequentialEventPublisher ¶

Bases: EventPublisher

GroupEventPublisher ¶

Bases: EventPublisher

exceptions ¶

MediatorError ¶

Bases: WakuError

Base exception for all cqrs-related errors.

ImproperlyConfiguredError ¶

Bases: MediatorError

Raised when cqrs configuration is invalid.

RequestHandlerAlreadyRegistered ¶

RequestHandlerAlreadyRegistered(
    request_type: type[IRequest[Any]],
    handler_type: type[IRequestHandler[Any, Any]],
)

Bases: MediatorError, KeyError

Raised when a request handler is already registered.

ATTRIBUTE	DESCRIPTION
`request_type`	The type of request that caused the error.
`handler_type`	The type of handler that was already registered.

Source code in src/waku/cqrs/exceptions.py

def __init__(self, request_type: type[IRequest[Any]], handler_type: type[IRequestHandler[Any, Any]]) -> None:
    self.request_type = request_type
    self.handler_type = handler_type

request_type `instance-attribute` ¶

request_type = request_type

handler_type `instance-attribute` ¶

handler_type = handler_type

RequestHandlerNotFound ¶

RequestHandlerNotFound(request_type: type[IRequest[Any]])

Bases: MediatorError, TypeError

Raised when a request handler is not found.

ATTRIBUTE	DESCRIPTION
`request_type`	The type of request that caused the error.

Source code in src/waku/cqrs/exceptions.py

def __init__(self, request_type: type[IRequest[Any]]) -> None:
    self.request_type = request_type

request_type `instance-attribute` ¶

request_type = request_type

EventHandlerAlreadyRegistered ¶

EventHandlerAlreadyRegistered(
    event_type: type[INotification],
    handler_type: type[INotificationHandler[Any]],
)

Bases: MediatorError, KeyError

Raised when an notification handler is already registered.

ATTRIBUTE	DESCRIPTION
`event_type`	The type of notification that caused the error.
`handler_type`	The type of handler that was already registered.

Source code in src/waku/cqrs/exceptions.py

def __init__(self, event_type: type[INotification], handler_type: type[INotificationHandler[Any]]) -> None:
    self.event_type = event_type
    self.handler_type = handler_type

event_type `instance-attribute` ¶

event_type = event_type

handler_type `instance-attribute` ¶

handler_type = handler_type

PipelineBehaviorAlreadyRegistered ¶

PipelineBehaviorAlreadyRegistered(
    request_type: type[IRequest[Any]],
    behavior_type: type[IPipelineBehavior[Any, Any]],
)

Bases: MediatorError, KeyError

Raised when a pipeline behavior is already registered.

ATTRIBUTE	DESCRIPTION
`request_type`	The type of request that caused the error.
`behavior_type`	The type of behavior that was already registered.

Source code in src/waku/cqrs/exceptions.py

def __init__(self, request_type: type[IRequest[Any]], behavior_type: type[IPipelineBehavior[Any, Any]]) -> None:
    self.request_type = request_type
    self.behavior_type = behavior_type

request_type `instance-attribute` ¶

request_type = request_type

behavior_type `instance-attribute` ¶

behavior_type = behavior_type

impl ¶

Mediator ¶

Mediator(
    container: AsyncContainer,
    event_publisher: EventPublisher,
    request_map: RequestMap,
    event_map: EventMap,
    behavior_map: PipelineBehaviorMap,
)

Bases: IMediator

Default CQRS implementation.

Source code in src/waku/cqrs/impl.py

def __init__(
    self,
    container: AsyncContainer,
    event_publisher: EventPublisher,
    request_map: RequestMap,
    event_map: EventMap,
    behavior_map: PipelineBehaviorMap,
) -> None:
    self._container = container
    self._event_publisher = event_publisher
    self._request_map = request_map
    self._event_map = event_map
    self._behavior_map = behavior_map

send `async` ¶

send(request: IRequest[None]) -> None

send(request: IRequest[ResponseT]) -> ResponseT

send(request: IRequest[Any]) -> Any

Source code in src/waku/cqrs/impl.py

@override
async def send(self, request: IRequest[Any], /) -> Any:
    request_type = type(request)
    handler = await self._resolve_request_handler(request_type)  # pyrefly: ignore[bad-argument-type]
    return await self._handle_request(handler, request)

publish `async` ¶

publish(notification: INotification) -> None

Source code in src/waku/cqrs/impl.py

@override
async def publish(self, notification: INotification, /) -> None:
    event_type = type(notification)
    handlers = await self._resolve_event_handlers(event_type)
    await self._event_publisher(handlers, notification)

interfaces ¶

ISender ¶

Bases: ABC

Send a request through the cqrs middleware chain to be handled by a single handler.

send `abstractmethod` `async` ¶

send(request: IRequest[None]) -> None

send(request: IRequest[ResponseT]) -> ResponseT

send(request: IRequest[ResponseT]) -> ResponseT

Asynchronously send a request to a single handler.

Source code in src/waku/cqrs/interfaces.py

@abc.abstractmethod
async def send(self, request: IRequest[ResponseT], /) -> ResponseT:
    """Asynchronously send a request to a single handler."""

IPublisher ¶

Bases: ABC

Publish notification through the cqrs to be handled by multiple handlers.

publish `abstractmethod` `async` ¶

publish(notification: INotification) -> None

Asynchronously send notification to multiple handlers.

Source code in src/waku/cqrs/interfaces.py

@abc.abstractmethod
async def publish(self, notification: INotification, /) -> None:
    """Asynchronously send notification to multiple handlers."""

IMediator ¶

Bases: ISender, IPublisher, ABC

Defines a cqrs to encapsulate request/response and publishing interaction patterns.

publish `abstractmethod` `async` ¶

publish(notification: INotification) -> None

Asynchronously send notification to multiple handlers.

Source code in src/waku/cqrs/interfaces.py

@abc.abstractmethod
async def publish(self, notification: INotification, /) -> None:
    """Asynchronously send notification to multiple handlers."""

send `abstractmethod` `async` ¶

send(request: IRequest[None]) -> None

send(request: IRequest[ResponseT]) -> ResponseT

send(request: IRequest[ResponseT]) -> ResponseT

Asynchronously send a request to a single handler.

Source code in src/waku/cqrs/interfaces.py

@abc.abstractmethod
async def send(self, request: IRequest[ResponseT], /) -> ResponseT:
    """Asynchronously send a request to a single handler."""

modules ¶

MediatorConfig `dataclass` ¶

MediatorConfig(
    *,
    mediator_implementation_type: type[
        IMediator
    ] = Mediator,
    event_publisher: type[
        EventPublisher
    ] = SequentialEventPublisher,
    pipeline_behaviors: Sequence[
        type[IPipelineBehavior[Any, Any]]
    ] = (),
)

Configuration for the Mediator extension.

This class defines the configuration options for setting up the cqrs pattern implementation in the application.

ATTRIBUTE	DESCRIPTION
`mediator_implementation_type`	The concrete implementation class for the cqrs interface (IMediator). Defaults to the standard Mediator class. TYPE: `type[IMediator]`
`event_publisher`	The implementation class for publishing events. Defaults to `SequentialEventPublisher`. TYPE: `type[EventPublisher]`
`pipeline_behaviors`	A sequence of pipeline behavior configurations that will be applied to the cqrs pipeline. Behaviors are executed in the order they are defined. Defaults to an empty sequence. TYPE: `Sequence[type[IPipelineBehavior[Any, Any]]]`

Example

config = MediatorConfig(
    pipeline_behaviors=[
        LoggingBehavior,
        ValidationBehavior,
    ]
)

mediator_implementation_type `class-attribute` `instance-attribute` ¶

mediator_implementation_type: type[IMediator] = Mediator

event_publisher `class-attribute` `instance-attribute` ¶

event_publisher: type[EventPublisher] = (
    SequentialEventPublisher
)

pipeline_behaviors `class-attribute` `instance-attribute` ¶

pipeline_behaviors: Sequence[
    type[IPipelineBehavior[Any, Any]]
] = ()

MediatorModule ¶

register `classmethod` ¶

register(
    config: MediatorConfig | None = None,
) -> DynamicModule

Application-level module for Mediator setup.

PARAMETER	DESCRIPTION
`config`	Configuration for the Mediator extension. TYPE: `MediatorConfig \| None` DEFAULT: `None`

Source code in src/waku/cqrs/modules.py

@classmethod
def register(cls, config: MediatorConfig | None = None, /) -> DynamicModule:
    """Application-level module for Mediator setup.

    Args:
        config: Configuration for the Mediator extension.
    """
    config_ = config or MediatorConfig()
    return DynamicModule(
        parent_module=cls,
        providers=[
            *cls._create_mediator_providers(config_),
            *cls._create_pipeline_behavior_providers(config_),
        ],
        extensions=[RequestMapAggregator(), EventMapAggregator(), PipelineBehaviorMapAggregator()],
        is_global=True,
    )

MediatorExtension ¶

MediatorExtension()

Bases: OnModuleConfigure

Source code in src/waku/cqrs/modules.py

def __init__(self) -> None:
    self._request_map = RequestMap()
    self._event_map = EventMap()
    self._behavior_map = PipelineBehaviorMap()

request_map `property` ¶

request_map: RequestMap

event_map `property` ¶

event_map: EventMap

behavior_map `property` ¶

behavior_map: PipelineBehaviorMap

bind_request ¶

bind_request(
    request_type: type[RequestT],
    handler_type: type[RequestHandler[RequestT, Any]],
    *,
    behaviors: list[type[IPipelineBehavior[RequestT, Any]]]
    | None = None,
) -> Self

Source code in src/waku/cqrs/modules.py

def bind_request(
    self,
    request_type: type[RequestT],
    handler_type: type[RequestHandler[RequestT, Any]],
    *,
    behaviors: list[type[IPipelineBehavior[RequestT, Any]]] | None = None,
) -> Self:
    self._request_map.bind(request_type, handler_type)
    if behaviors:
        self._behavior_map.bind(request_type, behaviors)
    return self

bind_event ¶

bind_event(
    event_type: type[NotificationT],
    handler_types: list[type[EventHandler[NotificationT]]],
) -> Self

Source code in src/waku/cqrs/modules.py

def bind_event(
    self,
    event_type: type[NotificationT],
    handler_types: list[type[EventHandler[NotificationT]]],
) -> Self:
    self._event_map.bind(event_type, handler_types)
    return self

on_module_configure ¶

on_module_configure(metadata: ModuleMetadata) -> None

Source code in src/waku/cqrs/modules.py

def on_module_configure(self, metadata: ModuleMetadata) -> None:
    metadata.providers.extend(
        chain(
            self._create_request_handler_providers(),
            self._create_event_handler_providers(),
            self._create_pipeline_behavior_providers(),
        ),
    )

RequestMapAggregator ¶

Bases: OnModuleRegistration

on_module_registration ¶

on_module_registration(
    registry: ModuleMetadataRegistry,
    owning_module: ModuleType,
    context: Mapping[Any, Any] | None,
) -> None

Source code in src/waku/cqrs/modules.py

@override
def on_module_registration(
    self,
    registry: ModuleMetadataRegistry,
    owning_module: 'ModuleType',
    context: Mapping[Any, Any] | None,
) -> None:
    aggregated = RequestMap()
    for _module_type, ext in registry.find_extensions(MediatorExtension):
        aggregated.merge(ext.request_map)
    registry.add_provider(owning_module, object_(aggregated))

EventMapAggregator ¶

Bases: OnModuleRegistration

on_module_registration ¶

on_module_registration(
    registry: ModuleMetadataRegistry,
    owning_module: ModuleType,
    context: Mapping[Any, Any] | None,
) -> None

Source code in src/waku/cqrs/modules.py

@override
def on_module_registration(
    self,
    registry: ModuleMetadataRegistry,
    owning_module: 'ModuleType',
    context: Mapping[Any, Any] | None,
) -> None:
    aggregated = EventMap()
    for _module_type, ext in registry.find_extensions(MediatorExtension):
        aggregated.merge(ext.event_map)
    registry.add_provider(owning_module, object_(aggregated))

PipelineBehaviorMapAggregator ¶

Bases: OnModuleRegistration

on_module_registration ¶

on_module_registration(
    registry: ModuleMetadataRegistry,
    owning_module: ModuleType,
    context: Mapping[Any, Any] | None,
) -> None

Source code in src/waku/cqrs/modules.py

@override
def on_module_registration(
    self,
    registry: ModuleMetadataRegistry,
    owning_module: 'ModuleType',
    context: Mapping[Any, Any] | None,
) -> None:
    aggregated = PipelineBehaviorMap()
    for _module_type, ext in registry.find_extensions(MediatorExtension):
        aggregated.merge(ext.behavior_map)
    registry.add_provider(owning_module, object_(aggregated))

pipeline ¶

PipelineBehaviorWrapper ¶

PipelineBehaviorWrapper(
    behaviors: Sequence[
        IPipelineBehavior[RequestT, ResponseT]
    ],
)

Bases: Generic[RequestT, ResponseT]

Composes pipeline behaviors into a processing chain.

Source code in src/waku/cqrs/pipeline/chain.py

def __init__(self, behaviors: Sequence[IPipelineBehavior[RequestT, ResponseT]]) -> None:
    self._behaviors = tuple(behaviors)

wrap ¶

wrap(
    handle: NextHandlerType[RequestT, ResponseT],
) -> NextHandlerType[RequestT, ResponseT]

Create a pipeline that wraps the handler function with behaviors.

Pipeline behaviors are executed in the order they are provided.

PARAMETER	DESCRIPTION
`handle`	The handler function to wrap with behaviors TYPE: `NextHandlerType[RequestT, ResponseT]`

RETURNS	DESCRIPTION
`NextHandlerType[RequestT, ResponseT]`	A function that executes the entire pipeline

Source code in src/waku/cqrs/pipeline/chain.py

def wrap(self, handle: NextHandlerType[RequestT, ResponseT]) -> NextHandlerType[RequestT, ResponseT]:
    """Create a pipeline that wraps the handler function with behaviors.

    Pipeline behaviors are executed in the order they are provided.

    Args:
        handle: The handler function to wrap with behaviors

    Returns:
        A function that executes the entire pipeline
    """
    if not self._behaviors:
        return handle

    behaviors = self._behaviors

    async def pipeline(request: RequestT) -> ResponseT:
        async def execute(req: RequestT, idx: int) -> ResponseT:
            if idx >= len(behaviors):
                return await handle(req)
            return await behaviors[idx].handle(req, next_handler=lambda r: execute(r, idx + 1))

        return await execute(request, 0)

    return pipeline

chain ¶

PipelineBehaviorWrapper ¶

PipelineBehaviorWrapper(
    behaviors: Sequence[
        IPipelineBehavior[RequestT, ResponseT]
    ],
)

Bases: Generic[RequestT, ResponseT]

Composes pipeline behaviors into a processing chain.

Source code in src/waku/cqrs/pipeline/chain.py

def __init__(self, behaviors: Sequence[IPipelineBehavior[RequestT, ResponseT]]) -> None:
    self._behaviors = tuple(behaviors)

wrap ¶

wrap(
    handle: NextHandlerType[RequestT, ResponseT],
) -> NextHandlerType[RequestT, ResponseT]

Create a pipeline that wraps the handler function with behaviors.

Pipeline behaviors are executed in the order they are provided.

PARAMETER	DESCRIPTION
`handle`	The handler function to wrap with behaviors TYPE: `NextHandlerType[RequestT, ResponseT]`

RETURNS	DESCRIPTION
`NextHandlerType[RequestT, ResponseT]`	A function that executes the entire pipeline

Source code in src/waku/cqrs/pipeline/chain.py

def wrap(self, handle: NextHandlerType[RequestT, ResponseT]) -> NextHandlerType[RequestT, ResponseT]:
    """Create a pipeline that wraps the handler function with behaviors.

    Pipeline behaviors are executed in the order they are provided.

    Args:
        handle: The handler function to wrap with behaviors

    Returns:
        A function that executes the entire pipeline
    """
    if not self._behaviors:
        return handle

    behaviors = self._behaviors

    async def pipeline(request: RequestT) -> ResponseT:
        async def execute(req: RequestT, idx: int) -> ResponseT:
            if idx >= len(behaviors):
                return await handle(req)
            return await behaviors[idx].handle(req, next_handler=lambda r: execute(r, idx + 1))

        return await execute(request, 0)

    return pipeline

map ¶

PipelineBehaviorMapRegistry `module-attribute` ¶

PipelineBehaviorMapRegistry: TypeAlias = MutableMapping[
    type[RequestT], PipelineBehaviorMapEntry
]

PipelineBehaviorMapEntry `dataclass` ¶

PipelineBehaviorMapEntry(
    di_lookup_type: type[IPipelineBehavior[Any, Any]],
    behavior_types: list[
        type[IPipelineBehavior[Any, Any]]
    ] = list(),
)

di_lookup_type `instance-attribute` ¶

di_lookup_type: type[IPipelineBehavior[Any, Any]]

behavior_types `class-attribute` `instance-attribute` ¶

behavior_types: list[type[IPipelineBehavior[Any, Any]]] = (
    field(default_factory=list)
)

PipelineBehaviorMap ¶

PipelineBehaviorMap()

Source code in src/waku/cqrs/pipeline/map.py

def __init__(self) -> None:
    self._registry: PipelineBehaviorMapRegistry[Any] = {}

registry `property` ¶

registry: PipelineBehaviorMapRegistry[Any]

bind ¶

bind(
    request_type: type[RequestT],
    behavior_types: list[
        type[IPipelineBehavior[RequestT, ResponseT]]
    ],
) -> Self

Source code in src/waku/cqrs/pipeline/map.py

def bind(
    self,
    request_type: type[RequestT],
    behavior_types: list[type[IPipelineBehavior[RequestT, ResponseT]]],
) -> Self:
    if request_type not in self._registry:
        response_type = get_request_response_type(request_type)
        di_lookup_type = IPipelineBehavior[request_type, response_type]  # type: ignore[valid-type]
        self._registry[request_type] = PipelineBehaviorMapEntry(di_lookup_type=di_lookup_type)  # type: ignore[type-abstract]

    entry = self._registry[request_type]
    for behavior_type in behavior_types:
        if behavior_type in entry.behavior_types:
            raise PipelineBehaviorAlreadyRegistered(request_type, behavior_type)
        entry.behavior_types.append(behavior_type)
    return self

merge ¶

merge(other: PipelineBehaviorMap) -> Self

Source code in src/waku/cqrs/pipeline/map.py

def merge(self, other: PipelineBehaviorMap) -> Self:
    for request_type, entry in other._registry.items():
        self.bind(request_type, entry.behavior_types)
    return self

has_behaviors ¶

has_behaviors(request_type: type[RequestT]) -> bool

Source code in src/waku/cqrs/pipeline/map.py

def has_behaviors(self, request_type: type[RequestT]) -> bool:
    return request_type in self._registry and len(self._registry[request_type].behavior_types) > 0

get_lookup_type ¶

get_lookup_type(
    request_type: type[RequestT],
) -> type[IPipelineBehavior[Any, Any]]

Source code in src/waku/cqrs/pipeline/map.py

def get_lookup_type(self, request_type: type[RequestT]) -> type[IPipelineBehavior[Any, Any]]:
    return self._registry[request_type].di_lookup_type

requests ¶

RequestHandler ¶

Bases: IRequestHandler[RequestT, ResponseT], ABC, Generic[RequestT, ResponseT]

Abstract base class for request handlers.

Use this class when you want explicit ABC inheritance and type checking. For structural subtyping, implement IRequestHandler directly.

Command handler example::

class JoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, request: JoinMeetingCommand, /) -> None:
        await self._meetings_api.join_user(request.user_id, request.meeting_id)

Query handler example::

class ReadMeetingQueryHandler(RequestHandler[ReadMeetingQuery, ReadMeetingQueryResult]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, request: ReadMeetingQuery, /) -> ReadMeetingQueryResult:
        link = await self._meetings_api.get_link(request.meeting_id)
        return ReadMeetingQueryResult(link=link, meeting_id=request.meeting_id)

handle `abstractmethod` `async` ¶

handle(request: RequestT) -> ResponseT

Source code in src/waku/cqrs/requests/handler.py

@abc.abstractmethod
async def handle(self, request: RequestT, /) -> ResponseT:
    raise NotImplementedError

handler ¶

IRequestHandler ¶

Bases: Protocol[RequestT, ResponseT]

Protocol for request handlers (commands/queries).

MediatR equivalent: IRequestHandler

This protocol allows structural subtyping - any class with a matching handle method signature is compatible.

Example::

class GetUserQueryHandler(IRequestHandler[GetUserQuery, UserDTO]):
    async def handle(self, request: GetUserQuery, /) -> UserDTO:
        return await self._repository.get(request.user_id)

handle `async` ¶

handle(request: RequestT) -> ResponseT

Handle the request and return a response.

Source code in src/waku/cqrs/requests/handler.py

async def handle(self, request: RequestT, /) -> ResponseT:
    """Handle the request and return a response."""
    ...

RequestHandler ¶

Bases: IRequestHandler[RequestT, ResponseT], ABC, Generic[RequestT, ResponseT]

Abstract base class for request handlers.

Use this class when you want explicit ABC inheritance and type checking. For structural subtyping, implement IRequestHandler directly.

Command handler example::

class JoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, request: JoinMeetingCommand, /) -> None:
        await self._meetings_api.join_user(request.user_id, request.meeting_id)

Query handler example::

class ReadMeetingQueryHandler(RequestHandler[ReadMeetingQuery, ReadMeetingQueryResult]):
    def __init__(self, meetings_api: MeetingAPIProtocol) -> None:
        self._meetings_api = meetings_api

    async def handle(self, request: ReadMeetingQuery, /) -> ReadMeetingQueryResult:
        link = await self._meetings_api.get_link(request.meeting_id)
        return ReadMeetingQueryResult(link=link, meeting_id=request.meeting_id)

handle `abstractmethod` `async` ¶

handle(request: RequestT) -> ResponseT

Source code in src/waku/cqrs/requests/handler.py

@abc.abstractmethod
async def handle(self, request: RequestT, /) -> ResponseT:
    raise NotImplementedError

map ¶

RequestMapRegistry `module-attribute` ¶

RequestMapRegistry: TypeAlias = MutableMapping[
    type[IRequest[Response | None]],
    RequestMapEntry[
        IRequest[Response | None], Response | None
    ],
]

RequestMapEntry `dataclass` ¶

RequestMapEntry(
    handler_type: type[RequestHandler[_MapReqT, _MapResT]],
    di_lookup_type: type[
        RequestHandler[_MapReqT, _MapResT]
    ],
)

Bases: Generic[_MapReqT, _MapResT]

handler_type `instance-attribute` ¶

handler_type: type[RequestHandler[_MapReqT, _MapResT]]

di_lookup_type `instance-attribute` ¶

di_lookup_type: type[RequestHandler[_MapReqT, _MapResT]]

RequestMap ¶

RequestMap()

Source code in src/waku/cqrs/requests/map.py

def __init__(self) -> None:
    self._registry: RequestMapRegistry = {}

registry `property` ¶

registry: RequestMapRegistry

bind ¶

bind(
    request_type: type[RequestT],
    handler_type: type[RequestHandler[RequestT, ResponseT]],
) -> Self

Source code in src/waku/cqrs/requests/map.py

def bind(
    self,
    request_type: type[RequestT],
    handler_type: type[RequestHandler[RequestT, ResponseT]],
) -> Self:
    if request_type in self._registry:
        raise RequestHandlerAlreadyRegistered(request_type, handler_type)
    response_type = get_request_response_type(request_type)
    di_lookup_type = RequestHandler[request_type, response_type]  # type: ignore[valid-type]
    self._registry[request_type] = RequestMapEntry(handler_type, di_lookup_type)  # type: ignore[type-abstract, arg-type]
    return self

merge ¶

merge(other: RequestMap) -> Self

Source code in src/waku/cqrs/requests/map.py

def merge(self, other: RequestMap) -> Self:
    for request_type, entry in other._registry.items():
        if request_type in self._registry:
            raise RequestHandlerAlreadyRegistered(request_type, entry.handler_type)
        self._registry[request_type] = RequestMapEntry(entry.handler_type, entry.di_lookup_type)
    return self

has_handler ¶

has_handler(request_type: type[RequestT]) -> bool

Source code in src/waku/cqrs/requests/map.py

def has_handler(self, request_type: type[RequestT]) -> bool:
    return request_type in self._registry

get_handler_type ¶

get_handler_type(
    request_type: type[RequestT],
) -> type[RequestHandler[RequestT, Response | None]]

Source code in src/waku/cqrs/requests/map.py

def get_handler_type(self, request_type: type[RequestT]) -> type[RequestHandler[RequestT, Response | None]]:
    return self._registry[request_type].di_lookup_type

utils ¶

get_request_response_type `cached` ¶

get_request_response_type(
    request_type: type[IRequest[ResponseT]],
) -> type[ResponseT]

Extract the response type from an IRequest implementation.

Searches through the class hierarchy to find IRequest or its subclasses, supporting direct implementations, Request subclasses, and nested inheritance.

RAISES	DESCRIPTION
`TypeError`	if response type cannot be extracted from the request type.

Source code in src/waku/cqrs/utils.py

@functools.cache
def get_request_response_type(request_type: type[IRequest[ResponseT]]) -> type[ResponseT]:
    """Extract the response type from an IRequest implementation.

    Searches through the class hierarchy to find IRequest or its subclasses,
    supporting direct implementations, Request subclasses, and nested inheritance.

    Raises:
        TypeError: if response type cannot be extracted from the request type.
    """
    for cls in request_type.__mro__:
        if cls is object:
            break
        if response_type := _extract_response_from_bases(cls):
            return response_type  # type: ignore[return-value]

    msg = f'Could not extract response type from {request_type.__name__}'
    raise TypeError(msg)

di ¶

Activator `module-attribute` ¶

Activator: TypeAlias = Callable[[ActivationContext], bool]

ProviderSpec `module-attribute` ¶

ProviderSpec: TypeAlias = Provider | ConditionalProvider

ActivationBuilder ¶

Bases: Protocol

has_active `abstractmethod` ¶

has_active(type_: Any) -> bool

Source code in src/waku/di/_activation.py

@abstractmethod
def has_active(self, type_: Any) -> bool:
    raise NotImplementedError

ActivationContext ¶

Bases: NamedTuple

Context passed to activators for provider activation decisions.

container_context `instance-attribute` ¶

container_context: dict[Any, Any] | None

module_type `instance-attribute` ¶

module_type: ModuleType | DynamicModule

provided_type `instance-attribute` ¶

provided_type: Any

builder `instance-attribute` ¶

builder: ActivationBuilder

ConditionalProvider `dataclass` ¶

ConditionalProvider(
    provider: Provider, when: Activator, provided_type: Any
)

Provider with activation condition.

provider `instance-attribute` ¶

provider: Provider

when `instance-attribute` ¶

when: Activator

provided_type `instance-attribute` ¶

provided_type: Any

Has `dataclass` ¶

Has(type_: Any)

Activator that checks if a provider for a type is registered.

type_ `instance-attribute` ¶

type_: Any

IProviderFilter ¶

Bases: Protocol

Strategy for filtering providers based on activation context.

filter ¶

filter(
    providers: list[ProviderSpec],
    context: dict[Any, Any] | None,
    module_type: ModuleType | DynamicModule,
    builder: ActivationBuilder,
) -> list[Provider]

Source code in src/waku/di/_activation.py

def filter(
    self,
    providers: list[ProviderSpec],
    context: dict[Any, Any] | None,
    module_type: ModuleType | DynamicModule,
    builder: ActivationBuilder,
) -> list[Provider]: ...

ProviderFilter `dataclass` ¶

ProviderFilter(on_skip: OnSkipCallback | None = None)

Default provider filter implementation.

on_skip `class-attribute` `instance-attribute` ¶

on_skip: OnSkipCallback | None = field(default=None)

filter ¶

filter(
    providers: list[ProviderSpec],
    context: dict[Any, Any] | None,
    module_type: ModuleType | DynamicModule,
    builder: ActivationBuilder,
) -> list[Provider]

Source code in src/waku/di/_activation.py

def filter(
    self,
    providers: list[ProviderSpec],
    context: dict[Any, Any] | None,
    module_type: ModuleType | DynamicModule,
    builder: ActivationBuilder,
) -> list[Provider]:
    result: list[Provider] = []

    for spec in providers:
        if isinstance(spec, ConditionalProvider):
            ctx = ActivationContext(
                container_context=context,
                module_type=module_type,
                provided_type=spec.provided_type,
                builder=builder,
            )
            if spec.when(ctx):
                result.append(spec.provider)
            elif self.on_skip:
                self.on_skip(spec, ctx)
        else:
            result.append(spec)

    return result

contextual ¶

contextual(
    provided_type: Any, *, scope: Scope = REQUEST
) -> Provider

contextual(
    provided_type: Any,
    *,
    scope: Scope = REQUEST,
    when: Activator,
) -> ConditionalProvider

contextual(
    provided_type: Any,
    *,
    scope: Scope = REQUEST,
    when: Activator | None = None,
) -> ProviderSpec

Provide a dependency from the current context (e.g., app/request).

PARAMETER	DESCRIPTION
`provided_type`	The type to resolve from context. TYPE: `Any`
`scope`	Scope of the context variable (default: Scope.REQUEST). TYPE: `Scope` DEFAULT: `REQUEST`
`when`	Optional predicate to conditionally activate the provider. TYPE: `Activator \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`ProviderSpec`	Provider or ConditionalProvider if `when` is specified.

Source code in src/waku/di/_providers.py

def contextual(
    provided_type: Any,
    *,
    scope: Scope = Scope.REQUEST,
    when: Activator | None = None,
) -> ProviderSpec:
    """Provide a dependency from the current context (e.g., app/request).

    Args:
        provided_type: The type to resolve from context.
        scope: Scope of the context variable (default: Scope.REQUEST).
        when: Optional predicate to conditionally activate the provider.

    Returns:
        Provider or ConditionalProvider if `when` is specified.
    """
    provider_ = Provider()
    provider_.from_context(provided_type, scope=scope)

    if when is None:
        return provider_
    return ConditionalProvider(provider=provider_, when=when, provided_type=provided_type)

many ¶

many(
    interface: Any,
    *implementations: Any,
    scope: Scope = REQUEST,
    cache: bool = True,
) -> Provider

many(
    interface: Any,
    *implementations: Any,
    scope: Scope = REQUEST,
    cache: bool = True,
    when: Activator,
) -> ConditionalProvider

many(
    interface: Any,
    *implementations: Any,
    scope: Scope = REQUEST,
    cache: bool = True,
    when: Activator | None = None,
) -> ProviderSpec

Register multiple implementations as a collection.

PARAMETER	DESCRIPTION
`interface`	Interface type for the collection. TYPE: `Any`
`*implementations`	Implementation types or factory functions to include in collection. TYPE: `Any` DEFAULT: `()`
`scope`	Scope of the collection (default: Scope.REQUEST). TYPE: `Scope` DEFAULT: `REQUEST`
`cache`	Whether to cache the resolve results within scope. TYPE: `bool` DEFAULT: `True`
`when`	Optional predicate to conditionally activate the provider. TYPE: `Activator \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`ProviderSpec`	Provider or ConditionalProvider if `when` is specified.

RAISES	DESCRIPTION
`ValueError`	If no implementations are provided.
`TypeError`	If a factory function lacks a return type annotation.

Source code in src/waku/di/_providers.py

def many(
    interface: Any,
    *implementations: Any,
    scope: Scope = Scope.REQUEST,
    cache: bool = True,
    when: Activator | None = None,
) -> ProviderSpec:
    """Register multiple implementations as a collection.

    Args:
        interface: Interface type for the collection.
        *implementations: Implementation types or factory functions to include in collection.
        scope: Scope of the collection (default: Scope.REQUEST).
        cache: Whether to cache the resolve results within scope.
        when: Optional predicate to conditionally activate the provider.

    Returns:
        Provider or ConditionalProvider if `when` is specified.

    Raises:
        ValueError: If no implementations are provided.
        TypeError: If a factory function lacks a return type annotation.
    """
    if not implementations:
        msg = 'At least one implementation must be provided'
        raise ValueError(msg)

    provided_types = [_get_provided_type(impl) for impl in implementations]

    provider_ = Provider(scope=scope)
    provider_.provide_all(*implementations, cache=cache)

    provider_.provide(
        lambda: [],  # noqa: PIE807
        provides=list[interface],
        cache=cache,
    )
    provider_.alias(list[interface], provides=Sequence[interface], cache=cache)

    for provided_type in provided_types:

        @provider_.decorate
        def _(many_: list[interface], one: provided_type) -> list[interface]:  # type: ignore[valid-type]
            return [*many_, one]

    if when is None:
        return provider_
    return ConditionalProvider(provider=provider_, when=when, provided_type=Sequence[interface])

object_ ¶

object_(
    obj: Any, *, provided_type: Any | None = None
) -> Provider

object_(
    obj: Any,
    *,
    provided_type: Any | None = None,
    when: Activator,
) -> ConditionalProvider

object_(
    obj: Any,
    *,
    provided_type: Any | None = None,
    when: Activator | None = None,
) -> ProviderSpec

Provide the exact object passed at creation time as a singleton dependency.

The provider always returns the same object instance, without instantiation or copying.

PARAMETER	DESCRIPTION
`obj`	The instance to provide as-is. TYPE: `Any`
`provided_type`	Explicit type to provide (default: inferred). TYPE: `Any \| None` DEFAULT: `None`
`when`	Optional predicate to conditionally activate the provider. TYPE: `Activator \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`ProviderSpec`	Provider or ConditionalProvider if `when` is specified.

Source code in src/waku/di/_providers.py

def object_(
    obj: Any,
    *,
    provided_type: Any | None = None,
    when: Activator | None = None,
) -> ProviderSpec:
    """Provide the exact object passed at creation time as a singleton dependency.

    The provider always returns the same object instance, without instantiation or copying.

    Args:
        obj: The instance to provide as-is.
        provided_type: Explicit type to provide (default: inferred).
        when: Optional predicate to conditionally activate the provider.

    Returns:
        Provider or ConditionalProvider if `when` is specified.
    """
    actual_type = provided_type if provided_type is not None else type(obj)
    base = provider(lambda: obj, scope=Scope.APP, provided_type=actual_type, cache=True)

    if when is None:
        return base
    return ConditionalProvider(provider=base, when=when, provided_type=actual_type)

provider ¶

provider(
    source: Callable[..., Any] | type[Any],
    *,
    scope: Scope = REQUEST,
    provided_type: Any | None = None,
    cache: bool = True,
) -> Provider

Create a Dishka provider for a callable or type.

PARAMETER	DESCRIPTION
`source`	Callable or type to provide as a dependency. TYPE: `Callable[..., Any] \| type[Any]`
`scope`	Scope of the dependency (default: Scope.REQUEST). TYPE: `Scope` DEFAULT: `REQUEST`
`provided_type`	Explicit type to provide (default: inferred). TYPE: `Any \| None` DEFAULT: `None`
`cache`	Whether to cache the instance in the scope. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`Provider`	Configured provider instance. TYPE: `Provider`

Source code in src/waku/di/_providers.py

def provider(
    source: Callable[..., Any] | type[Any],
    *,
    scope: Scope = Scope.REQUEST,
    provided_type: Any | None = None,
    cache: bool = True,
) -> Provider:
    """Create a Dishka provider for a callable or type.

    Args:
        source: Callable or type to provide as a dependency.
        scope: Scope of the dependency (default: Scope.REQUEST).
        provided_type: Explicit type to provide (default: inferred).
        cache: Whether to cache the instance in the scope.

    Returns:
        Provider: Configured provider instance.
    """
    provider_ = Provider(scope=scope)
    provider_.provide(source, provides=provided_type, cache=cache)
    return provider_

scoped ¶

scoped(source: type[_T] | Callable[..., _T]) -> Provider

scoped(
    source: type[_T] | Callable[..., _T],
    /,
    *,
    when: Activator,
) -> ConditionalProvider

scoped(
    interface: Any,
    implementation: type[_T] | Callable[..., _T],
) -> Provider

scoped(
    interface: Any,
    implementation: type[_T] | Callable[..., _T],
    /,
    *,
    when: Activator,
) -> ConditionalProvider

scoped(
    interface_or_source: type[Any] | Callable[..., Any],
    implementation: type[Any]
    | Callable[..., Any]
    | None = None,
    /,
    *,
    when: Activator | None = None,
) -> ProviderSpec

Create a scoped provider (lifetime: request).

PARAMETER	DESCRIPTION
`interface_or_source`	Interface type or source if no separate implementation. TYPE: `type[Any] \| Callable[..., Any]`
`implementation`	Implementation type if interface is provided. TYPE: `type[Any] \| Callable[..., Any] \| None` DEFAULT: `None`
`when`	Optional predicate to conditionally activate the provider. TYPE: `Activator \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`ProviderSpec`	Provider or ConditionalProvider if `when` is specified.

Source code in src/waku/di/_providers.py

def scoped(
    interface_or_source: type[Any] | Callable[..., Any],
    implementation: type[Any] | Callable[..., Any] | None = None,
    /,
    *,
    when: Activator | None = None,
) -> ProviderSpec:
    """Create a scoped provider (lifetime: request).

    Args:
        interface_or_source: Interface type or source if no separate implementation.
        implementation: Implementation type if interface is provided.
        when: Optional predicate to conditionally activate the provider.

    Returns:
        Provider or ConditionalProvider if `when` is specified.
    """
    if implementation is not None:
        provided_type = interface_or_source
        base = provider(implementation, scope=Scope.REQUEST, provided_type=provided_type)
    else:
        provided_type = _get_provided_type(interface_or_source)
        base = provider(interface_or_source, scope=Scope.REQUEST)

    if when is None:
        return base
    return ConditionalProvider(provider=base, when=when, provided_type=provided_type)

singleton ¶

singleton(source: type[_T] | Callable[..., _T]) -> Provider

singleton(
    source: type[_T] | Callable[..., _T],
    /,
    *,
    when: Activator,
) -> ConditionalProvider

singleton(
    interface: Any,
    implementation: type[_T] | Callable[..., _T],
) -> Provider

singleton(
    interface: Any,
    implementation: type[_T] | Callable[..., _T],
    /,
    *,
    when: Activator,
) -> ConditionalProvider

singleton(
    interface_or_source: type[Any] | Callable[..., Any],
    implementation: type[Any]
    | Callable[..., Any]
    | None = None,
    /,
    *,
    when: Activator | None = None,
) -> ProviderSpec

Create a singleton provider (lifetime: app).

PARAMETER	DESCRIPTION
`interface_or_source`	Interface type or source if no separate implementation. TYPE: `type[Any] \| Callable[..., Any]`
`implementation`	Implementation type if interface is provided. TYPE: `type[Any] \| Callable[..., Any] \| None` DEFAULT: `None`
`when`	Optional predicate to conditionally activate the provider. TYPE: `Activator \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`ProviderSpec`	Provider or ConditionalProvider if `when` is specified.

Source code in src/waku/di/_providers.py

def singleton(
    interface_or_source: type[Any] | Callable[..., Any],
    implementation: type[Any] | Callable[..., Any] | None = None,
    /,
    *,
    when: Activator | None = None,
) -> ProviderSpec:
    """Create a singleton provider (lifetime: app).

    Args:
        interface_or_source: Interface type or source if no separate implementation.
        implementation: Implementation type if interface is provided.
        when: Optional predicate to conditionally activate the provider.

    Returns:
        Provider or ConditionalProvider if `when` is specified.
    """
    if implementation is not None:
        provided_type = interface_or_source
        base = provider(implementation, scope=Scope.APP, provided_type=provided_type)
    else:
        provided_type = _get_provided_type(interface_or_source)
        base = provider(interface_or_source, scope=Scope.APP)

    if when is None:
        return base
    return ConditionalProvider(provider=base, when=when, provided_type=provided_type)

transient ¶

transient(source: type[_T] | Callable[..., _T]) -> Provider

transient(
    source: type[_T] | Callable[..., _T],
    /,
    *,
    when: Activator,
) -> ConditionalProvider

transient(
    interface: Any,
    implementation: type[_T" backlink-type="used-by" backlink-anchor="waku.di.transient" optional hover>_T] | Callable[..., _T],
) -> Provider

transient(
    interface: Any,
    implementation: type[_T] | Callable[..., _T],
    /,
    *,
    when: Activator,
) -> ConditionalProvider

transient(
    interface_or_source: type[Any] | Callable[..., Any],
    implementation: type[Any]
    | Callable[..., Any]
    | None = None,
    /,
    *,
    when: Activator | None = None,
) -> ProviderSpec

Create a transient provider (new instance per injection).

PARAMETER	DESCRIPTION
`interface_or_source`	Interface type or source if no separate implementation. TYPE: `type[Any] \| Callable[..., Any]`
`implementation`	Implementation type if interface is provided. TYPE: `type[Any] \| Callable[..., Any] \| None` DEFAULT: `None`
`when`	Optional predicate to conditionally activate the provider. TYPE: `Activator \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`ProviderSpec`	Provider or ConditionalProvider if `when` is specified.

Source code in src/waku/di/_providers.py

def transient(
    interface_or_source: type[Any] | Callable[..., Any],
    implementation: type[Any] | Callable[..., Any] | None = None,
    /,
    *,
    when: Activator | None = None,
) -> ProviderSpec:
    """Create a transient provider (new instance per injection).

    Args:
        interface_or_source: Interface type or source if no separate implementation.
        implementation: Implementation type if interface is provided.
        when: Optional predicate to conditionally activate the provider.

    Returns:
        Provider or ConditionalProvider if `when` is specified.
    """
    if implementation is not None:
        provided_type = interface_or_source
        base = provider(implementation, scope=Scope.REQUEST, provided_type=provided_type, cache=False)
    else:
        provided_type = _get_provided_type(interface_or_source)
        base = provider(interface_or_source, scope=Scope.REQUEST, cache=False)

    if when is None:
        return base
    return ConditionalProvider(provider=base, when=when, provided_type=provided_type)

exceptions ¶

WakuError ¶

Bases: Exception

extensions ¶

ApplicationExtension `module-attribute` ¶

ApplicationExtension: TypeAlias = (
    OnApplicationInit
    | AfterApplicationInit
    | OnApplicationShutdown
    | OnModuleRegistration
)

ModuleExtension `module-attribute` ¶

ModuleExtension: TypeAlias = (
    OnModuleConfigure
    | OnModuleInit
    | OnModuleDestroy
    | OnModuleRegistration
)

DEFAULT_EXTENSIONS `module-attribute` ¶

DEFAULT_EXTENSIONS: Sequence[ApplicationExtension] = (
    ValidationExtension(
        [DependenciesAccessibleRule()], strict=True
    ),
)

AfterApplicationInit ¶

Bases: Protocol

Extension for application post-initialization actions.

after_app_init `async` ¶

after_app_init(app: WakuApplication) -> None

Source code in src/waku/extensions/protocols.py

async def after_app_init(self, app: WakuApplication) -> None: ...

OnApplicationInit ¶

Bases: Protocol

Extension for application pre-initialization actions.

on_app_init `async` ¶

on_app_init(app: WakuApplication) -> None

Source code in src/waku/extensions/protocols.py

async def on_app_init(self, app: WakuApplication) -> None: ...

OnApplicationShutdown ¶

Bases: Protocol

Extension for application shutdown actions.

on_app_shutdown `async` ¶

on_app_shutdown(app: WakuApplication) -> None

Source code in src/waku/extensions/protocols.py

async def on_app_shutdown(self, app: WakuApplication) -> None: ...

OnModuleConfigure ¶

Bases: Protocol

Extension for module configuration.

on_module_configure ¶

on_module_configure(metadata: ModuleMetadata) -> None

Perform actions before module metadata transformed to module.

Source code in src/waku/extensions/protocols.py

def on_module_configure(self, metadata: ModuleMetadata) -> None:
    """Perform actions before module metadata transformed to module."""
    ...

OnModuleDestroy ¶

Bases: Protocol

Extension for module destroying.

on_module_destroy `async` ¶

on_module_destroy(module: Module) -> None

Source code in src/waku/extensions/protocols.py

async def on_module_destroy(self, module: Module) -> None: ...

OnModuleInit ¶

Bases: Protocol

Extension for module initialization.

on_module_init `async` ¶

on_module_init(module: Module) -> None

Source code in src/waku/extensions/protocols.py

async def on_module_init(self, module: Module) -> None: ...

OnModuleRegistration ¶

Bases: Protocol

Extension for contributing providers to module metadata during registration.

This hook runs after all module metadata is collected but before Module objects are created. Use this for cross-module aggregation that produces providers which should belong to the owning module.

Can be declared at both application level (passed to WakuFactory) and module level (in module's extensions list).

Execution order

Application-level extensions (assigned to root module)
Module-level extensions (in topological order)

Key differences from OnModuleConfigure

Runs after ALL modules are collected (cross-module visibility)
Receives registry with access to all modules' metadata
Can add providers to owning module

on_module_registration ¶

on_module_registration(
    registry: ModuleMetadataRegistry,
    owning_module: ModuleType,
    context: Mapping[Any, Any] | None,
) -> None

Contribute providers to module metadata before Module objects are created.

PARAMETER	DESCRIPTION
`registry`	Registry of all collected module metadata. Use find_extensions() to discover extensions across modules, add_provider() to contribute. TYPE: `ModuleMetadataRegistry`
`owning_module`	The module type that owns this extension. Providers added via registry.add_provider() should target this module. TYPE: `ModuleType`
`context`	Application context passed to WakuFactory (read-only). TYPE: `Mapping[Any, Any] \| None`

Source code in src/waku/extensions/protocols.py

def on_module_registration(
    self,
    registry: ModuleMetadataRegistry,
    owning_module: ModuleType,
    context: Mapping[Any, Any] | None,
) -> None:
    """Contribute providers to module metadata before Module objects are created.

    Args:
        registry: Registry of all collected module metadata. Use find_extensions()
                  to discover extensions across modules, add_provider() to contribute.
        owning_module: The module type that owns this extension. Providers
                      added via registry.add_provider() should target this module.
        context: Application context passed to WakuFactory (read-only).
    """
    ...

ExtensionRegistry ¶

ExtensionRegistry()

Registry for extensions.

This registry maintains references to all extensions in the application, allowing for centralized management and discovery.

Source code in src/waku/extensions/registry.py

def __init__(self) -> None:
    self._app_extensions: dict[type[ApplicationExtension], list[ApplicationExtension]] = defaultdict(list)
    self._module_extensions: dict[ModuleType, list[ModuleExtension]] = defaultdict(list)

register_application_extension ¶

register_application_extension(
    extension: ApplicationExtension,
) -> Self

Register an application extension with optional priority and tags.

Source code in src/waku/extensions/registry.py

def register_application_extension(self, extension: ApplicationExtension) -> Self:
    """Register an application extension with optional priority and tags."""
    ext_type = type(extension)
    extension_bases = [
        base
        for base in inspect.getmro(ext_type)
        if (isinstance(base, ApplicationExtension) and base != ext_type)  # type: ignore[unreachable]
    ]
    for base in extension_bases:
        self._app_extensions[cast('type[ApplicationExtension]', base)].append(extension)
    return self

register_module_extension ¶

register_module_extension(
    module_type: ModuleType, extension: ModuleExtension
) -> Self

Source code in src/waku/extensions/registry.py

def register_module_extension(self, module_type: ModuleType, extension: ModuleExtension) -> Self:
    self._module_extensions[module_type].append(extension)
    return self

get_application_extensions ¶

get_application_extensions(
    extension_type: type[_AppExtT],
) -> list[_AppExtT]

Source code in src/waku/extensions/registry.py

def get_application_extensions(self, extension_type: type[_AppExtT]) -> list[_AppExtT]:
    return cast('list[_AppExtT]', self._app_extensions.get(cast('type[ApplicationExtension]', extension_type), []))

get_module_extensions ¶

get_module_extensions(
    module_type: ModuleType, extension_type: type[_ModExtT]
) -> list[_ModExtT]

Source code in src/waku/extensions/registry.py

def get_module_extensions(self, module_type: ModuleType, extension_type: type[_ModExtT]) -> list[_ModExtT]:
    extensions = cast('list[_ModExtT]', self._module_extensions.get(module_type, []))
    return [ext for ext in extensions if isinstance(ext, extension_type)]

protocols ¶

Extension protocols for application and module lifecycle hooks.

ApplicationExtension `module-attribute` ¶

ApplicationExtension: TypeAlias = (
    OnApplicationInit
    | AfterApplicationInit
    | OnApplicationShutdown
    | OnModuleRegistration
)

ModuleExtension `module-attribute` ¶

ModuleExtension: TypeAlias = (
    OnModuleConfigure
    | OnModuleInit
    | OnModuleDestroy
    | OnModuleRegistration
)

OnApplicationInit ¶

Bases: Protocol

Extension for application pre-initialization actions.

on_app_init `async` ¶

on_app_init(app: WakuApplication) -> None

Source code in src/waku/extensions/protocols.py

async def on_app_init(self, app: WakuApplication) -> None: ...

AfterApplicationInit ¶

Bases: Protocol

Extension for application post-initialization actions.

after_app_init `async` ¶

after_app_init(app: WakuApplication) -> None

Source code in src/waku/extensions/protocols.py

async def after_app_init(self, app: WakuApplication) -> None: ...

OnApplicationShutdown ¶

Bases: Protocol

Extension for application shutdown actions.

on_app_shutdown `async` ¶

on_app_shutdown(app: WakuApplication) -> None

Source code in src/waku/extensions/protocols.py

async def on_app_shutdown(self, app: WakuApplication) -> None: ...

OnModuleRegistration ¶

Bases: Protocol

Extension for contributing providers to module metadata during registration.

This hook runs after all module metadata is collected but before Module objects are created. Use this for cross-module aggregation that produces providers which should belong to the owning module.

Can be declared at both application level (passed to WakuFactory) and module level (in module's extensions list).

Execution order

Application-level extensions (assigned to root module)
Module-level extensions (in topological order)

Key differences from OnModuleConfigure

Runs after ALL modules are collected (cross-module visibility)
Receives registry with access to all modules' metadata
Can add providers to owning module

on_module_registration ¶

on_module_registration(
    registry: ModuleMetadataRegistry,
    owning_module: ModuleType,
    context: Mapping[Any, Any] | None,
) -> None

Contribute providers to module metadata before Module objects are created.

PARAMETER	DESCRIPTION
`registry`	Registry of all collected module metadata. Use find_extensions() to discover extensions across modules, add_provider() to contribute. TYPE: `ModuleMetadataRegistry`
`owning_module`	The module type that owns this extension. Providers added via registry.add_provider() should target this module. TYPE: `ModuleType`
`context`	Application context passed to WakuFactory (read-only). TYPE: `Mapping[Any, Any] \| None`

Source code in src/waku/extensions/protocols.py

def on_module_registration(
    self,
    registry: ModuleMetadataRegistry,
    owning_module: ModuleType,
    context: Mapping[Any, Any] | None,
) -> None:
    """Contribute providers to module metadata before Module objects are created.

    Args:
        registry: Registry of all collected module metadata. Use find_extensions()
                  to discover extensions across modules, add_provider() to contribute.
        owning_module: The module type that owns this extension. Providers
                      added via registry.add_provider() should target this module.
        context: Application context passed to WakuFactory (read-only).
    """
    ...

OnModuleConfigure ¶

Bases: Protocol

Extension for module configuration.

on_module_configure ¶

on_module_configure(metadata: ModuleMetadata) -> None

Perform actions before module metadata transformed to module.

Source code in src/waku/extensions/protocols.py

def on_module_configure(self, metadata: ModuleMetadata) -> None:
    """Perform actions before module metadata transformed to module."""
    ...

OnModuleInit ¶

Bases: Protocol

Extension for module initialization.

on_module_init `async` ¶

on_module_init(module: Module) -> None

Source code in src/waku/extensions/protocols.py

async def on_module_init(self, module: Module) -> None: ...

OnModuleDestroy ¶

Bases: Protocol

Extension for module destroying.

on_module_destroy `async` ¶

on_module_destroy(module: Module) -> None

Source code in src/waku/extensions/protocols.py

async def on_module_destroy(self, module: Module) -> None: ...

registry ¶

Extension registry for centralized management of extensions.

ExtensionRegistry ¶

ExtensionRegistry()

Registry for extensions.

This registry maintains references to all extensions in the application, allowing for centralized management and discovery.

Source code in src/waku/extensions/registry.py

def __init__(self) -> None:
    self._app_extensions: dict[type[ApplicationExtension], list[ApplicationExtension]] = defaultdict(list)
    self._module_extensions: dict[ModuleType, list[ModuleExtension]] = defaultdict(list)

register_application_extension ¶

register_application_extension(
    extension: ApplicationExtension,
) -> Self

Register an application extension with optional priority and tags.

Source code in src/waku/extensions/registry.py

def register_application_extension(self, extension: ApplicationExtension) -> Self:
    """Register an application extension with optional priority and tags."""
    ext_type = type(extension)
    extension_bases = [
        base
        for base in inspect.getmro(ext_type)
        if (isinstance(base, ApplicationExtension) and base != ext_type)  # type: ignore[unreachable]
    ]
    for base in extension_bases:
        self._app_extensions[cast('type[ApplicationExtension]', base)].append(extension)
    return self

register_module_extension ¶

register_module_extension(
    module_type: ModuleType, extension: ModuleExtension
) -> Self

Source code in src/waku/extensions/registry.py

def register_module_extension(self, module_type: ModuleType, extension: ModuleExtension) -> Self:
    self._module_extensions[module_type].append(extension)
    return self

get_application_extensions ¶

get_application_extensions(
    extension_type: type[_AppExtT],
) -> list[_AppExtT]

Source code in src/waku/extensions/registry.py

def get_application_extensions(self, extension_type: type[_AppExtT]) -> list[_AppExtT]:
    return cast('list[_AppExtT]', self._app_extensions.get(cast('type[ApplicationExtension]', extension_type), []))

get_module_extensions ¶

get_module_extensions(
    module_type: ModuleType, extension_type: type[_ModExtT]
) -> list[_ModExtT]

Source code in src/waku/extensions/registry.py

def get_module_extensions(self, module_type: ModuleType, extension_type: type[_ModExtT]) -> list[_ModExtT]:
    extensions = cast('list[_ModExtT]', self._module_extensions.get(module_type, []))
    return [ext for ext in extensions if isinstance(ext, extension_type)]

factory ¶

ContainerConfig `dataclass` ¶

ContainerConfig(
    *,
    lock_factory: _LockFactory = Lock,
    start_scope: Scope | None = None,
    skip_validation: bool = False,
)

lock_factory `class-attribute` `instance-attribute` ¶

lock_factory: _LockFactory = Lock

start_scope `class-attribute` `instance-attribute` ¶

start_scope: Scope | None = None

skip_validation `class-attribute` `instance-attribute` ¶

skip_validation: bool = False

WakuFactory ¶

WakuFactory(
    root_module_type: ModuleType,
    /,
    context: dict[Any, Any] | None = None,
    lifespan: Sequence[LifespanFunc] = (),
    extensions: Sequence[
        ApplicationExtension
    ] = DEFAULT_EXTENSIONS,
    container_config: ContainerConfig | None = None,
    provider_filter: IProviderFilter | None = None,
)

Source code in src/waku/factory.py

def __init__(
    self,
    root_module_type: ModuleType,
    /,
    context: dict[Any, Any] | None = None,
    lifespan: Sequence[LifespanFunc] = (),
    extensions: Sequence[ApplicationExtension] = DEFAULT_EXTENSIONS,
    container_config: ContainerConfig | None = None,
    provider_filter: IProviderFilter | None = None,
) -> None:
    self._root_module_type = root_module_type

    self._context = context
    self._lifespan = lifespan
    self._extensions = extensions
    self._container_config = container_config or ContainerConfig()
    self._provider_filter = provider_filter

create ¶

create() -> WakuApplication

Source code in src/waku/factory.py

def create(self) -> WakuApplication:
    registry = ModuleRegistryBuilder(
        self._root_module_type,
        context=self._context,
        provider_filter=self._provider_filter,
        app_extensions=self._extensions,
    ).build()

    container = self._build_container(registry.providers)
    return WakuApplication(
        container=container,
        registry=registry,
        lifespan=self._lifespan,
        extension_registry=self._build_extension_registry(registry.modules),
    )

lifespan ¶

LifespanFunc `module-attribute` ¶

LifespanFunc: TypeAlias = (
    Callable[
        ['WakuApplication'],
        AbstractAsyncContextManager[None],
    ]
    | AbstractAsyncContextManager[None]
)

LifespanWrapper ¶

LifespanWrapper(lifespan_func: LifespanFunc)

Source code in src/waku/lifespan.py

def __init__(self, lifespan_func: LifespanFunc) -> None:
    self._lifespan_func = lifespan_func

lifespan `async` ¶

lifespan(app: WakuApplication) -> AsyncIterator[None]

Source code in src/waku/lifespan.py

@asynccontextmanager
async def lifespan(self, app: WakuApplication) -> AsyncIterator[None]:
    ctx_manager = (
        self._lifespan_func
        if isinstance(self._lifespan_func, AbstractAsyncContextManager)
        else self._lifespan_func(app)
    )
    async with ctx_manager:
        yield

modules ¶

ModuleType `module-attribute` ¶

ModuleType: TypeAlias = type[object | HasModuleMetadata]

DynamicModule `dataclass` ¶

DynamicModule(
    *,
    providers: list[ProviderSpec] = list(),
    imports: list[ModuleType | DynamicModule] = list(),
    exports: list[
        type[object] | ModuleType | DynamicModule
    ] = list(),
    extensions: list[ModuleExtension] = list(),
    is_global: bool = False,
    id: UUID = uuid4(),
    parent_module: ModuleType,
)

Bases: ModuleMetadata

providers `class-attribute` `instance-attribute` ¶

providers: list[ProviderSpec] = field(default_factory=list)

List of providers for dependency injection.

imports `class-attribute` `instance-attribute` ¶

imports: list[ModuleType | DynamicModule] = field(
    default_factory=list
)

List of modules imported by this module.

exports `class-attribute` `instance-attribute` ¶

exports: list[type[object] | ModuleType | DynamicModule] = (
    field(default_factory=list)
)

List of types or modules exported by this module.

extensions `class-attribute` `instance-attribute` ¶

extensions: list[ModuleExtension] = field(
    default_factory=list
)

List of module extensions for lifecycle hooks.

is_global `class-attribute` `instance-attribute` ¶

is_global: bool = False

Whether this module is global or not.

id `class-attribute` `instance-attribute` ¶

id: UUID = field(default_factory=uuid4)

parent_module `instance-attribute` ¶

parent_module: ModuleType

HasModuleMetadata ¶

Bases: Protocol

ModuleCompiler ¶

extract_metadata ¶

extract_metadata(
    module_type: ModuleType | DynamicModule,
) -> tuple[ModuleType, ModuleMetadata]

Source code in src/waku/modules/_metadata.py

def extract_metadata(self, module_type: ModuleType | DynamicModule) -> tuple[ModuleType, ModuleMetadata]:
    try:
        return self._extract_metadata(cast('Hashable', module_type))
    except AttributeError:
        msg = f'{type(module_type).__name__} is not module'
        raise ValueError(msg) from None

ModuleMetadata `dataclass` ¶

ModuleMetadata(
    *,
    providers: list[ProviderSpec] = list(),
    imports: list[ModuleType | DynamicModule] = list(),
    exports: list[
        type[object] | ModuleType | DynamicModule
    ] = list(),
    extensions: list[ModuleExtension] = list(),
    is_global: bool = False,
    id: UUID = uuid4(),
)

providers `class-attribute` `instance-attribute` ¶

providers: list[ProviderSpec] = field(default_factory=list)

List of providers for dependency injection.

imports `class-attribute` `instance-attribute` ¶

imports: list[ModuleType | DynamicModule] = field(
    default_factory=list
)

List of modules imported by this module.

exports `class-attribute` `instance-attribute` ¶

exports: list[type[object] | ModuleType | DynamicModule] = (
    field(default_factory=list)
)

List of types or modules exported by this module.

extensions `class-attribute` `instance-attribute` ¶

extensions: list[ModuleExtension] = field(
    default_factory=list
)

List of module extensions for lifecycle hooks.

is_global `class-attribute` `instance-attribute` ¶

is_global: bool = False

Whether this module is global or not.

id `class-attribute` `instance-attribute` ¶

id: UUID = field(default_factory=uuid4)

ModuleMetadataRegistry ¶

ModuleMetadataRegistry(
    metadata_by_type: dict[ModuleType, ModuleMetadata],
    topological_order: tuple[ModuleType, ...],
)

Registry providing access to collected module metadata.

Provides read access to all modules' metadata for aggregation purposes, with controlled write access through explicit methods.

This class is used during the module registration phase to enable cross-module aggregation of providers.

Source code in src/waku/modules/_metadata_registry.py

def __init__(
    self,
    metadata_by_type: dict[ModuleType, ModuleMetadata],
    topological_order: tuple[ModuleType, ...],
) -> None:
    self._metadata_by_type = metadata_by_type
    self._topological_order = topological_order

modules `property` ¶

modules: tuple[ModuleType, ...]

All module types in topological order (dependencies first).

get_metadata ¶

get_metadata(module_type: ModuleType) -> ModuleMetadata

Get metadata for a specific module.

Source code in src/waku/modules/_metadata_registry.py

def get_metadata(self, module_type: ModuleType) -> ModuleMetadata:
    """Get metadata for a specific module."""
    return self._metadata_by_type[module_type]

find_extensions ¶

find_extensions(
    protocol: type[_ExtT],
) -> Iterator[tuple[ModuleType, _ExtT]]

Find all extensions of a given type across all modules.

Yields (module_type, extension) pairs in topological order. This is useful for aggregating data from extensions across modules.

PARAMETER	DESCRIPTION
`protocol`	The extension protocol/type to search for. TYPE: `type[_ExtT]`

YIELDS	DESCRIPTION
`tuple[ModuleType, _ExtT]`	Tuples of (module_type, extension) for each matching extension.

Source code in src/waku/modules/_metadata_registry.py

def find_extensions(self, protocol: type[_ExtT]) -> Iterator[tuple[ModuleType, _ExtT]]:
    """Find all extensions of a given type across all modules.

    Yields (module_type, extension) pairs in topological order.
    This is useful for aggregating data from extensions across modules.

    Args:
        protocol: The extension protocol/type to search for.

    Yields:
        Tuples of (module_type, extension) for each matching extension.
    """
    for module_type in self._topological_order:
        metadata = self._metadata_by_type[module_type]
        for ext in metadata.extensions:
            if isinstance(ext, protocol):
                yield module_type, ext

add_provider ¶

add_provider(
    module_type: ModuleType, provider: ProviderSpec
) -> None

Add a provider to a module's metadata.

This is the preferred way to add providers during registration hooks. The provider will become part of the owning module.

PARAMETER	DESCRIPTION
`module_type`	The module to add the provider to. TYPE: `ModuleType`
`provider`	The provider specification to add. TYPE: `ProviderSpec`

RAISES	DESCRIPTION
`KeyError`	If module_type is not in the registry.

Source code in src/waku/modules/_metadata_registry.py

def add_provider(self, module_type: ModuleType, provider: ProviderSpec) -> None:
    """Add a provider to a module's metadata.

    This is the preferred way to add providers during registration hooks.
    The provider will become part of the owning module.

    Args:
        module_type: The module to add the provider to.
        provider: The provider specification to add.

    Raises:
        KeyError: If module_type is not in the registry.
    """
    self._metadata_by_type[module_type].providers.append(provider)

Module ¶

Module(module_type: ModuleType, metadata: ModuleMetadata)

Source code in src/waku/modules/_module.py

def __init__(self, module_type: ModuleType, metadata: ModuleMetadata) -> None:
    self.id: Final[UUID] = metadata.id
    self.target: Final[ModuleType] = module_type

    self.providers: Final[Sequence[ProviderSpec]] = metadata.providers
    self.imports: Final[Sequence[ModuleType | DynamicModule]] = metadata.imports
    self.exports: Final[Sequence[type[object] | ModuleType | DynamicModule]] = metadata.exports
    self.extensions: Final[Sequence[ModuleExtension]] = metadata.extensions
    self.is_global: Final[bool] = metadata.is_global

    self._provider: BaseProvider | None = None

id `instance-attribute` ¶

id: Final[UUID] = id

target `instance-attribute` ¶

target: Final[ModuleType] = module_type

providers `instance-attribute` ¶

providers: Final[Sequence[ProviderSpec]] = providers

imports `instance-attribute` ¶

imports: Final[Sequence[ModuleType | DynamicModule]] = (
    imports
)

exports `instance-attribute` ¶

exports: Final[
    Sequence[type[object] | ModuleType | DynamicModule]
] = exports

extensions `instance-attribute` ¶

extensions: Final[Sequence[ModuleExtension]] = extensions

is_global `instance-attribute` ¶

is_global: Final[bool] = is_global

name `property` ¶

name: str

provider `property` ¶

provider: BaseProvider

Get the aggregated provider for this module.

This property returns the provider created by create_provider(). Must be called after create_provider() has been invoked.

RAISES	DESCRIPTION
`RuntimeError`	If create_provider() has not been called yet.

create_provider ¶

create_provider(
    context: dict[Any, Any] | None,
    builder: ActivationBuilder,
    provider_filter: IProviderFilter,
) -> BaseProvider

Create aggregated provider with activation filtering applied.

PARAMETER	DESCRIPTION
`context`	Context dict for activation decisions. TYPE: `dict[Any, Any] \| None`
`builder`	Activation builder for checking if types are registered. TYPE: `ActivationBuilder`
`provider_filter`	Filter strategy for conditional provider activation. TYPE: `IProviderFilter`

RETURNS	DESCRIPTION
`BaseProvider`	BaseProvider with only active providers aggregated.

Source code in src/waku/modules/_module.py

def create_provider(
    self,
    context: dict[Any, Any] | None,
    builder: ActivationBuilder,
    provider_filter: IProviderFilter,
) -> BaseProvider:
    """Create aggregated provider with activation filtering applied.

    Args:
        context: Context dict for activation decisions.
        builder: Activation builder for checking if types are registered.
        provider_filter: Filter strategy for conditional provider activation.

    Returns:
        BaseProvider with only active providers aggregated.
    """
    active_providers = provider_filter.filter(
        list(self.providers),
        context=context,
        module_type=self.target,
        builder=builder,
    )

    cls = cast('type[_ModuleProvider]', type(f'{self.name}Provider', (_ModuleProvider,), {}))
    self._provider = cls(active_providers)
    return self._provider

ModuleRegistry ¶

ModuleRegistry(
    *,
    compiler: ModuleCompiler,
    root_module: Module,
    modules: dict[UUID, Module],
    providers: list[BaseProvider],
    adjacency: AdjacencyMatrix,
)

Immutable registry and graph for module queries, traversal, and lookups.

Source code in src/waku/modules/_registry.py

def __init__(
    self,
    *,
    compiler: ModuleCompiler,
    root_module: Module,
    modules: dict[UUID, Module],
    providers: list[BaseProvider],
    adjacency: AdjacencyMatrix,
) -> None:
    self._compiler = compiler
    self._root_module = root_module
    self._modules = modules
    self._providers = tuple(providers)
    self._adjacency = adjacency
    self._parent_to_module = self._build_parent_mapping(modules)

root_module `property` ¶

root_module: Module

modules `property` ¶

modules: tuple[Module, ...]

providers `property` ¶

providers: tuple[BaseProvider, ...]

compiler `property` ¶

compiler: ModuleCompiler

get ¶

get(module_type: ModuleType | DynamicModule) -> Module

Source code in src/waku/modules/_registry.py

def get(self, module_type: ModuleType | DynamicModule) -> Module:
    # For plain module classes, check if they're registered via parent mapping first.
    # This handles the case where ConfigModule.register() was imported,
    # but ConfigModule (the class) is being exported.
    if isinstance(module_type, type) and module_type in self._parent_to_module:
        return self._parent_to_module[module_type]

    module_id = self.compiler.extract_metadata(module_type)[1].id
    return self.get_by_id(module_id)

get_by_id ¶

get_by_id(module_id: UUID) -> Module

Source code in src/waku/modules/_registry.py

def get_by_id(self, module_id: UUID) -> Module:
    module = self._modules.get(module_id)
    if module is None:
        msg = f'Module with ID {module_id} is not registered in the graph.'
        raise KeyError(msg)
    return module

traverse ¶

traverse(from_: Module | None = None) -> Iterator[Module]

Traverse the module graph in depth-first post-order (children before parent) recursively.

PARAMETER	DESCRIPTION
`from_`	Start module (default: root) TYPE: `Module \| None` DEFAULT: `None`

YIELDS	DESCRIPTION
`Module`	Each traversed module (post-order) TYPE:: `Module`

Source code in src/waku/modules/_registry.py

def traverse(self, from_: Module | None = None) -> Iterator[Module]:
    """Traverse the module graph in depth-first post-order (children before parent) recursively.

    Args:
        from_: Start module (default: root)

    Yields:
        Module: Each traversed module (post-order)
    """
    start_module = from_ or self._root_module
    visited: set[UUID] = set()

    def _dfs(module: Module) -> Iterator[Module]:
        if module.id in visited:
            return

        visited.add(module.id)

        # Process children first (maintain original order)
        neighbor_ids = self._adjacency[module.id]
        for neighbor_id in neighbor_ids:
            if neighbor_id == module.id:
                continue
            neighbor = self.get_by_id(neighbor_id)
            yield from _dfs(neighbor)

        # Process current module after children (post-order)
        yield module

    yield from _dfs(start_module)

ModuleRegistryBuilder ¶

ModuleRegistryBuilder(
    root_module_type: ModuleType,
    compiler: ModuleCompiler | None = None,
    context: dict[Any, Any] | None = None,
    provider_filter: IProviderFilter | None = None,
    app_extensions: Sequence[ApplicationExtension] = (),
)

Source code in src/waku/modules/_registry_builder.py

def __init__(
    self,
    root_module_type: ModuleType,
    compiler: ModuleCompiler | None = None,
    context: dict[Any, Any] | None = None,
    provider_filter: IProviderFilter | None = None,
    app_extensions: Sequence[ApplicationExtension] = (),
) -> None:
    self._compiler: Final = compiler or ModuleCompiler()
    self._root_module_type: Final = root_module_type
    self._context: Final = context
    self._provider_filter: Final[IProviderFilter] = provider_filter or ProviderFilter()
    self._app_extensions: Final = app_extensions
    self._modules: dict[UUID, Module] = {}
    self._providers: list[BaseProvider] = []

    self._metadata_cache: dict[ModuleType | DynamicModule, tuple[ModuleType, ModuleMetadata]] = {}
    self._builder: Final = _ActivationBuilder()

build ¶

build() -> ModuleRegistry

Source code in src/waku/modules/_registry_builder.py

def build(self) -> ModuleRegistry:
    modules, adjacency = self._collect_modules()
    self._execute_registration_hooks(modules)
    self._build_type_registry(modules)
    root_module = self._register_modules(modules)
    return self._build_registry(root_module, adjacency)

module ¶

module(
    *,
    providers: Sequence[ProviderSpec] = (),
    imports: Sequence[ModuleType | DynamicModule] = (),
    exports: Sequence[
        type[object] | ModuleType | DynamicModule
    ] = (),
    extensions: Sequence[ModuleExtension] = (),
    is_global: bool = False,
) -> Callable[[type[_T]], type[_T]]

Decorator to define a module.

PARAMETER	DESCRIPTION
`providers`	Sequence of providers for dependency injection. TYPE: `Sequence[ProviderSpec]` DEFAULT: `()`
`imports`	Sequence of modules imported by this module. TYPE: `Sequence[ModuleType \| DynamicModule]` DEFAULT: `()`
`exports`	Sequence of types or modules exported by this module. TYPE: `Sequence[type[object] \| ModuleType \| DynamicModule]` DEFAULT: `()`
`extensions`	Sequence of module extensions for lifecycle hooks. TYPE: `Sequence[ModuleExtension]` DEFAULT: `()`
`is_global`	Whether this module is global or not. TYPE: `bool` DEFAULT: `False`

Source code in src/waku/modules/_metadata.py

def module(
    *,
    providers: Sequence[ProviderSpec] = (),
    imports: Sequence[ModuleType | DynamicModule] = (),
    exports: Sequence[type[object] | ModuleType | DynamicModule] = (),
    extensions: Sequence[ModuleExtension] = (),
    is_global: bool = False,
) -> Callable[[type[_T]], type[_T]]:
    """Decorator to define a module.

    Args:
        providers: Sequence of providers for dependency injection.
        imports: Sequence of modules imported by this module.
        exports: Sequence of types or modules exported by this module.
        extensions: Sequence of module extensions for lifecycle hooks.
        is_global: Whether this module is global or not.
    """

    def decorator(cls: type[_T]) -> type[_T]:
        metadata = ModuleMetadata(
            providers=list(providers),
            imports=list(imports),
            exports=list(exports),
            extensions=list(extensions),
            is_global=is_global,
        )
        for extension in metadata.extensions:
            if isinstance(extension, OnModuleConfigure):
                extension.on_module_configure(metadata)

        setattr(cls, _MODULE_METADATA_KEY, metadata)
        return cls

    return decorator

testing ¶

override ¶

override(
    container: AsyncContainer,
    *providers: BaseProvider,
    context: dict[Any, Any] | None = None,
) -> Iterator[None]

Temporarily override providers and/or context in an AsyncContainer for testing.

PARAMETER	DESCRIPTION
`container`	The container whose providers/context will be overridden. TYPE: `AsyncContainer`
`*providers`	Providers to override in the container. TYPE: `BaseProvider` DEFAULT: `()`
`context`	Context values to override. TYPE: `dict[Any, Any] \| None` DEFAULT: `None`

YIELDS	DESCRIPTION
`None`	Context in which the container uses the overridden providers/context. TYPE:: `None`

Example

from waku import WakuFactory, module
from waku.di import Scope, singleton
from waku.testing import override


class Service: ...


class ServiceOverride(Service): ...


# Override providers
with override(application.container, singleton(ServiceOverride, provided_type=Service)):
    service = await application.container.get(Service)
    assert isinstance(service, ServiceOverride)

# Override context
with override(application.container, context={int: 123}):
    ...

RAISES	DESCRIPTION
`ValueError`	If container is not at root (APP) scope.

Source code in src/waku/testing.py

@contextmanager
def override(
    container: AsyncContainer,
    *providers: BaseProvider,
    context: dict[Any, Any] | None = None,
) -> Iterator[None]:
    """Temporarily override providers and/or context in an AsyncContainer for testing.

    Args:
        container: The container whose providers/context will be overridden.
        *providers: Providers to override in the container.
        context: Context values to override.

    Yields:
        None: Context in which the container uses the overridden providers/context.

    Example:
        ```python
        from waku import WakuFactory, module
        from waku.di import Scope, singleton
        from waku.testing import override


        class Service: ...


        class ServiceOverride(Service): ...


        # Override providers
        with override(application.container, singleton(ServiceOverride, provided_type=Service)):
            service = await application.container.get(Service)
            assert isinstance(service, ServiceOverride)

        # Override context
        with override(application.container, context={int: 123}):
            ...
        ```

    Raises:
        ValueError: If container is not at root (APP) scope.
    """
    if container.scope != Scope.APP:
        msg = (
            f'override() only supports root (APP scope) containers, '
            f'got {container.scope.name} scope. '
            f'Use application.container instead of a scoped container.'
        )
        raise ValueError(msg)

    _mark_as_overrides(providers)

    original_context = cast('dict[Any, Any]', container._context)  # noqa: SLF001
    merged_context = {**original_context, **(context or {})}
    context_override_types = frozenset(context.keys()) if context else frozenset()

    new_container = make_async_container(
        _container_provider(container, context_override_types),
        *providers,
        context=merged_context,
        start_scope=container.scope,
        validation_settings=STRICT_VALIDATION,
    )

    # Only copy cache when no providers are overridden (context-only override)
    # Provider overrides may have transitive effects, so rebuild everything
    if not providers:
        _copy_cache(container, new_container, context_override_types)

    _swap(container, new_container)
    yield
    _swap(new_container, container)

create_test_app `async` ¶

create_test_app(
    *,
    base: ModuleType | DynamicModule | None = None,
    providers: Sequence[ProviderSpec] = (),
    imports: Sequence[ModuleType | DynamicModule] = (),
    extensions: Sequence[ModuleExtension] = (),
    app_extensions: Sequence[
        ApplicationExtension
    ] = DEFAULT_EXTENSIONS,
    context: dict[Any, Any] | None = None,
) -> AsyncIterator[WakuApplication]

Create a minimal test application with given configuration.

Useful for testing extensions and module configurations in isolation without needing to set up a full application structure.

PARAMETER	DESCRIPTION
`base`	Base module to build upon. When provided, the test module imports this module and providers act as overrides. TYPE: `ModuleType \| DynamicModule \| None` DEFAULT: `None`
`providers`	Providers to register in the test module. When `base` is provided, these override existing providers. TYPE: `Sequence[ProviderSpec]` DEFAULT: `()`
`imports`	Additional modules to import into the test module. TYPE: `Sequence[ModuleType \| DynamicModule]` DEFAULT: `()`
`extensions`	Module extensions to register. TYPE: `Sequence[ModuleExtension]` DEFAULT: `()`
`app_extensions`	Application extensions to register (default: DEFAULT_EXTENSIONS). TYPE: `Sequence[ApplicationExtension]` DEFAULT: `DEFAULT_EXTENSIONS`
`context`	Context values to pass to the container. TYPE: `dict[Any, Any] \| None` DEFAULT: `None`

YIELDS	DESCRIPTION
`AsyncIterator[WakuApplication]`	Initialized WakuApplication.

Example

from waku.testing import create_test_app
from waku.di import singleton


class IRepository(Protocol):
    async def get(self, id: str) -> Entity: ...


class FakeRepository(IRepository):
    async def get(self, id: str) -> Entity:
        return Entity(id=id)


# Create test app from scratch
async def test_my_extension():
    extension = MyExtension().bind(SomeEvent, SomeHandler)

    async with create_test_app(
        extensions=[extension],
        providers=[singleton(IRepository, FakeRepository)],
    ) as app:
        service = await app.container.get(MyService)
        result = await service.do_something()
        assert result == expected


# Create test app based on existing module with overrides
async def test_with_base_module():
    async with create_test_app(
        base=AppModule,
        providers=[singleton(IRepository, FakeRepository)],
    ) as app:
        # FakeRepository replaces the real one from AppModule
        repo = await app.container.get(IRepository)
        assert isinstance(repo, FakeRepository)

Source code in src/waku/testing.py

@asynccontextmanager
async def create_test_app(
    *,
    base: ModuleType | DynamicModule | None = None,
    providers: Sequence[ProviderSpec] = (),
    imports: Sequence[ModuleType | DynamicModule] = (),
    extensions: Sequence[ModuleExtension] = (),
    app_extensions: Sequence[ApplicationExtension] = DEFAULT_EXTENSIONS,
    context: dict[Any, Any] | None = None,
) -> AsyncIterator[WakuApplication]:
    """Create a minimal test application with given configuration.

    Useful for testing extensions and module configurations in isolation
    without needing to set up a full application structure.

    Args:
        base: Base module to build upon. When provided, the test module
            imports this module and providers act as overrides.
        providers: Providers to register in the test module.
            When `base` is provided, these override existing providers.
        imports: Additional modules to import into the test module.
        extensions: Module extensions to register.
        app_extensions: Application extensions to register (default: DEFAULT_EXTENSIONS).
        context: Context values to pass to the container.

    Yields:
        Initialized WakuApplication.

    Example:
        ```python
        from waku.testing import create_test_app
        from waku.di import singleton


        class IRepository(Protocol):
            async def get(self, id: str) -> Entity: ...


        class FakeRepository(IRepository):
            async def get(self, id: str) -> Entity:
                return Entity(id=id)


        # Create test app from scratch
        async def test_my_extension():
            extension = MyExtension().bind(SomeEvent, SomeHandler)

            async with create_test_app(
                extensions=[extension],
                providers=[singleton(IRepository, FakeRepository)],
            ) as app:
                service = await app.container.get(MyService)
                result = await service.do_something()
                assert result == expected


        # Create test app based on existing module with overrides
        async def test_with_base_module():
            async with create_test_app(
                base=AppModule,
                providers=[singleton(IRepository, FakeRepository)],
            ) as app:
                # FakeRepository replaces the real one from AppModule
                repo = await app.container.get(IRepository)
                assert isinstance(repo, FakeRepository)
        ```
    """
    all_imports = list(imports)
    if base is not None:
        all_imports.insert(0, base)

    override_providers = list(providers)
    if base is not None:
        _mark_as_overrides(override_providers)

    @module(
        providers=override_providers,
        imports=all_imports,
        extensions=list(extensions),
    )
    class _TestModule:
        pass

    app = WakuFactory(_TestModule, context=context, extensions=app_extensions).create()
    async with app:
        yield app

validation ¶

ValidationRule ¶

Bases: Protocol

validate ¶

validate(
    context: ValidationContext,
) -> list[ValidationError]

Source code in src/waku/validation/_abc.py

def validate(self, context: ValidationContext) -> list[ValidationError]: ...

ValidationError ¶

Bases: WakuError

ValidationExtension ¶

ValidationExtension(
    rules: Sequence[ValidationRule], *, strict: bool = True
)

Bases: AfterApplicationInit

Source code in src/waku/validation/_extension.py

def __init__(self, rules: Sequence[ValidationRule], *, strict: bool = True) -> None:
    self.rules = rules
    self.strict: Final = strict

rules `instance-attribute` ¶

rules = rules

strict `instance-attribute` ¶

strict: Final = strict

after_app_init `async` ¶

after_app_init(app: WakuApplication) -> None

Source code in src/waku/validation/_extension.py

async def after_app_init(self, app: WakuApplication) -> None:
    context = ValidationContext(app=app)

    errors_chain = chain.from_iterable(rule.validate(context) for rule in self.rules)
    if errors := list(errors_chain):
        self._raise(errors)

rules ¶

DependenciesAccessibleRule ¶

DependenciesAccessibleRule(cache_size: int = 1000)

Bases: ValidationRule

Validates that all dependencies required by providers are accessible.

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(self, cache_size: int = 1000) -> None:
    self._cache = LRUCache[set[type[object]]](cache_size)
    self._types_extractor = ModuleTypesExtractor(self._cache)

validate ¶

validate(
    context: ValidationContext,
) -> list[ValidationError]

Source code in src/waku/validation/rules/dependency_accessible.py

@override
def validate(self, context: ValidationContext) -> list[ValidationError]:
    self._cache.clear()

    registry = context.app.registry
    modules = list(registry.modules)
    container = context.app.container

    strategies: list[AccessibilityStrategy] = [
        GlobalProvidersStrategy(modules, container, self._types_extractor, registry),
        LocalProvidersStrategy(self._types_extractor),
        ContextVarsStrategy(self._types_extractor),
        ImportedModulesStrategy(registry, self._types_extractor),
    ]

    checker = DependencyAccessChecker(strategies)
    errors: list[ValidationError] = []

    for module in modules:
        for factory in module.provider.factories:
            inaccessible_deps = checker.find_inaccessible_dependencies(
                dependencies=factory.dependencies,
                module=module,
            )
            errors.extend(
                DependencyInaccessibleError(
                    required_type=dep_type,
                    required_by=factory.source,
                    from_module=module,
                )
                for dep_type in inaccessible_deps
            )

    return errors

DependencyInaccessibleError ¶

DependencyInaccessibleError(
    required_type: type[object],
    required_by: object,
    from_module: Module,
)

Bases: ValidationError

Error indicating a dependency is not accessible to a provider/module.

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(
    self,
    required_type: type[object],
    required_by: object,
    from_module: Module,
) -> None:
    self.required_type = required_type
    self.required_by = required_by
    self.from_module = from_module
    super().__init__(str(self))

required_type `instance-attribute` ¶

required_type = required_type

required_by `instance-attribute` ¶

required_by = required_by

from_module `instance-attribute` ¶

from_module = from_module

dependency_accessible ¶

DependencyInaccessibleError ¶

DependencyInaccessibleError(
    required_type: type[object],
    required_by: object,
    from_module: Module,
)

Bases: ValidationError

Error indicating a dependency is not accessible to a provider/module.

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(
    self,
    required_type: type[object],
    required_by: object,
    from_module: Module,
) -> None:
    self.required_type = required_type
    self.required_by = required_by
    self.from_module = from_module
    super().__init__(str(self))

required_type `instance-attribute` ¶

required_type = required_type

required_by `instance-attribute` ¶

required_by = required_by

from_module `instance-attribute` ¶

from_module = from_module

AccessibilityStrategy ¶

Bases: ABC

Strategy for checking if a type is accessible to a module.

is_accessible `abstractmethod` ¶

is_accessible(
    required_type: type[object], module: Module
) -> bool

Check if the required type is accessible to the given module.

Source code in src/waku/validation/rules/dependency_accessible.py

@abstractmethod
def is_accessible(self, required_type: type[object], module: Module) -> bool:
    """Check if the required type is accessible to the given module."""

GlobalProvidersStrategy ¶

GlobalProvidersStrategy(
    modules: Sequence[Module],
    container: AsyncContainer,
    types_extractor: ModuleTypesExtractor,
    registry: ModuleRegistry,
)

Bases: AccessibilityStrategy

Check if type is provided by a global module or APP-scoped context.

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(
    self,
    modules: Sequence[Module],
    container: AsyncContainer,
    types_extractor: ModuleTypesExtractor,
    registry: ModuleRegistry,
) -> None:
    self._global_types = self._build_global_types(modules, container, types_extractor, registry)

is_accessible ¶

is_accessible(
    required_type: type[object], module: Module
) -> bool

Source code in src/waku/validation/rules/dependency_accessible.py

@override
def is_accessible(self, required_type: type[object], module: Module) -> bool:
    return required_type in self._global_types

LocalProvidersStrategy ¶

LocalProvidersStrategy(
    types_extractor: ModuleTypesExtractor,
)

Bases: AccessibilityStrategy

Check if type is provided by the module itself.

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(self, types_extractor: ModuleTypesExtractor) -> None:
    self._types_extractor = types_extractor

is_accessible ¶

is_accessible(
    required_type: type[object], module: Module
) -> bool

Source code in src/waku/validation/rules/dependency_accessible.py

@override
def is_accessible(self, required_type: type[object], module: Module) -> bool:
    return required_type in self._types_extractor.get_provided_types(module)

ContextVarsStrategy ¶

ContextVarsStrategy(types_extractor: ModuleTypesExtractor)

Bases: AccessibilityStrategy

Check if type is provided by application or request container context.

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(self, types_extractor: ModuleTypesExtractor) -> None:
    self._types_extractor = types_extractor

is_accessible ¶

is_accessible(
    required_type: type[object], module: Module
) -> bool

Source code in src/waku/validation/rules/dependency_accessible.py

@override
def is_accessible(self, required_type: type[object], module: Module) -> bool:
    return required_type in self._types_extractor.get_context_vars(module)

ImportedModulesStrategy ¶

ImportedModulesStrategy(
    registry: ModuleRegistry,
    types_extractor: ModuleTypesExtractor,
)

Bases: AccessibilityStrategy

Check if type is accessible via imported modules (direct export or re-export).

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(self, registry: ModuleRegistry, types_extractor: ModuleTypesExtractor) -> None:
    self._registry = registry
    self._types_extractor = types_extractor

is_accessible ¶

is_accessible(
    required_type: type[object], module: Module
) -> bool

Source code in src/waku/validation/rules/dependency_accessible.py

@override
def is_accessible(self, required_type: type[object], module: Module) -> bool:
    for imported in module.imports:
        imported_module = self._registry.get(imported)
        if self._is_directly_exported(required_type, imported_module):
            return True
        if self._is_reexported(required_type, imported_module):
            return True
    return False

DependencyAccessChecker ¶

DependencyAccessChecker(
    strategies: Sequence[AccessibilityStrategy],
)

Handles dependency accessibility checks between modules.

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(self, strategies: Sequence[AccessibilityStrategy]) -> None:
    self._strategies = strategies

find_inaccessible_dependencies ¶

find_inaccessible_dependencies(
    dependencies: Sequence[DependencyKey], module: Module
) -> Iterable[type[object]]

Source code in src/waku/validation/rules/dependency_accessible.py

def find_inaccessible_dependencies(
    self,
    dependencies: Sequence[DependencyKey],
    module: Module,
) -> Iterable[type[object]]:
    for dependency in dependencies:
        if not self._is_accessible(dependency.type_hint, module):
            yield dependency.type_hint

DependenciesAccessibleRule ¶

DependenciesAccessibleRule(cache_size: int = 1000)

Bases: ValidationRule

Validates that all dependencies required by providers are accessible.

Source code in src/waku/validation/rules/dependency_accessible.py

def __init__(self, cache_size: int = 1000) -> None:
    self._cache = LRUCache[set[type[object]]](cache_size)
    self._types_extractor = ModuleTypesExtractor(self._cache)

validate ¶

validate(
    context: ValidationContext,
) -> list[ValidationError]

Source code in src/waku/validation/rules/dependency_accessible.py

@override
def validate(self, context: ValidationContext) -> list[ValidationError]:
    self._cache.clear()

    registry = context.app.registry
    modules = list(registry.modules)
    container = context.app.container

    strategies: list[AccessibilityStrategy] = [
        GlobalProvidersStrategy(modules, container, self._types_extractor, registry),
        LocalProvidersStrategy(self._types_extractor),
        ContextVarsStrategy(self._types_extractor),
        ImportedModulesStrategy(registry, self._types_extractor),
    ]

    checker = DependencyAccessChecker(strategies)
    errors: list[ValidationError] = []

    for module in modules:
        for factory in module.provider.factories:
            inaccessible_deps = checker.find_inaccessible_dependencies(
                dependencies=factory.dependencies,
                module=module,
            )
            errors.extend(
                DependencyInaccessibleError(
                    required_type=dep_type,
                    required_by=factory.source,
                    from_module=module,
                )
                for dep_type in inaccessible_deps
            )

    return errors

waku ¶

WakuApplication ¶

container property ¶

registry property ¶

initialize async ¶

close async ¶

WakuFactory ¶

create ¶

DynamicModule dataclass ¶

providers class-attribute instance-attribute ¶

imports class-attribute instance-attribute ¶

exports class-attribute instance-attribute ¶

extensions class-attribute instance-attribute ¶

is_global class-attribute instance-attribute ¶

id class-attribute instance-attribute ¶

parent_module instance-attribute ¶

Module ¶

id instance-attribute ¶

target instance-attribute ¶

providers instance-attribute ¶

imports instance-attribute ¶

exports instance-attribute ¶

extensions instance-attribute ¶

is_global instance-attribute ¶

name property ¶

provider property ¶

create_provider ¶

module ¶

application ¶

WakuApplication ¶

container property ¶

registry property ¶

initialize async ¶

close async ¶

cqrs ¶

NextHandlerType module-attribute ¶

Event dataclass ¶

event_id class-attribute instance-attribute ¶

INotification ¶

IPipelineBehavior ¶

handle abstractmethod async ¶

IRequest ¶

Request dataclass ¶

request_id class-attribute instance-attribute ¶

Response dataclass ¶

EventHandler ¶

handle abstractmethod async ¶

INotificationHandler ¶

handle async ¶

Mediator ¶

send async ¶

publish async ¶

IMediator ¶

publish abstractmethod async ¶

send abstractmethod async ¶

IPublisher ¶

publish abstractmethod async ¶

ISender ¶

send abstractmethod async ¶

MediatorConfig dataclass ¶

mediator_implementation_type class-attribute instance-attribute ¶

event_publisher class-attribute instance-attribute ¶

pipeline_behaviors class-attribute instance-attribute ¶

MediatorExtension ¶

request_map property ¶

event_map property ¶

behavior_map property ¶

bind_request ¶

bind_event ¶

on_module_configure ¶

MediatorModule ¶

register classmethod ¶

IRequestHandler ¶

handle async ¶

RequestHandler ¶

handle abstractmethod async ¶

contracts ¶

NotificationT module-attribute ¶

NextHandlerType module-attribute ¶

RequestT module-attribute ¶

container `property` ¶

registry `property` ¶

initialize `async` ¶

close `async` ¶

DynamicModule `dataclass` ¶

providers `class-attribute` `instance-attribute` ¶

imports `class-attribute` `instance-attribute` ¶

exports `class-attribute` `instance-attribute` ¶

extensions `class-attribute` `instance-attribute` ¶

is_global `class-attribute` `instance-attribute` ¶

id `class-attribute` `instance-attribute` ¶

parent_module `instance-attribute` ¶

id `instance-attribute` ¶

target `instance-attribute` ¶

providers `instance-attribute` ¶

imports `instance-attribute` ¶

exports `instance-attribute` ¶

extensions `instance-attribute` ¶

is_global `instance-attribute` ¶

name `property` ¶

provider `property` ¶

container `property` ¶

registry `property` ¶

initialize `async` ¶

close `async` ¶

NextHandlerType `module-attribute` ¶

Event `dataclass` ¶

event_id `class-attribute` `instance-attribute` ¶

handle `abstractmethod` `async` ¶

Request `dataclass` ¶

request_id `class-attribute` `instance-attribute` ¶

Response `dataclass` ¶

handle `abstractmethod` `async` ¶

handle `async` ¶

send `async` ¶

publish `async` ¶

publish `abstractmethod` `async` ¶

send `abstractmethod` `async` ¶

publish `abstractmethod` `async` ¶

send `abstractmethod` `async` ¶

MediatorConfig `dataclass` ¶

mediator_implementation_type `class-attribute` `instance-attribute` ¶

event_publisher `class-attribute` `instance-attribute` ¶

pipeline_behaviors `class-attribute` `instance-attribute` ¶

request_map `property` ¶

event_map `property` ¶

behavior_map `property` ¶

register `classmethod` ¶

handle `async` ¶

handle `abstractmethod` `async` ¶

NotificationT `module-attribute` ¶

NextHandlerType `module-attribute` ¶

RequestT `module-attribute` ¶

ResponseT `module-attribute` ¶

Event `dataclass` ¶

event_id `class-attribute` `instance-attribute` ¶

handle `abstractmethod` `async` ¶

Request `dataclass` ¶

request_id `class-attribute` `instance-attribute` ¶

Response `dataclass` ¶

NotificationT `module-attribute` ¶

Event `dataclass` ¶

event_id `class-attribute` `instance-attribute` ¶

NotificationT `module-attribute` ¶

NextHandlerType `module-attribute` ¶

handle `abstractmethod` `async` ¶

ResponseT `module-attribute` ¶

RequestT `module-attribute` ¶

Request `dataclass` ¶

request_id `class-attribute` `instance-attribute` ¶

Response `dataclass` ¶

handle `abstractmethod` `async` ¶

handle `async` ¶

handle `abstractmethod` `async` ¶

EventMapRegistry `module-attribute` ¶

EventMapEntry `dataclass` ¶

di_lookup_type `instance-attribute` ¶

handler_types `class-attribute` `instance-attribute` ¶

registry `property` ¶

request_type `instance-attribute` ¶