Overview

枠 · framework for Python backends that grow

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. Python has no built-in way to enforce component boundaries — so what starts as clean code quietly becomes a tangle of implicit dependencies that discipline alone can't prevent.

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.

Installation¶

uvpip

uv add waku

pip install waku

Structure that scales¶

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.

Built-in capabilities¶

CQRS & Message Bus

DI alone doesn't decouple components — you need events. The message bus dispatches commands, queries, and events so components never reference each other directly. Pipeline behaviors handle cross-cutting concerns.
Event Sourcing

Full event sourcing support — aggregates, projections, snapshots, upcasting, and the decider pattern with built-in SQLAlchemy adapters.
Lifecycle & Extensions

Hook into startup, shutdown, and module initialization with extensions. Add validation, logging, or custom behaviors — decoupled from your business logic.

How it works¶

Group related providers into modules with explicit imports and exports. WakuFactory wires the module tree into a DI container. Plug it into your framework — FastAPI, Litestar, or anything else — and you're done.

Recommended project structure

Following Explicit Architecture and its code reflection, only UI and shared infrastructure live at the top level — each feature component is a vertical slice with its own domain, application, and infrastructure layers, wired together by a waku module:

your_app/
├── core/
│   ├── components/
│   │   ├── users/              # feature component
│   │   │   ├── domain/         # entities, value objects, events
│   │   │   ├── application/    # use cases, handlers, ports
│   │   │   ├── infra/          # repositories, adapters
│   │   │   └── module.py       # waku module
│   │   └── orders/
│   │       ├── domain/
│   │       ├── application/
│   │       ├── infra/
│   │       └── module.py
│   ├── ports/                  # shared system ports
│   └── shared_kernel/          # cross-component contracts
├── infra/                      # cross-cutting infrastructure
│   └── module.py
├── ui/                         # API routes, CLI handlers
└── app.py                      # composition root

Quick example¶

BasicWith FastAPIWith module boundaries

A service, a module, and a container — the minimal waku app:

app.py
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())

Wire waku into FastAPI with dishka's integration — same modules, real HTTP:

main.py
import contextlib
from collections.abc import AsyncIterator

import uvicorn
from dishka.integrations.fastapi import inject, setup_dishka
from fastapi import FastAPI

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


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


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


@contextlib.asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    async with app.state.waku:
        yield


app = FastAPI(lifespan=lifespan)
waku_app = WakuFactory(AppModule).create()
app.state.waku = waku_app
setup_dishka(waku_app.container, app)


@app.get('/')
@inject
async def hello(greeting: Injected[GreetingService]) -> dict[str, str]:
    return {'message': await greeting.greet('waku')}


if __name__ == '__main__':
    uvicorn.run(app)

Modules control visibility. InfrastructureModule exports IUserRepository — UserModule imports it. Swap the storage layer by changing one provider.

app.py
import asyncio
from typing import Protocol

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


class IUserRepository(Protocol):
    async def get(self, user_id: str) -> str | None: ...
    async def save(self, user_id: str, name: str) -> None: ...


class InMemoryUserRepository(IUserRepository):
    def __init__(self) -> None:
        self._users: dict[str, str] = {}

    async def get(self, user_id: str) -> str | None:
        return self._users.get(user_id)

    async def save(self, user_id: str, name: str) -> None:
        self._users[user_id] = name


class UserService:
    def __init__(self, repo: IUserRepository) -> None:
        self._repo = repo

    async def create_user(self, username: str) -> str:
        user_id = f'user_{username}'
        await self._repo.save(user_id, username)
        return user_id


@module(
    providers=[singleton(IUserRepository, InMemoryUserRepository)],  # (1)!
    exports=[IUserRepository],  # (2)!
)
class InfrastructureModule:
    pass


@module(
    imports=[InfrastructureModule],  # (3)!
    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())

singleton(IUserRepository, InMemoryUserRepository) — binds the interface to an implementation. Swap to a database-backed repository by changing this one provider.
Only the interface is exported — other modules depend on IUserRepository, never the concrete class.
UserModule imports InfrastructureModule to access the exported IUserRepository.

Next steps¶

Getting Started

Install waku, build a modular app, and connect it to your framework
Examples

Working projects showing real usage patterns with FastAPI, Litestar, and more
API Reference

Full module, class, and function reference