waku-py/waku

^{waku [枠 or わく] means framework in Japanese.}

Python makes it easy to build a backend. waku makes it easy to keep growing one.

As your project scales, problems creep in: services import each other freely,
swapping a database means editing dozens of files, and nobody can tell which module
depends on what. waku gives you modules with explicit boundaries, type-safe DI
powered by dishka, and integrated CQRS
and event sourcing — so your codebase stays manageable as it scales.

Tip

Check out the full documentation and our examples to get started.

The Problem

Python has no built-in way to enforce component boundaries. Packages don't control visibility, imports aren't validated, and nothing stops module A from reaching into the internals of module B. As a project grows, what started as clean separation quietly becomes a web of implicit dependencies — where testing requires the whole system, onboarding means reading everything, and changing one module risks breaking three others.

What waku gives you

Structure

🧩 Package by component:
Each module is a self-contained unit with its own providers.
Explicit imports and exports control what crosses boundaries —
validated at startup, not discovered in production.
💉 Dependency inversion:
Define interfaces in your application core, bind adapters in infrastructure modules.
Swap a database, a cache, or an API client by changing one provider —
powered by dishka.
🔌 One core, any entrypoint:
Build your module tree once with WakuFactory.
Plug it into FastAPI, Litestar, FastStream, Aiogram, CLI, or workers —
same logic everywhere.

Capabilities

📨 CQRS & mediator:
DI alone doesn't decouple components — you need events.
The mediator dispatches commands, queries, and events so components
never reference each other directly. Pipeline behaviors handle cross-cutting concerns.
📜 Event sourcing:
Aggregates, projections, snapshots, upcasting, and the decider pattern
with built-in SQLAlchemy adapters.
🧪 Testing:
Override any provider in tests with override(),
or spin up a minimal app with create_test_app().
🧰 Lifecycle & extensions:
Hook into startup, shutdown, and module initialization.
Add validation, logging, or custom behaviors —
decoupled from your business logic.

Quick Start

Installation

uv add waku

Minimal Example

Define a service, register it in a module, and resolve it from the container:

import asyncio

from waku import WakuFactory, module
from waku.di import scoped


class GreetingService:
    async def greet(self, name: str) -> str:
        return f'Hello, {name}!'


@module(providers=[scoped(GreetingService)])
class GreetingModule:
    pass


@module(imports=[GreetingModule])
class AppModule:
    pass


async def main() -> None:
    app = WakuFactory(AppModule).create()

    async with app, app.container() as c:
        svc = await c.get(GreetingService)
        print(await svc.greet('waku'))


if __name__ == '__main__':
    asyncio.run(main())

Module Boundaries in Action

Modules control visibility. InfrastructureModule exports ILogger — UserModule imports it. Dependencies are explicit, not implicit:

import asyncio
from typing import Protocol

from waku import WakuFactory, module
from waku.di import scoped, singleton


class ILogger(Protocol):
    async def log(self, message: str) -> None: ...


class ConsoleLogger(ILogger):
    async def log(self, message: str) -> None:
        print(f'[LOG] {message}')


class UserService:
    def __init__(self, logger: ILogger) -> None:
        self.logger = logger

    async def create_user(self, username: str) -> str:
        user_id = f'user_{username}'
        await self.logger.log(f'Created user: {username}')
        return user_id


@module(
    providers=[singleton(ILogger, ConsoleLogger)],
    exports=[ILogger],
)
class InfrastructureModule:
    pass


@module(
    imports=[InfrastructureModule],
    providers=[scoped(UserService)],
)
class UserModule:
    pass


@module(imports=[UserModule])
class AppModule:
    pass


async def main() -> None:
    app = WakuFactory(AppModule).create()

    async with app, app.container() as c:
        user_service = await c.get(UserService)
        user_id = await user_service.create_user('alice')
        print(f'Created user with ID: {user_id}')


if __name__ == '__main__':
    asyncio.run(main())

Next steps

Learn about module imports and exports
Try different provider scopes
Add CQRS for clean command handling
Connect with your favorite framework
Browse the examples directory for inspiration

Documentation

Contributing

Top contributors

Roadmap

Create logo
Improve inner architecture
Improve documentation
Add new and improve existing validation rules
Provide example projects for common architectures

Support

License

This project is licensed under the terms of the MIT License.

Acknowledgements

dishka – Dependency Injection framework powering waku IoC container.
NestJS – Inspiration for modular architecture and design patterns.
MediatR (C#) – Inspiration for the CQRS subsystem.
Emmett – Functional-first event sourcing patterns.
Marten – Projection lifecycle taxonomy.
Eventuous – Event store interface design.
Jérémie Chassaing – Decider pattern formalization.