Skip to content

waku

Application 

Application(container, lifespan, extensions)
Source code in src/waku/application.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    container: ApplicationContainer,
    lifespan: Sequence[LifespanFunc | LifespanWrapper],
    extensions: Sequence[ApplicationExtension],
) -> None:
    self._container = container
    self._lifespan = tuple(
        LifespanWrapper(lifespan_func) if not isinstance(lifespan_func, LifespanWrapper) else lifespan_func
        for lifespan_func in lifespan
    )
    self._extensions = list(extensions)

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

container property 

container

initialize async 

initialize()
Source code in src/waku/application.py 
37
38
39
40
41
42
async def initialize(self) -> None:
    if self._initialized:
        return
    await self._call_on_init_extensions()
    self._initialized = True
    await self._call_after_init_extensions()

ApplicationFactory

create classmethod 

create(
    root_module,
    /,
    dependency_provider,
    lifespan=(),
    extensions=DEFAULT_EXTENSIONS,
)
Source code in src/waku/factory.py 
19
20
21
22
23
24
25
26
27
28
29
@classmethod
def create(
    cls,
    root_module: ModuleType,
    /,
    dependency_provider: DependencyProvider,
    lifespan: Sequence[LifespanFunc] = (),
    extensions: Sequence[ApplicationExtension] = DEFAULT_EXTENSIONS,
) -> Application:
    container = cls._build_container(root_module, dependency_provider)
    return Application(container, lifespan, extensions)

DynamicModule dataclass 

DynamicModule(
    *,
    providers=list(),
    imports=list(),
    exports=list(),
    extensions=list(),
    is_global=False,
    id=uuid4(),
    parent_module,
)

Bases: ModuleMetadata

providers class-attribute instance-attribute 

providers = field(default_factory=list)

List of providers for dependency injection.

imports class-attribute instance-attribute 

imports = field(default_factory=list)

List of modules imported by this module.

exports class-attribute instance-attribute 

exports = field(default_factory=list)

List of types or modules exported by this module.

extensions class-attribute instance-attribute 

extensions = field(default_factory=list)

List of module extensions for lifecycle hooks.

is_global class-attribute instance-attribute 

is_global = False

Whether this module is global or not.

id class-attribute instance-attribute 

id = field(default_factory=uuid4)

parent_module instance-attribute 

parent_module

module 

module(
    *,
    providers=(),
    imports=(),
    exports=(),
    extensions=(),
    is_global=False,
)

Decorator to define a module.

PARAMETER DESCRIPTION
providers

Sequence of providers for dependency injection.

TYPE: Sequence[Provider[Any]] 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[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 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def module(
    *,
    providers: Sequence[Provider[Any]] = (),
    imports: Sequence[ModuleType | DynamicModule] = (),
    exports: Sequence[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,
        )
        setattr(cls, MODULE_METADATA_KEY, metadata)
        return cls

    return decorator

application

Application 

Application(container, lifespan, extensions)
Source code in src/waku/application.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    container: ApplicationContainer,
    lifespan: Sequence[LifespanFunc | LifespanWrapper],
    extensions: Sequence[ApplicationExtension],
) -> None:
    self._container = container
    self._lifespan = tuple(
        LifespanWrapper(lifespan_func) if not isinstance(lifespan_func, LifespanWrapper) else lifespan_func
        for lifespan_func in lifespan
    )
    self._extensions = list(extensions)

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

container property 

container

initialize async 

initialize()
Source code in src/waku/application.py 
37
38
39
40
41
42
async def initialize(self) -> None:
    if self._initialized:
        return
    await self._call_on_init_extensions()
    self._initialized = True
    await self._call_after_init_extensions()

container

ApplicationContainer 

ApplicationContainer(dependency_provider, root_module)
Source code in src/waku/container.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, dependency_provider: DependencyProvider, root_module: ModuleType) -> None:
    self._dependency_provider = dependency_provider

    self._modules: dict[UUID, Module] = {}
    self._compiler = ModuleCompiler()

    self._root_module = Module(*self._compiler.extract_metadata(root_module))
    self._graph = ModuleGraph(self._root_module)

    self._dependency_provider.register(Object(dependency_provider, DependencyProvider))
    self._dependency_provider.register(Object(self, ApplicationContainer))

    self._exit_stack = AsyncExitStack()

graph property 

graph

compiler property 

compiler

add_module 

add_module(module_type)
Source code in src/waku/container.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def add_module(self, module_type: ModuleType | DynamicModule) -> tuple[Module, bool]:
    type_, metadata = self._compiler.extract_metadata(module_type)
    if self.has(metadata.id):
        return self._modules[metadata.id], False

    for extension in metadata.extensions:
        if isinstance(extension, OnModuleConfigure):
            extension.on_module_configure(metadata)

    module = Module(type_, metadata)

    self._modules[module.id] = module
    self._graph.add_node(module)

    for provider in module.providers:
        self._dependency_provider.register(provider)

    return module, True

has 

has(id_)
Source code in src/waku/container.py 
58
59
def has(self, id_: UUID) -> bool:
    return id_ in self._modules

get_module 

get_module(module_type)
Source code in src/waku/container.py 
61
62
def get_module(self, module_type: ModuleType | DynamicModule) -> Module:
    return self._modules[self._compiler.extract_metadata(module_type)[1].id]

get_module_by_id 

get_module_by_id(id_)
Source code in src/waku/container.py 
64
65
def get_module_by_id(self, id_: UUID) -> Module:
    return self._modules[id_]

get_modules 

get_modules(from_=None)
Source code in src/waku/container.py 
67
68
def get_modules(self, from_: Module | None = None) -> Iterable[Module]:
    return self._graph.traverse(from_)

override 

override(*providers)
Source code in src/waku/container.py 
70
71
72
73
@contextmanager
def override(self, *providers: Provider[Any]) -> Iterator[None]:
    with self._dependency_provider.override(*providers):
        yield

is_global_module 

is_global_module(module)
Source code in src/waku/container.py 
83
84
def is_global_module(self, module: Module) -> bool:
    return module.is_global or module is self._root_module

context async 

context(context=None)
Source code in src/waku/container.py 
86
87
88
89
@asynccontextmanager
async def context(self, context: Mapping[Any, Any] | None = None) -> AsyncIterator[InjectionContext]:
    async with self._dependency_provider.context(context) as ctx:
        yield ctx

get async 

get(type_)
Source code in src/waku/container.py 
91
92
async def get(self, type_: type[_T]) -> _T:
    return await self._dependency_provider.get(type_)

get_all async 

get_all(type_)
Source code in src/waku/container.py 
94
95
async def get_all(self, type_: type[_T]) -> Iterable[_T]:
    return await self._dependency_provider.get_all(type_)

contrib

asgi

Scope module-attribute 

Scope = MutableMapping[str, Any]

Message module-attribute 

Message = MutableMapping[str, Any]

Receive module-attribute 

Receive = Callable[[], Awaitable[Message]]

Send module-attribute 

Send = Callable[[Message], Awaitable[None]]

ASGIApp module-attribute 

ASGIApp = Callable[[Scope, Receive, Send], Awaitable[None]]

ApplicationMiddleware 

ApplicationMiddleware(app, *, application)
Source code in src/waku/contrib/asgi.py 
22
23
24
def __init__(self, app: ASGIApp, *, application: Application) -> None:
    self.app = app
    self._application = application
app instance-attribute 
app = app

litestar

ApplicationMiddleware 

ApplicationMiddleware(app)

Bases: MiddlewareProtocol

Source code in src/waku/contrib/litestar.py 
28
29
def __init__(self, app: ASGIApp) -> None:
    self.app = app
app instance-attribute 
app = app

ApplicationPlugin 

ApplicationPlugin(application)

Bases: InitPluginProtocol

Source code in src/waku/contrib/litestar.py 
55
56
def __init__(self, application: Application) -> None:
    self._application = application
on_app_init 
on_app_init(app_config)
Source code in src/waku/contrib/litestar.py 
63
64
65
66
67
68
def on_app_init(self, app_config: AppConfig) -> AppConfig:
    app_config.state[_STATE_KEY] = self._application
    app_config.middleware.append(ApplicationMiddleware)
    app_config.lifespan.append(self._lifespan)  # pyright: ignore [reportUnknownMemberType]
    app_config.after_exception.append(_after_exception)  # pyright: ignore [reportUnknownMemberType]
    return app_config

di

AnyProvider module-attribute 

AnyProvider = (
    Scoped[_T] | Singleton[_T] | Transient[_T] | Object[_T]
)

Injected

Bases: Generic[_T]

DependencyProvider

Bases: ABC

register abstractmethod 

register(*providers)
Source code in src/waku/di/_providers.py 
112
113
@abstractmethod
def register(self, *providers: Provider[Any]) -> None: ...

try_register abstractmethod 

try_register(*providers)
Source code in src/waku/di/_providers.py 
115
116
@abstractmethod
def try_register(self, *providers: Provider[Any]) -> None: ...

context async 

context(context=None)
Source code in src/waku/di/_providers.py 
118
119
120
121
122
123
124
125
126
127
128
@contextlib.asynccontextmanager
async def context(self, context: Mapping[Any, Any] | None = None) -> AsyncIterator[InjectionContext]:
    if current_ctx := context_var.get(None):
        yield current_ctx
    else:
        async with self._context(context) as ctx:
            token = context_var.set(ctx)
            try:
                yield ctx
            finally:
                context_var.reset(token)

get async 

get(type_)
Source code in src/waku/di/_providers.py 
130
131
132
async def get(self, type_: type[_T]) -> _T:
    async with self.context() as ctx:
        return await ctx.resolve(type_)

get_all async 

get_all(type_)
Source code in src/waku/di/_providers.py 
134
135
136
async def get_all(self, type_: type[_T]) -> list[_T]:
    async with self.context() as ctx:
        return await ctx.resolve_iterable(type_)

override abstractmethod 

override(*providers)
Source code in src/waku/di/_providers.py 
138
139
@abstractmethod
def override(self, *providers: Provider[Any]) -> AbstractContextManager[None]: ...

InjectionContext

Bases: Protocol

resolve async 

resolve(type_)
Source code in src/waku/di/_context.py 
16
async def resolve(self, type_: type[_T]) -> _T: ...

resolve_iterable async 

resolve_iterable(type_)
Source code in src/waku/di/_context.py 
18
async def resolve_iterable(self, type_: type[_T]) -> list[_T]: ...

execute async 

execute(
    function: Callable[_P, Coroutine[Any, Any, _T]],
    dependencies: Iterable[Dependency[object]],
    *args: args,
    **kwargs: kwargs,
) -> _T

execute(
    function: Callable[_P, _T],
    dependencies: Iterable[Dependency[object]],
    *args: args,
    **kwargs: kwargs,
) -> _T

execute(function, dependencies, *args, **kwargs)
Source code in src/waku/di/_context.py 
38
39
40
41
42
43
44
async def execute(
    self,
    function: Callable[_P, Coroutine[Any, Any, _T] | _T],
    dependencies: Iterable[Dependency[object]],
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> _T: ...

Object 

Object(object_, type_=None)

Bases: Provider[_T]

Source code in src/waku/di/_providers.py 
 97
 98
 99
100
101
102
103
def __init__(
    self,
    object_: _T,
    type_: type[_T] | None = None,
) -> None:
    self.type_ = type_ or type(object_)
    self.impl = object_

impl instance-attribute 

impl = object_

type_ instance-attribute 

type_ = type_ or type(object_)

collect_dependencies 

collect_dependencies()
Source code in src/waku/di/_providers.py 
52
53
54
55
56
57
def collect_dependencies(self) -> Sequence[Dependency[object]]:
    try:
        return self._cached_dependencies
    except AttributeError:
        self._cached_dependencies = self._collect_dependencies()
        return self._cached_dependencies

Provider

Bases: Hashable, Protocol[_T]

impl instance-attribute 

impl

type_ instance-attribute 

type_

collect_dependencies 

collect_dependencies()
Source code in src/waku/di/_providers.py 
52
53
54
55
56
57
def collect_dependencies(self) -> Sequence[Dependency[object]]:
    try:
        return self._cached_dependencies
    except AttributeError:
        self._cached_dependencies = self._collect_dependencies()
        return self._cached_dependencies

Scoped 

Scoped(factory, type_=None)

Bases: Provider[_T]

Source code in src/waku/di/_providers.py 
77
78
79
80
81
82
83
def __init__(
    self,
    factory: FactoryType[_T],
    type_: type[_T] | None = None,
) -> None:
    self.impl = factory
    self.type_ = type_ or guess_return_type(factory)

impl instance-attribute 

impl = factory

type_ instance-attribute 

type_ = type_ or guess_return_type(factory)

collect_dependencies 

collect_dependencies()
Source code in src/waku/di/_providers.py 
52
53
54
55
56
57
def collect_dependencies(self) -> Sequence[Dependency[object]]:
    try:
        return self._cached_dependencies
    except AttributeError:
        self._cached_dependencies = self._collect_dependencies()
        return self._cached_dependencies

Singleton 

Singleton(factory, type_=None)

Bases: Scoped[_T]

Source code in src/waku/di/_providers.py 
77
78
79
80
81
82
83
def __init__(
    self,
    factory: FactoryType[_T],
    type_: type[_T] | None = None,
) -> None:
    self.impl = factory
    self.type_ = type_ or guess_return_type(factory)

impl instance-attribute 

impl = factory

type_ instance-attribute 

type_ = type_ or guess_return_type(factory)

collect_dependencies 

collect_dependencies()
Source code in src/waku/di/_providers.py 
52
53
54
55
56
57
def collect_dependencies(self) -> Sequence[Dependency[object]]:
    try:
        return self._cached_dependencies
    except AttributeError:
        self._cached_dependencies = self._collect_dependencies()
        return self._cached_dependencies

Transient 

Transient(factory, type_=None)

Bases: Scoped[_T]

Source code in src/waku/di/_providers.py 
77
78
79
80
81
82
83
def __init__(
    self,
    factory: FactoryType[_T],
    type_: type[_T] | None = None,
) -> None:
    self.impl = factory
    self.type_ = type_ or guess_return_type(factory)

impl instance-attribute 

impl = factory

type_ instance-attribute 

type_ = type_ or guess_return_type(factory)

collect_dependencies 

collect_dependencies()
Source code in src/waku/di/_providers.py 
52
53
54
55
56
57
def collect_dependencies(self) -> Sequence[Dependency[object]]:
    try:
        return self._cached_dependencies
    except AttributeError:
        self._cached_dependencies = self._collect_dependencies()
        return self._cached_dependencies

Dependency dataclass 

Dependency(*, name, type_)

Bases: Generic[_T]

name instance-attribute 

name

type_ instance-attribute 

type_

inner_type cached property 

inner_type

is_iterable cached property 

is_iterable

inject 

inject(function)
Source code in src/waku/di/_inject.py 
27
28
29
def inject(function: Callable[_P, _T]) -> Callable[_P, _T]:
    wrapper = _inject(function)
    return clear_wrapper(wrapper)

contrib

aioinject

AioinjectDependencyProvider 
AioinjectDependencyProvider(container=None)

Bases: DependencyProvider

Source code in src/waku/di/contrib/aioinject.py 
19
20
21
def __init__(self, container: aioinject.Container | None = None) -> None:
    self._container = container or aioinject.Container()
    self._exit_stack = AsyncExitStack()
context async 
context(context=None)
Source code in src/waku/di/_providers.py 
118
119
120
121
122
123
124
125
126
127
128
@contextlib.asynccontextmanager
async def context(self, context: Mapping[Any, Any] | None = None) -> AsyncIterator[InjectionContext]:
    if current_ctx := context_var.get(None):
        yield current_ctx
    else:
        async with self._context(context) as ctx:
            token = context_var.set(ctx)
            try:
                yield ctx
            finally:
                context_var.reset(token)
get async 
get(type_)
Source code in src/waku/di/_providers.py 
130
131
132
async def get(self, type_: type[_T]) -> _T:
    async with self.context() as ctx:
        return await ctx.resolve(type_)
get_all async 
get_all(type_)
Source code in src/waku/di/_providers.py 
134
135
136
async def get_all(self, type_: type[_T]) -> list[_T]:
    async with self.context() as ctx:
        return await ctx.resolve_iterable(type_)
register 
register(*providers)
Source code in src/waku/di/contrib/aioinject.py 
23
24
25
@override_
def register(self, *providers: Provider[Any]) -> None:
    self._container.register(*[self._map_provider(provider) for provider in providers])
try_register 
try_register(*providers)
Source code in src/waku/di/contrib/aioinject.py 
27
28
29
@override_
def try_register(self, *providers: Provider[Any]) -> None:
    self._container.try_register(*[self._map_provider(provider) for provider in providers])
override 
override(*providers)
Source code in src/waku/di/contrib/aioinject.py 
31
32
33
34
35
36
@override_
@contextmanager
def override(self, *providers: Provider[Any]) -> Iterator[None]:
    override_providers = tuple(self._map_provider(provider) for provider in providers)
    with self._container.override(*override_providers):
        yield

ext

DEFAULT_EXTENSIONS module-attribute 

DEFAULT_EXTENSIONS = (
    ValidationExtension(
        [DependenciesAccessible(), DIScopeMismatch()],
        strict=True,
    ),
)

validation

ValidationRule

Bases: Protocol

validate 
validate(context)
Source code in src/waku/ext/validation/_abc.py 
10
def validate(self, context: ValidationContext) -> list[ValidationError]: ...

ValidationError

Bases: Exception

ValidationExtension 

ValidationExtension(rules, *, strict=True)

Bases: AfterApplicationInit

Source code in src/waku/ext/validation/_extension.py 
29
30
31
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 = strict
after_app_init async 
after_app_init(app)
Source code in src/waku/ext/validation/_extension.py 
33
34
35
36
37
38
async def after_app_init(self, app: Application) -> 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

DependenciesAccessible

Bases: ValidationRule

Check if all dependencies of providers are accessible.

This validation rule ensures that all dependencies required by providers are either: 1. Available globally 2. Provided by the current module 3. Provided by any of the imported modules

validate 
validate(context)
Source code in src/waku/ext/validation/rules.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@override
def validate(self, context: ValidationContext) -> list[ValidationError]:
    container = context.app.container

    # fmt: off
    global_providers = {
        provider_type
        for module in container.get_modules()
        for provider_type in _module_provided_types(module)
        if container.is_global_module(module)
    }
    # fmt: on
    global_providers |= {Application, DependencyProvider}

    errors: list[ValidationError] = []
    for module in container.get_modules():
        for provider in module.providers:
            for dependency in provider.collect_dependencies():
                # 1. Dep available globally or provided by current module
                dep_type = dependency.inner_type
                if dep_type in global_providers or dep_type in _module_provided_types(module):
                    continue
                # 2. Dep provided by any of imported modules
                dependency_accessible = any(
                    _is_exported_dependency(dependency, imported_module)
                    for imported_module in container.get_modules(module)
                )
                if not dependency_accessible:
                    err_msg = (
                        f'Provider "{provider!r}" from "{module!r}" depends on "{dep_type!r}" but it\'s not accessible to it\n'
                        f'To resolve this issue:\n'
                        f'   1. Export "{dep_type!r}" from some module\n'
                        f'   2. Add module which exports "{dep_type!r}" to "{module!r}" imports'
                    )
                    errors.append(ValidationError(err_msg))

    return errors
DIScopeMismatch

Bases: ValidationRule

Check if Singleton and Object providers don't depend on Scoped and Transient ones.

validate 
validate(context)
Source code in src/waku/ext/validation/rules.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@override
def validate(self, context: ValidationContext) -> list[ValidationError]:
    lifespan_scoped: UnionType = Singleton | Object  # pyright: ignore [reportMissingTypeArgument]
    scope_scoped: UnionType = Scoped | Transient  # pyright: ignore [reportMissingTypeArgument]

    providers: _Providers[Any] = defaultdict(list)
    for provider in self._all_providers(context.app):
        providers[provider.type_].append(provider)

    errors: list[ValidationError] = []
    for provider in chain.from_iterable(list(providers.values())):
        for dependency in provider.collect_dependencies():
            for dependency_provider in providers[dependency.inner_type]:
                if isinstance(provider, Object):
                    continue

                if (
                    isinstance(provider, lifespan_scoped)
                    and isinstance(dependency_provider, scope_scoped)
                    and not isinstance(dependency_provider, lifespan_scoped)
                ):
                    err_msg = (
                        f'Application level provider "{provider!r}" depends on request level "{dependency_provider!r}"\n'
                        f'To resolve this issue, consider either:\n'
                        f'  - Making {provider!r} {Scoped.__name__} or {Transient.__name__}\n'
                        f'  - Making {dependency_provider!r} {Singleton.__name__} or {Object.__name__}'
                    )
                    errors.append(ValidationError(err_msg))

    return errors

extensions

Extension protocols for the waku framework.

This module defines protocols for extending module behavior. These protocols allow for hooking into various lifecycle events.

ApplicationExtension module-attribute 

ApplicationExtension = (
    OnApplicationInit | AfterApplicationInit
)

ModuleExtension module-attribute 

ModuleExtension = OnModuleConfigure | OnModuleInit

OnApplicationInit

Bases: Protocol

Extension for application pre-initialization actions.

on_app_init async 

on_app_init(app)

Perform actions before application initialization.

Source code in src/waku/extensions.py 
31
32
async def on_app_init(self, app: Application) -> None:
    """Perform actions before application initialization."""

AfterApplicationInit

Bases: Protocol

Extension for application post-initialization actions.

after_app_init async 

after_app_init(app)

Perform actions after application initialization.

Source code in src/waku/extensions.py 
41
42
async def after_app_init(self, app: Application) -> None:
    """Perform actions after application initialization."""

OnModuleConfigure

Bases: Protocol

Extension for module configuration.

on_module_configure 

on_module_configure(metadata)

Perform actions before module metadata transformed to module.

Source code in src/waku/extensions.py 
51
52
53
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)

Perform actions before application initialization.

Source code in src/waku/extensions.py 
62
63
64
async def on_module_init(self, module: Module) -> None:
    """Perform actions before application initialization."""
    ...

factory

ApplicationFactory

create classmethod 

create(
    root_module,
    /,
    dependency_provider,
    lifespan=(),
    extensions=DEFAULT_EXTENSIONS,
)
Source code in src/waku/factory.py 
19
20
21
22
23
24
25
26
27
28
29
@classmethod
def create(
    cls,
    root_module: ModuleType,
    /,
    dependency_provider: DependencyProvider,
    lifespan: Sequence[LifespanFunc] = (),
    extensions: Sequence[ApplicationExtension] = DEFAULT_EXTENSIONS,
) -> Application:
    container = cls._build_container(root_module, dependency_provider)
    return Application(container, lifespan, extensions)

graph

ModuleGraph 

ModuleGraph(root_module)
Source code in src/waku/graph.py 
14
15
16
def __init__(self, root_module: Module) -> None:
    self._root_module = root_module
    self._adjacency: dict[UUID, set[Module]] = defaultdict(set)

add_node 

add_node(module)
Source code in src/waku/graph.py 
18
19
def add_node(self, module: Module) -> None:
    self._adjacency[module.id].add(module)

add_edge 

add_edge(from_module, to_module)
Source code in src/waku/graph.py 
21
22
def add_edge(self, from_module: Module, to_module: Module) -> None:
    self._adjacency[from_module.id].add(to_module)

traverse 

traverse(from_=None)
Source code in src/waku/graph.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
def traverse(self, from_: Module | None = None) -> Iterable[Module]:
    start_module = from_ or self._root_module
    visited = {start_module.id}
    queue = deque([start_module])

    while queue:
        vertex = queue.popleft()
        yield vertex

        for neighbor in self._adjacency[vertex.id]:
            if neighbor.id not in visited:
                visited.add(neighbor.id)
                queue.append(neighbor)

lifespan

LifespanFunc module-attribute 

LifespanFunc = (
    Callable[
        ['Application'], AbstractAsyncContextManager[None]
    ]
    | AbstractAsyncContextManager[None]
)

LifespanWrapper 

LifespanWrapper(lifespan_func)
Source code in src/waku/lifespan.py 
25
26
def __init__(self, lifespan_func: LifespanFunc) -> None:
    self._lifespan_func = lifespan_func

lifespan async 

lifespan(app)
Source code in src/waku/lifespan.py 
28
29
30
31
32
33
34
35
36
@contextlib.asynccontextmanager
async def lifespan(self, app: Application) -> AsyncIterator[None]:
    ctx_manager = (
        self._lifespan_func
        if isinstance(self._lifespan_func, AbstractAsyncContextManager)
        else self._lifespan_func(app)
    )
    async with ctx_manager:
        yield

mediator

Request dataclass 

Request(*, request_id=uuid4())

Bases: Generic[ResponseT]

Base class for request-type objects.

request_id class-attribute instance-attribute 

request_id = field(default_factory=uuid4)

Response dataclass 

Response()

Base class for response type objects.

Mediator 

Mediator(dependency_provider, middlewares, event_publisher)

Bases: IMediator

Default mediator implementation.

Source code in src/waku/mediator/impl.py 
21
22
23
24
25
26
27
28
29
def __init__(
    self,
    dependency_provider: DependencyProvider,
    middlewares: Sequence[AnyMiddleware],
    event_publisher: EventPublisher,
) -> None:
    self._dependency_provider = dependency_provider
    self._middleware_chain = MiddlewareChain(middlewares)
    self._event_publisher = event_publisher

send async 

send(request)

Send a request through the mediator middleware chain.

PARAMETER DESCRIPTION
request

The request to process

TYPE: Request[ResponseT]

RETURNS DESCRIPTION
ResponseT

Response from the handler
RAISES DESCRIPTION
RequestHandlerNotFound

If no handler is registered for the request type
Source code in src/waku/mediator/impl.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@override
async def send(self, request: Request[ResponseT]) -> ResponseT:
    """Send a request through the mediator middleware chain.

    Args:
        request: The request to process

    Returns:
        Response from the handler

    Raises:
        RequestHandlerNotFound: If no handler is registered for the request type
    """
    request_type = type(request)
    handler = await self._resolve_request_handler(request_type)
    return await self._handle_request(handler, request)

publish async 

publish(event)

Publish an event to all registered handlers.

PARAMETER DESCRIPTION
event

The event to publish

TYPE: Event

RAISES DESCRIPTION
EventHandlerNotFound

If no handlers are registered for the event type
Source code in src/waku/mediator/impl.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
@override
async def publish(self, event: Event) -> None:
    """Publish an event to all registered handlers.

    Args:
        event: The event to publish

    Raises:
        EventHandlerNotFound: If no handlers are registered for the event type
    """
    event_type = type(event)
    handlers = await self._resolve_event_handlers(event_type)
    await self._event_publisher(handlers, event)

IMediator

Bases: ISender, IPublisher, ABC

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

publish abstractmethod async 

publish(event)

Asynchronously send event to multiple handlers.

Source code in src/waku/mediator/interfaces.py 
24
25
26
@abc.abstractmethod
async def publish(self, event: Event) -> None:
    """Asynchronously send event to multiple handlers."""

send abstractmethod async 

send(request)

Asynchronously send a request to a single handler.

Source code in src/waku/mediator/interfaces.py 
16
17
18
@abc.abstractmethod
async def send(self, request: Request[ResponseT]) -> ResponseT:
    """Asynchronously send a request to a single handler."""

IPublisher

Bases: ABC

Publish event through the mediator to be handled by multiple handlers.

publish abstractmethod async 

publish(event)

Asynchronously send event to multiple handlers.

Source code in src/waku/mediator/interfaces.py 
24
25
26
@abc.abstractmethod
async def publish(self, event: Event) -> None:
    """Asynchronously send event to multiple handlers."""

ISender

Bases: ABC

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

send abstractmethod async 

send(request)

Asynchronously send a request to a single handler.

Source code in src/waku/mediator/interfaces.py 
16
17
18
@abc.abstractmethod
async def send(self, request: Request[ResponseT]) -> ResponseT:
    """Asynchronously send a request to a single handler."""

MiddlewareChain 

MiddlewareChain(middlewares=())
Source code in src/waku/mediator/middlewares.py 
43
44
def __init__(self, middlewares: Sequence[AnyMiddleware] = ()) -> None:
    self._middlewares: list[AnyMiddleware] = list(middlewares)

wrap 

wrap(handle)
Source code in src/waku/mediator/middlewares.py 
46
47
48
49
50
51
52
53
def wrap(self, handle: HandleType[RequestT, ResponseT]) -> HandleType[RequestT, ResponseT]:
    original_handle = handle
    for middleware in reversed(self._middlewares):
        handle = functools.update_wrapper(
            wrapper=functools.partial(middleware.__call__, handle=handle),
            wrapped=original_handle,
        )
    return handle

MediatorConfig dataclass 

MediatorConfig(
    *,
    mediator_implementation_type=Mediator,
    event_publisher=SequentialEventPublisher,
    middlewares=(),
)

Configuration for the Mediator extension.

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

ATTRIBUTE DESCRIPTION
mediator_implementation_type

The concrete implementation class for the mediator 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]

middlewares

A sequence of middleware classes that will be applied to the mediator pipeline. Middlewares are executed in the order they are defined. Defaults to an empty sequence.

TYPE: Sequence[type[AnyMiddleware]]

Example 
config = MediatorConfig(
    mediator_implementation_type=CustomMediator,
    middlewares=[LoggingMiddleware, ValidationMiddleware],
)

mediator_implementation_type class-attribute instance-attribute 

mediator_implementation_type = Mediator

event_publisher class-attribute instance-attribute 

event_publisher = SequentialEventPublisher

middlewares class-attribute instance-attribute 

middlewares = ()

MediatorExtension 

MediatorExtension()

Bases: OnModuleConfigure

Source code in src/waku/mediator/modules.py 
110
111
112
def __init__(self) -> None:
    self._request_map = RequestMap()
    self._event_map = EventMap()

bind_request 

bind_request(request_type, handler_type)
Source code in src/waku/mediator/modules.py 
114
115
116
117
118
119
120
def bind_request(
    self,
    request_type: type[RequestT],
    handler_type: RequestHandlerType[RequestT, ResponseT],
) -> Self:
    self._request_map.bind(request_type, handler_type)
    return self

bind_event 

bind_event(event_type, handler_types)
Source code in src/waku/mediator/modules.py 
122
123
124
125
126
127
128
def bind_event(
    self,
    event_type: type[EventT],
    handler_types: list[EventHandlerType[EventT]],
) -> Self:
    self._event_map.bind(event_type, handler_types)
    return self

on_module_configure 

on_module_configure(metadata)
Source code in src/waku/mediator/modules.py 
130
131
132
133
134
135
136
def on_module_configure(self, metadata: ModuleMetadata) -> None:
    metadata.providers.extend(
        chain(
            self._create_request_handler_providers(),
            self._create_event_handler_providers(),
        ),
    )

MediatorModule

register classmethod 

register(config=None)

Application-level module for Mediator setup.

PARAMETER DESCRIPTION
config

Configuration for the Mediator extension.

TYPE: MediatorConfig | None DEFAULT: None

Source code in src/waku/mediator/modules.py 
69
70
71
72
73
74
75
76
77
78
79
80
@classmethod
def register(cls, config: MediatorConfig | None = None, /) -> DynamicModule:
    """Application-level module for Mediator setup.

    Args:
        config: Configuration for the Mediator extension.
    """
    return DynamicModule(
        parent_module=cls,
        providers=list(cls._create_providers(config or MediatorConfig())),
        is_global=True,
    )

RequestHandler

Bases: ABC, Generic[RequestT, ResponseT]

The request handler interface.

The request handler is an object, which gets a request as input and may return a response as a result.

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)
Source code in src/waku/mediator/requests/handler.py 
40
41
42
@abc.abstractmethod
async def handle(self, request: RequestT) -> ResponseT:
    raise NotImplementedError

RequestMap 

RequestMap()
Source code in src/waku/mediator/requests/map.py 
19
20
def __init__(self) -> None:
    self._registry: RequestMapRegistry[Any, Any] = {}

registry property 

registry

bind 

bind(request_type, handler_type)
Source code in src/waku/mediator/requests/map.py 
22
23
24
25
26
27
def bind(self, request_type: type[RequestT], handler_type: RequestHandlerType[RequestT, ResponseT]) -> Self:
    if request_type in self._registry:
        msg = f'{request_type.__name__} already exists in registry'
        raise RequestHandlerAlreadyRegistered(msg, request_type, handler_type)
    self._registry[request_type] = handler_type
    return self

merge 

merge(other)
Source code in src/waku/mediator/requests/map.py 
29
30
31
32
def merge(self, other: RequestMap) -> Self:
    for request_type, handler_type in other.registry.items():
        self.bind(request_type, handler_type)
    return self

contracts

event

EventT module-attribute 
EventT = TypeVar(
    'EventT', bound='Event', contravariant=True
)
Event dataclass 
Event()

Base class for events.

request

RequestT module-attribute 
RequestT = TypeVar(
    'RequestT', bound='Request[Any]', contravariant=True
)
ResponseT module-attribute 
ResponseT = TypeVar(
    'ResponseT',
    bound='Response | None',
    default=None,
    covariant=True,
)
Request dataclass 
Request(*, request_id=uuid4())

Bases: Generic[ResponseT]

Base class for request-type objects.

request_id class-attribute instance-attribute 
request_id = field(default_factory=uuid4)
Response dataclass 
Response()

Base class for response type objects.

events

handler

EventHandlerType module-attribute 
EventHandlerType = type[EventHandler[EventT]]
EventHandler

Bases: ABC, Generic[EventT]

The event handler interface.

Usage::

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)
Source code in src/waku/mediator/events/handler.py 
23
24
25
@abc.abstractmethod
async def handle(self, event: EventT) -> None:
    raise NotImplementedError

map

EventMapRegistry module-attribute 
EventMapRegistry = MutableMapping[
    type[EventT], list[EventHandlerType[EventT]]
]
EventMap 
EventMap()
Source code in src/waku/mediator/events/map.py 
20
21
def __init__(self) -> None:
    self._registry: EventMapRegistry[Any] = defaultdict(list)
registry property 
registry
bind 
bind(event_type, handler_types)
Source code in src/waku/mediator/events/map.py 
23
24
25
26
27
28
29
def bind(self, event_type: type[EventT], handler_types: list[EventHandlerType[EventT]]) -> Self:
    for handler_type in handler_types:
        if handler_type in self._registry[event_type]:
            msg = f'{handler_type.__name__} already registered for {event_type.__name__} event'
            raise EventHandlerAlreadyRegistered(msg, event_type, handler_type)
        self._registry[event_type].append(handler_type)
    return self
merge 
merge(other)
Source code in src/waku/mediator/events/map.py 
31
32
33
34
def merge(self, other: EventMap) -> Self:
    for event_type, handlers in other.registry.items():
        self.bind(event_type, handlers)
    return self

publish

EventPublisher

Bases: Protocol

SequentialEventPublisher

Bases: EventPublisher

GroupEventPublisher

Bases: EventPublisher

exceptions

MediatorError

Bases: Exception

Base exception for all mediator-related errors.

ImproperlyConfiguredError

Bases: MediatorError

Raised when mediator configuration is invalid.

RequestHandlerAlreadyRegistered 

RequestHandlerAlreadyRegistered(
    msg, request_type, handler_type
)

Bases: MediatorError, KeyError

Raised when a request handler is already registered.

ATTRIBUTE DESCRIPTION
request_type

The type of request that caused the error.

TYPE: type[Request[Any]]

handler_type

The type of handler that was already registered.

TYPE: RequestHandlerType[Any, Any]

Source code in src/waku/mediator/exceptions.py 
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    msg: str,
    request_type: type[Request[Any]],
    handler_type: RequestHandlerType[Any, Any],
) -> None:
    super().__init__(msg)
    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(msg, request_type)

Bases: MediatorError, TypeError

Raised when a request handler is not found.

ATTRIBUTE DESCRIPTION
request_type

The type of request that caused the error.

TYPE: type[Request[Any]]

Source code in src/waku/mediator/exceptions.py 
60
61
62
def __init__(self, msg: str, request_type: type[Request[Any]]) -> None:
    super().__init__(msg)
    self.request_type = request_type
request_type instance-attribute 
request_type = request_type

EventHandlerAlreadyRegistered 

EventHandlerAlreadyRegistered(
    msg, event_type, handler_type
)

Bases: MediatorError, KeyError

Raised when an event handler is already registered.

ATTRIBUTE DESCRIPTION
event_type

The type of event that caused the error.

TYPE: type[Event]

handler_type

The type of handler that was already registered.

TYPE: EventHandlerType[Any]

Source code in src/waku/mediator/exceptions.py 
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    msg: str,
    event_type: type[Event],
    handler_type: EventHandlerType[Any],
) -> None:
    super().__init__(msg)
    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

EventHandlerNotFound 

EventHandlerNotFound(msg, event_type)

Bases: MediatorError, TypeError

Raised when an event handler is not found.

ATTRIBUTE DESCRIPTION
event_type

The type of event that caused the error.

TYPE: type[Event]

Source code in src/waku/mediator/exceptions.py 
96
97
98
def __init__(self, msg: str, event_type: type[Event]) -> None:
    super().__init__(msg)
    self.event_type = event_type
event_type instance-attribute 
event_type = event_type

impl

Mediator 

Mediator(dependency_provider, middlewares, event_publisher)

Bases: IMediator

Default mediator implementation.

Source code in src/waku/mediator/impl.py 
21
22
23
24
25
26
27
28
29
def __init__(
    self,
    dependency_provider: DependencyProvider,
    middlewares: Sequence[AnyMiddleware],
    event_publisher: EventPublisher,
) -> None:
    self._dependency_provider = dependency_provider
    self._middleware_chain = MiddlewareChain(middlewares)
    self._event_publisher = event_publisher
send async 
send(request)

Send a request through the mediator middleware chain.

PARAMETER DESCRIPTION
request

The request to process

TYPE: Request[ResponseT]

RETURNS DESCRIPTION
ResponseT

Response from the handler
RAISES DESCRIPTION
RequestHandlerNotFound

If no handler is registered for the request type
Source code in src/waku/mediator/impl.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@override
async def send(self, request: Request[ResponseT]) -> ResponseT:
    """Send a request through the mediator middleware chain.

    Args:
        request: The request to process

    Returns:
        Response from the handler

    Raises:
        RequestHandlerNotFound: If no handler is registered for the request type
    """
    request_type = type(request)
    handler = await self._resolve_request_handler(request_type)
    return await self._handle_request(handler, request)
publish async 
publish(event)

Publish an event to all registered handlers.

PARAMETER DESCRIPTION
event

The event to publish

TYPE: Event

RAISES DESCRIPTION
EventHandlerNotFound

If no handlers are registered for the event type
Source code in src/waku/mediator/impl.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
@override
async def publish(self, event: Event) -> None:
    """Publish an event to all registered handlers.

    Args:
        event: The event to publish

    Raises:
        EventHandlerNotFound: If no handlers are registered for the event type
    """
    event_type = type(event)
    handlers = await self._resolve_event_handlers(event_type)
    await self._event_publisher(handlers, event)

interfaces

ISender

Bases: ABC

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

send abstractmethod async 
send(request)

Asynchronously send a request to a single handler.

Source code in src/waku/mediator/interfaces.py 
16
17
18
@abc.abstractmethod
async def send(self, request: Request[ResponseT]) -> ResponseT:
    """Asynchronously send a request to a single handler."""

IPublisher

Bases: ABC

Publish event through the mediator to be handled by multiple handlers.

publish abstractmethod async 
publish(event)

Asynchronously send event to multiple handlers.

Source code in src/waku/mediator/interfaces.py 
24
25
26
@abc.abstractmethod
async def publish(self, event: Event) -> None:
    """Asynchronously send event to multiple handlers."""

IMediator

Bases: ISender, IPublisher, ABC

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

publish abstractmethod async 
publish(event)

Asynchronously send event to multiple handlers.

Source code in src/waku/mediator/interfaces.py 
24
25
26
@abc.abstractmethod
async def publish(self, event: Event) -> None:
    """Asynchronously send event to multiple handlers."""
send abstractmethod async 
send(request)

Asynchronously send a request to a single handler.

Source code in src/waku/mediator/interfaces.py 
16
17
18
@abc.abstractmethod
async def send(self, request: Request[ResponseT]) -> ResponseT:
    """Asynchronously send a request to a single handler."""

middlewares

HandleType module-attribute 

HandleType = Callable[[RequestT], Awaitable[ResponseT]]

AnyMiddleware module-attribute 

AnyMiddleware = Middleware[Any, Any]

Middleware

Bases: ABC, Generic[RequestT, ResponseT]

NoopMiddleware

Bases: Middleware[RequestT, ResponseT]

MiddlewareChain 

MiddlewareChain(middlewares=())
Source code in src/waku/mediator/middlewares.py 
43
44
def __init__(self, middlewares: Sequence[AnyMiddleware] = ()) -> None:
    self._middlewares: list[AnyMiddleware] = list(middlewares)
wrap 
wrap(handle)
Source code in src/waku/mediator/middlewares.py 
46
47
48
49
50
51
52
53
def wrap(self, handle: HandleType[RequestT, ResponseT]) -> HandleType[RequestT, ResponseT]:
    original_handle = handle
    for middleware in reversed(self._middlewares):
        handle = functools.update_wrapper(
            wrapper=functools.partial(middleware.__call__, handle=handle),
            wrapped=original_handle,
        )
    return handle

modules

MediatorConfig dataclass 

MediatorConfig(
    *,
    mediator_implementation_type=Mediator,
    event_publisher=SequentialEventPublisher,
    middlewares=(),
)

Configuration for the Mediator extension.

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

ATTRIBUTE DESCRIPTION
mediator_implementation_type

The concrete implementation class for the mediator 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]

middlewares

A sequence of middleware classes that will be applied to the mediator pipeline. Middlewares are executed in the order they are defined. Defaults to an empty sequence.

TYPE: Sequence[type[AnyMiddleware]]

Example 
config = MediatorConfig(
    mediator_implementation_type=CustomMediator,
    middlewares=[LoggingMiddleware, ValidationMiddleware],
)
mediator_implementation_type class-attribute instance-attribute 
mediator_implementation_type = Mediator
event_publisher class-attribute instance-attribute 
event_publisher = SequentialEventPublisher
middlewares class-attribute instance-attribute 
middlewares = ()

MediatorModule

register classmethod 
register(config=None)

Application-level module for Mediator setup.

PARAMETER DESCRIPTION
config

Configuration for the Mediator extension.

TYPE: MediatorConfig | None DEFAULT: None

Source code in src/waku/mediator/modules.py 
69
70
71
72
73
74
75
76
77
78
79
80
@classmethod
def register(cls, config: MediatorConfig | None = None, /) -> DynamicModule:
    """Application-level module for Mediator setup.

    Args:
        config: Configuration for the Mediator extension.
    """
    return DynamicModule(
        parent_module=cls,
        providers=list(cls._create_providers(config or MediatorConfig())),
        is_global=True,
    )

MediatorExtension 

MediatorExtension()

Bases: OnModuleConfigure

Source code in src/waku/mediator/modules.py 
110
111
112
def __init__(self) -> None:
    self._request_map = RequestMap()
    self._event_map = EventMap()
bind_request 
bind_request(request_type, handler_type)
Source code in src/waku/mediator/modules.py 
114
115
116
117
118
119
120
def bind_request(
    self,
    request_type: type[RequestT],
    handler_type: RequestHandlerType[RequestT, ResponseT],
) -> Self:
    self._request_map.bind(request_type, handler_type)
    return self
bind_event 
bind_event(event_type, handler_types)
Source code in src/waku/mediator/modules.py 
122
123
124
125
126
127
128
def bind_event(
    self,
    event_type: type[EventT],
    handler_types: list[EventHandlerType[EventT]],
) -> Self:
    self._event_map.bind(event_type, handler_types)
    return self
on_module_configure 
on_module_configure(metadata)
Source code in src/waku/mediator/modules.py 
130
131
132
133
134
135
136
def on_module_configure(self, metadata: ModuleMetadata) -> None:
    metadata.providers.extend(
        chain(
            self._create_request_handler_providers(),
            self._create_event_handler_providers(),
        ),
    )

requests

handler

RequestHandlerType module-attribute 
RequestHandlerType = type[
    RequestHandler[RequestT, ResponseT]
]
RequestHandler

Bases: ABC, Generic[RequestT, ResponseT]

The request handler interface.

The request handler is an object, which gets a request as input and may return a response as a result.

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)
Source code in src/waku/mediator/requests/handler.py 
40
41
42
@abc.abstractmethod
async def handle(self, request: RequestT) -> ResponseT:
    raise NotImplementedError

map

RequestMapRegistry module-attribute 
RequestMapRegistry = MutableMapping[
    type[RequestT], RequestHandlerType[RequestT, ResponseT]
]
RequestMap 
RequestMap()
Source code in src/waku/mediator/requests/map.py 
19
20
def __init__(self) -> None:
    self._registry: RequestMapRegistry[Any, Any] = {}
registry property 
registry
bind 
bind(request_type, handler_type)
Source code in src/waku/mediator/requests/map.py 
22
23
24
25
26
27
def bind(self, request_type: type[RequestT], handler_type: RequestHandlerType[RequestT, ResponseT]) -> Self:
    if request_type in self._registry:
        msg = f'{request_type.__name__} already exists in registry'
        raise RequestHandlerAlreadyRegistered(msg, request_type, handler_type)
    self._registry[request_type] = handler_type
    return self
merge 
merge(other)
Source code in src/waku/mediator/requests/map.py 
29
30
31
32
def merge(self, other: RequestMap) -> Self:
    for request_type, handler_type in other.registry.items():
        self.bind(request_type, handler_type)
    return self

modules

ModuleType module-attribute 

ModuleType = 'type[object | HasModuleMetadata]'

DynamicModule dataclass 

DynamicModule(
    *,
    providers=list(),
    imports=list(),
    exports=list(),
    extensions=list(),
    is_global=False,
    id=uuid4(),
    parent_module,
)

Bases: ModuleMetadata

providers class-attribute instance-attribute 

providers = field(default_factory=list)

List of providers for dependency injection.

imports class-attribute instance-attribute 

imports = field(default_factory=list)

List of modules imported by this module.

exports class-attribute instance-attribute 

exports = field(default_factory=list)

List of types or modules exported by this module.

extensions class-attribute instance-attribute 

extensions = field(default_factory=list)

List of module extensions for lifecycle hooks.

is_global class-attribute instance-attribute 

is_global = False

Whether this module is global or not.

id class-attribute instance-attribute 

id = field(default_factory=uuid4)

parent_module instance-attribute 

parent_module

ModuleCompiler

extract_metadata 

extract_metadata(module_type)
Source code in src/waku/modules/_metadata.py 
94
95
96
97
98
99
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(),
    imports=list(),
    exports=list(),
    extensions=list(),
    is_global=False,
    id=uuid4(),
)

providers class-attribute instance-attribute 

providers = field(default_factory=list)

List of providers for dependency injection.

imports class-attribute instance-attribute 

imports = field(default_factory=list)

List of modules imported by this module.

exports class-attribute instance-attribute 

exports = field(default_factory=list)

List of types or modules exported by this module.

extensions class-attribute instance-attribute 

extensions = field(default_factory=list)

List of module extensions for lifecycle hooks.

is_global class-attribute instance-attribute 

is_global = False

Whether this module is global or not.

id class-attribute instance-attribute 

id = field(default_factory=uuid4)

Module 

Module(module_type, metadata)
Source code in src/waku/modules/_module.py 
16
17
18
19
20
21
22
23
24
25
def __init__(self, module_type: ModuleType, metadata: ModuleMetadata) -> None:
    self.id: Final[UUID] = metadata.id
    self.target: Final[ModuleType] = module_type
    self.distance: Final[int] = sys.maxsize if metadata.is_global else 0

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

id instance-attribute 

id = id

target instance-attribute 

target = module_type

distance instance-attribute 

distance = maxsize if is_global else 0

providers instance-attribute 

providers = providers

imports instance-attribute 

imports = imports

exports instance-attribute 

exports = exports

extensions instance-attribute 

extensions = extensions

is_global instance-attribute 

is_global = is_global

name property 

name

module 

module(
    *,
    providers=(),
    imports=(),
    exports=(),
    extensions=(),
    is_global=False,
)

Decorator to define a module.

PARAMETER DESCRIPTION
providers

Sequence of providers for dependency injection.

TYPE: Sequence[Provider[Any]] 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[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 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def module(
    *,
    providers: Sequence[Provider[Any]] = (),
    imports: Sequence[ModuleType | DynamicModule] = (),
    exports: Sequence[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,
        )
        setattr(cls, MODULE_METADATA_KEY, metadata)
        return cls

    return decorator