^{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

• Getting Started
• Module System
• Providers
• Extensions
• CQRS
• Event Sourcing
• API Reference
• dishka Documentation
• DeepWiki

Contributing

• Contributing Guide
• Development Setup

Top contributors

Roadmap

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

Support

• RU Telegram group
• GitHub Issues
• Discussions

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.