### Start Live Preview Server

Source: https://github.com/waku-py/waku/blob/master/docs/contributing/docs.md

This command starts a live preview server for the Waku-Py documentation. It allows developers to see their changes in real-time as they edit files in the 'docs/' directory. Ensure you have followed the development setup instructions before running this command.

```bash
zensical serve
```

--------------------------------

### Minimal waku Application Setup in Python

Source: https://github.com/waku-py/waku/blob/master/README.md

Demonstrates the basic setup of a waku application. It defines a service, registers it within a module, and then resolves and uses the service from the application's container. This example highlights dependency injection and module registration.

```python
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())

```

--------------------------------

### Run Waku-Py Application

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Command to execute the Waku-Py application from the command line.

```shell
python -m app
```

--------------------------------

### Project Structure for Realistic Application

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Defines a comprehensive project structure for a realistic Waku application, organizing modules for greetings and users, along with settings.

```text
app/
├── __init__.py
├── __main__.py
├── application.py
├── modules/
│   ├── __init__.py
│   ├── greetings/
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── services.py
│   │   └── module.py
│   └── users/
│       ├── __init__.py
│       ├── models.py
│       ├── services.py
│       └── module.py
└── settings.py

```

--------------------------------

### Create User Module Components

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Sets up the user module for a Waku application, defining models, repository interfaces and implementations, and services. It emphasizes dependency injection for storage backends.

```python
--8<-- "docs/code/getting_started/users/models.py"

```

```python
Define a repository interface and an in-memory implementation.
The interface lets you swap storage backends without touching service code:

```

```python
--8<-- "docs/code/getting_started/users/repositories.py"

```

```python
The service depends on the `IUserRepository` interface — it doesn't know
which implementation is behind it:

```

```python
--8<-- "docs/code/getting_started/users/services.py"

```

```python
The module wires the interface to the implementation. Swap `InMemoryUserRepository`
for a database-backed one by changing a single provider:

```

```python
--8<-- "docs/code/getting_started/users/module.py"

```

--------------------------------

### Basic Waku App Example (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/index.md

Demonstrates the minimal waku application structure with a service, module, and container. It shows how to create a Waku app, get a service from the container, and use it.

```python
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())

```

--------------------------------

### User Service With Waku (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Illustrates a Waku-based service where dependencies such as AsyncSession and INotifier are injected via the constructor. This promotes better testability and modularity by decoupling the service from concrete implementations.

```python
class UserService:
    def __init__(self, session: AsyncSession, notifier: INotifier) -> None:
        self.session = session
        self.notifier = notifier

    async def create_user(self, name: str) -> User:
        user = User(name=name)
        self.session.add(user)
        await self.notifier.notify(user.email, '...')
        return user
```

--------------------------------

### Basic Greeting Service (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Defines a simple asynchronous service named GreetingService with a greet method. This service takes a name as input and returns a personalized greeting string. It serves as a basic example for defining services within a Waku application.

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

```

--------------------------------

### Bootstrap Standalone Waku Application

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Defines modules and bootstraps a standalone Waku application using GreetingService. It demonstrates dependency injection and application creation for simple scripts and demos.

```python
import asyncio

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

from services import GreetingService


@module(
    providers=[scoped(GreetingService)],  # (1)!
    exports=[GreetingService],  # (2)!
)
class GreetingModule:
    pass


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


def bootstrap() -> WakuApplication:  # (3)!
    return WakuFactory(AppModule).create()


async def main() -> None:
    application = bootstrap()

    async with application, application.container() as container:  # (4)!
        svc = await container.get(GreetingService)
        print(await svc.greet('waku'))


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

```

--------------------------------

### Integrate Waku with FastAPI

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Illustrates integrating Waku with FastAPI, managing Waku lifecycle through FastAPI's lifespan and enabling automatic dependency resolution per request using dishka.

```python
--8<-- "docs/code/code/integrations/fastapi_example.py"

```

--------------------------------

### Manual Dependency Injection Example

Source: https://github.com/waku-py/waku/blob/master/docs/fundamentals/providers.md

Demonstrates manual dependency injection using a MockClient injected into a Service. This setup facilitates isolated testing of the Service without depending on a real client implementation.

```python
class MockClient:
    def __init__(self) -> None:
        pass

    def get_data(self) -> str:
        return "mock data"


class Service:
    def __init__(self, client: MockClient) -> None:
        self._client = client

    def process_data(self) -> str:
        return self._client.get_data().upper()


client = MockClient()
service = Service(client)

print(service.process_data())
```

--------------------------------

### Create Greeting Module Components

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Defines the models, services, and module for the greeting feature in a Waku application. This includes data models, greeting service logic, and module configuration.

```python
--8<-- "docs/code/getting_started/greetings/models.py"

```

```python
--8<-- "docs/code/getting_started/greetings/services.py"

```

```python
--8<-- "docs/code/getting_started/greetings/module.py"

```

--------------------------------

### Define Application Settings Module

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Sets up a configuration module for the Waku application using settings.py. It highlights the use of `pydantic-settings` and the `is_global` flag for module exports.

```python
--8<-- "docs/code/getting_started/settings.py"

```

--------------------------------

### User Service Without Waku (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Demonstrates a typical Python service with hardcoded imports for dependencies like database sessions, configuration settings, and email notifications. This approach makes testing difficult due to direct dependencies on external modules.

```python
from db import get_session
from config import settings
from notifications import send_email


class UserService:
    def create_user(self, name: str) -> User:
        session = get_session()              # (1)!
        user = User(name=name)
        session.add(user)
        session.commit()
        if settings.SEND_WELCOME_EMAIL:      # (2)!
            send_email(user.email, '...')     # (3)!
        return user
```

--------------------------------

### Set Up PostgreSQL Event Store with SQLAlchemy

Source: https://github.com/waku-py/waku/blob/master/docs/features/eventsourcing/event-store.md

This section details the setup for using PostgreSQL with SQLAlchemy for event sourcing in Waku. It includes installing necessary dependencies and provides a Python code snippet for setting up the event store tables and engine.

```python
--8<-- "docs/code/eventsourcing/postgres_setup.py"
```

--------------------------------

### Live Projection Examples (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/features/eventsourcing/projections.md

Illustrates how to implement live projections in Waku using Python. These examples show projecting events from a single stream and from the global log based on specific event types.

```python
# Single-stream: project all events from one aggregate
events = await event_reader.read_stream(stream_id)
await projection.project(events)

# Cross-stream: project specific event types from the global log
events = await event_reader.read_all(event_types=['OrderPlaced', 'PaymentReceived'])
await projection.project(events)
```

--------------------------------

### Create Main Entrypoint for Application in Python

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Creates the main entrypoint for the Waku-Py application. In production, this would typically be integrated with frameworks like FastAPI or Litestar, which handle container scoping per request.

```python
import asyncio
from app.application import bootstrap

async def main():
    app = await bootstrap()
    async with app, app.container():
        print("Hello, Alice!")
        print("Bonjour, Bob!")
        print("¡Hola, Carlos!")

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

--------------------------------

### Collector Registration with `collect=True` (Default)

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/di/multi-bindings.md

This example demonstrates the default behavior of `many()` where `collect` is `True`. It shows how a collector is automatically created to aggregate implementations, resolving to an empty list if no implementations are provided.

```python
from waku.di import many

# When collect=True (default), passing no implementations is valid
# It creates a collector that resolves to an empty list.
empty_collector_provider = many(SomeInterface)

```

--------------------------------

### Complete CQRS Example with Handlers

Source: https://github.com/waku-py/waku/blob/master/docs/features/cqrs/index.md

Provides a full example of an order placement flow, including domain models (commands, events, responses) and their respective handlers. This demonstrates the end-to-end integration of CQRS with the mediator.

```python
from dataclasses import dataclass

from typing_extensions import override

from waku import WakuFactory, module
from waku.cqrs import (
    Event,
    EventHandler,
    IMediator,
    MediatorConfig,
    MediatorExtension,
    MediatorModule,
    Request,
    RequestHandler,
    Response,
)


# --- Domain ---

@dataclass(frozen=True, kw_only=True)
class OrderConfirmation(Response):
    order_id: str
    status: str


@dataclass(frozen=True, kw_only=True)
class PlaceOrder(Request[OrderConfirmation]):
    customer_id: str
    product_id: str


@dataclass(frozen=True, kw_only=True)
class OrderPlaced(Event):
    order_id: str
    customer_id: str


```

--------------------------------

### Install Waku with Event Sourcing and PostgreSQL (Bash)

Source: https://github.com/waku-py/waku/blob/master/docs/features/eventsourcing/index.md

Command to install Waku with event sourcing and the SQLAlchemy adapter for PostgreSQL persistence. This allows Waku to use PostgreSQL as the event store.

```bash
uv add waku --extra eventsourcing --extra eventsourcing-sqla
```

--------------------------------

### Install waku Package

Source: https://github.com/waku-py/waku/blob/master/README.md

This command installs the waku package using the uv package manager. Ensure uv is installed and configured in your environment before running this command.

```sh
uv add waku
```

--------------------------------

### Waku CQRS Mediator Extension Configuration Example

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/extensions/custom-extensions.md

Shows an example of configuring the `MediatorExtension` for CQRS patterns using a fluent builder. It demonstrates binding commands and events to their respective handlers.

```python
from waku import module
from waku.cqrs import MediatorExtension

from .handlers import CreateOrderHandler, OrderCreatedHandler
from .contracts import CreateOrderCommand, OrderCreatedEvent

mediator_ext = (
    MediatorExtension()
    .bind_request(CreateOrderCommand, CreateOrderHandler)
    .bind_event(OrderCreatedEvent, [OrderCreatedHandler])
)

@module(extensions=[mediator_ext])
class OrderModule:
    pass
```

--------------------------------

### Install Waku with Event Sourcing Extra (Bash)

Source: https://github.com/waku-py/waku/blob/master/docs/features/eventsourcing/index.md

Command to install the Waku library with the event sourcing capabilities enabled. This command uses 'uv' as the package installer.

```bash
uv add waku --extra eventsourcing
```

--------------------------------

### Complete Service-Repository DI Pattern (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/fundamentals/providers.md

Combines individual provider patterns into a complete service-and-repository setup, a common DI pattern. This example shows binding an interface to an implementation and injecting the repository into the service via constructor type hints.

```python
--8<-- "docs/code/providers/service_repository.py"
```

--------------------------------

### Install Waku using pip

Source: https://github.com/waku-py/waku/blob/master/docs/index.md

This snippet demonstrates how to install the 'waku' package using the standard Python package installer, 'pip'. This is a common method for adding Python libraries to a project.

```shell
pip install waku

```

--------------------------------

### Install Waku using uv

Source: https://github.com/waku-py/waku/blob/master/docs/index.md

This snippet shows how to add the 'waku' package to your project using the 'uv' package manager. 'uv' is a fast Python package installer and dependency manager.

```shell
uv add waku

```

--------------------------------

### Waku Module Boundaries Example (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/index.md

Illustrates how to control visibility and dependencies between modules in waku. This example uses an interface (`IUserRepository`) exported from `InfrastructureModule` and imported by `UserModule`, allowing for easy swapping of storage implementations.

```python
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())

```

--------------------------------

### CatchUpProjectionRunner Configuration Example (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/features/eventsourcing/projections.md

This Python code snippet is a placeholder for a configuration file, likely 'docs/code/eventsourcing/projections/runner.py'. It would contain the actual implementation details for the CatchUpProjectionRunner, which polls the event store and dispatches events.

```python
--8<-- "docs/code/eventsourcing/projections/runner.py"
```

--------------------------------

### Combine OnApplicationInit and OnApplicationShutdown Hooks in Waku Extension

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/extensions/custom-extensions.md

Shows how a single extension class can implement both initialization and shutdown logic. This is useful for managing resources that need setup and cleanup, like background tasks.

```python
import asyncio
import contextlib
import logging

from waku.application import WakuApplication
from waku.extensions import OnApplicationInit, OnApplicationShutdown

logger = logging.getLogger(__name__)


class PeriodicHealthReport(OnApplicationInit, OnApplicationShutdown):
    def __init__(self, interval: float = 60.0) -> None:
        self._interval = interval
        self._task: asyncio.Task[None] | None = None

    async def on_app_init(self, app: WakuApplication) -> None:
        self._task = asyncio.create_task(self._report_loop())
        logger.info('Health reporting started (every %.0fs)', self._interval)

    async def on_app_shutdown(self, app: WakuApplication) -> None:
        if self._task is not None:
            self._task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._task
        logger.info('Health reporting stopped')

    async def _report_loop(self) -> None:
        while True:
            await asyncio.sleep(self._interval)
            logger.info('Application healthy')
```

--------------------------------

### Implement OnApplicationInit Extension in Python

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/extensions/custom-extensions.md

This Python code defines an `OnApplicationInit` extension called `StartupBanner`. It implements the `on_app_init` asynchronous method, which is called after all `OnModuleInit` hooks have completed during application initialization.

```python
from waku.application import WakuApplication
from waku.extensions import OnApplicationInit


class StartupBanner(OnApplicationInit):
    async def on_app_init(self, app: WakuApplication) -> None:
        print(f'Application started with {len(app.registry.modules)} modules')
```

--------------------------------

### Collector Creation with `many()`

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/di/multi-bindings.md

This example illustrates the default behavior of `many()` when `collect=True`. It registers individual implementations and then creates a collector that aggregates them into a `Sequence` and an alias for `list`.

```python
from waku.di import many

# Assuming INotificationChannel is defined elsewhere
# and EmailChannel, SmsChannel are implementations

# This implicitly registers EmailChannel and SmsChannel
# and creates a collector for Sequence[INotificationChannel]
# and an alias for list[INotificationChannel]
provider = many(INotificationChannel, EmailChannel, SmsChannel)

```

--------------------------------

### Waku Project Structure Example (Text)

Source: https://github.com/waku-py/waku/blob/master/docs/index.md

Illustrates a recommended directory structure for a Waku application based on Explicit Architecture. It shows how to organize feature components (e.g., users, orders) into vertical slices, each containing domain, application, and infrastructure layers, wired together by a Waku module.

```text
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
```

--------------------------------

### Create WakuApplication with WakuFactory

Source: https://github.com/waku-py/waku/blob/master/docs/fundamentals/application.md

Demonstrates the basic creation of a WakuApplication using WakuFactory and a root module. This involves defining a simple module and then instantiating the factory to create the application instance.

```python
from waku import WakuFactory, module


@module()
class AppModule:
    pass


app = WakuFactory(AppModule).create()
```

--------------------------------

### Manage Waku Application Lifecycle with Async Context Manager (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/fundamentals/application.md

Demonstrates how to use `async with` to manage the complete startup and shutdown sequence of a Waku application. This includes initializing extensions, entering lifespan functions, and entering the DI container context.

```python
from waku import WakuFactory, module

class MyService:
    async def run(self):
        print("MyService is running")

class AppModule:
    pass

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

    async with app, app.container() as container:
        svc = await container.get(MyService)
        await svc.run()
```

--------------------------------

### FastAPI Integration

Source: https://context7.com/waku-py/waku/llms.txt

Demonstrates how to integrate waku with FastAPI using dishka for automatic dependency injection in route handlers. This section includes setup, service definitions, and example endpoints.

```APIDOC
## FastAPI Integration

This section details the integration of waku with FastAPI, leveraging dishka for dependency injection.

### Description
Integrates waku with FastAPI using dishka's integration layer for automatic dependency injection in route handlers. This allows for seamless management of services and dependencies within FastAPI applications.

### Setup
1. Initialize FastAPI application.
2. Create a WakuFactory instance with the application's module.
3. Set the Waku application instance on `app.state.waku`.
4. Use `setup_dishka` to configure dishka with the Waku container and the FastAPI app.
5. Define `lifespan` context manager to ensure Waku resources are managed correctly.

### Example Endpoints

#### GET /

##### Description
A simple endpoint to demonstrate basic waku and FastAPI integration.

##### Method
GET

##### Endpoint
/

##### Parameters
None

##### Request Example
None

##### Response
*   **message** (str) - A greeting message.

##### Response Example
```json
{
  "message": "Hello, waku!"
}
```

#### GET /users/{user_id}

##### Description
An endpoint that utilizes injected services to retrieve user information.

##### Method
GET

##### Endpoint
/users/{user_id}

##### Parameters
*   **user_id** (str) - Path parameter representing the user's unique identifier.

##### Request Example
None

##### Response
*   **user_id** (str) - The ID of the user.
*   **message** (str) - A welcome message for the user.

##### Response Example
```json
{
  "user_id": "123",
  "message": "Hello, User 123!"
}
```

### Code Snippet
```python
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}!'


class UserService:
    def __init__(self, greeting: GreetingService) -> None:
        self._greeting = greeting

    async def welcome_user(self, user_id: str) -> dict:
        message = await self._greeting.greet(f'User {user_id}')
        return {'user_id': user_id, 'message': message}


@module(
    providers=[
        scoped(GreetingService),
        scoped(UserService),
    ],
)
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')}


@app.get('/users/{user_id}')
@inject
async def get_user(user_id: str, user_service: Injected[UserService]) -> dict:
    return await user_service.welcome_user(user_id)


if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=8000)
```
```

--------------------------------

### Waku Module System Example: Defining and Composing Modules

Source: https://context7.com/waku-py/waku/llms.txt

Demonstrates how to define interfaces, implement repositories, create services, and compose them into modules (InfrastructureModule, UserModule, AppModule) using Waku's module system and dependency injection. It shows how to create a Waku application and retrieve services from its container.

```python
import asyncio
from typing import Protocol

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


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


# Implementation
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


# Service that depends on the repository interface
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


# Infrastructure module exports the interface
@module(
    providers=[singleton(IUserRepository, InMemoryUserRepository)],
    exports=[IUserRepository],
)
class InfrastructureModule:
    pass


# User module imports infrastructure and provides UserService
@module(
    imports=[InfrastructureModule],
    providers=[scoped(UserService)],
    exports=[UserService],
)
class UserModule:
    pass


# Root module composes everything
@module(imports=[UserModule])
class AppModule:
    pass


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

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


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

```

--------------------------------

### Run Application Entrypoints

Source: https://github.com/waku-py/waku/blob/master/docs/fundamentals/application.md

Illustrates how different application entrypoints (e.g., API server, worker) can utilize the bootstrap function to create and run the WakuApplication. This highlights the separation of concerns between application bootstrapping and framework-specific integration.

```python
from app.application import bootstrap_application
from app.settings import load_settings


def run_api(settings: Settings) -> None:
    app = bootstrap_application(settings)
    fastapi_app = bootstrap_fastapi(app, settings)  # (1)!
    uvicorn.run(fastapi_app, ...)

def run_worker(settings: Settings) -> None:
    app = bootstrap_application(settings)
    worker = bootstrap_worker(app, settings)
    worker.run()


settings = load_settings()
run_api(settings)
```

--------------------------------

### Creating a Collector with `collect=False`

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/di/multi-bindings.md

This example shows how to use `many()` with `collect=False`. This registers individual implementations without creating an automatic collector, allowing for deferred collection or splitting registration across modules.

```python
from waku.di import many

# Assuming INotificationChannel is defined elsewhere
# and EmailChannel, SmsChannel are implementations

# This registers EmailChannel and SmsChannel but does NOT create a collector.
# You would need to manually provide a collector if needed.
provider = many(INotificationChannel, EmailChannel, SmsChannel, collect=False)

```

--------------------------------

### Define Application Module in Python

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Defines the root module and bootstrap function for the Waku-Py application. It utilizes ConfigModule.register for dynamic module configuration and serves as the composition root for wiring the module tree.

```python
from waku.module import Module
from waku.providers import Provider

class ConfigModule(Module):
    def __init__(self, env: str):
        self.env = env

    def configure(self, binder):
        binder.bind(str, self.env, to=Provider.constant())

class AppModule(Module):
    def configure(self, binder):
        binder.install(ConfigModule, env='dev')

async def bootstrap() -> AppModule:
    app = AppModule()
    await app.bootstrap()
    return app
```

--------------------------------

### Bootstrap WakuApplication with Settings

Source: https://github.com/waku-py/waku/blob/master/docs/fundamentals/application.md

Shows how to create a bootstrap function for a WakuApplication, integrating settings and other modules. This function serves as a central point for application creation, making it easier to manage dependencies and configurations.

```python
from waku import WakuApplication, WakuFactory, module
from waku.cqrs import MediatorConfig, MediatorModule

from app.settings import Settings


@module(
    imports=[
        MediatorModule.register(
            MediatorConfig(pipeline_behaviors=[...]),
        ),
        SettingsModule,
        InfraModule,
        DomainModule,
    ],
)
class AppModule:
    pass


def bootstrap_application(settings: Settings) -> WakuApplication:
    return WakuFactory(
        AppModule,
        context=settings.as_context(),
    ).create()
```

--------------------------------

### Registering Multiple Notification Channels with `many()`

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/di/multi-bindings.md

This example demonstrates how to use `many()` to register multiple `INotificationChannel` implementations (EmailChannel, SmsChannel, PushChannel) within an `AppModule`. The `NotificationService` then receives a sequence of these channels to dispatch notifications.

```python
from collections.abc import Sequence
from typing import Protocol

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


class INotificationChannel(Protocol):
    def send(self, recipient: str, message: str) -> str: ...


class EmailChannel(INotificationChannel):
    def send(self, recipient: str, message: str) -> str:
        return f'email to {recipient}: {message}'


class SmsChannel(INotificationChannel):
    def send(self, recipient: str, message: str) -> str:
        return f'sms to {recipient}: {message}'


class PushChannel(INotificationChannel):
    def send(self, recipient: str, message: str) -> str:
        return f'push to {recipient}: {message}'


class NotificationService:
    def __init__(self, channels: Sequence[INotificationChannel]) -> None:
        self._channels = channels

    def notify_all(self, recipient: str, message: str) -> list[str]:
        return [ch.send(recipient, message) for ch in self._channels]


@module(
    providers=[
        many(INotificationChannel, EmailChannel, SmsChannel, PushChannel),
        scoped(NotificationService),
    ],
)
class AppModule:
    pass


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

    async with app, app.container() as container:
        service = await container.get(NotificationService)
        results = service.notify_all('alice', 'Hello!')
        # results: ['email to alice: Hello!', 'sms to alice: Hello!',
        #           'push to alice: Hello!']

```

--------------------------------

### OnApplicationInit

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/extensions/custom-extensions.md

Called during app.initialize(), after all OnModuleInit hooks have completed. Used for tasks that need the application to be partially initialized.

```APIDOC
## OnApplicationInit

### Description
Called during `app.initialize()`, **after** all `OnModuleInit` hooks have completed.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from waku.application import WakuApplication
from waku.extensions import OnApplicationInit


class StartupBanner(OnApplicationInit):
    async def on_app_init(self, app: WakuApplication) -> None:
        print(f'Application started with {len(app.registry.modules)} modules')
```

### Response
#### Success Response (200)
None

#### Response Example
None
```

--------------------------------

### Waku User Module Definition (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/getting-started.md

Shows how to define a Waku module, specifying providers, imported modules, and exported services. This declarative approach centralizes dependency management and explicitates module boundaries, aiding in validation and preventing circular dependencies.

```python
@module(
    providers=[scoped(UserService)],               # (1)!
    imports=[DatabaseModule, NotificationModule],   # (2)!
    exports=[UserService],                         # (3)!
)
class UserModule:
    pass
```

--------------------------------

### Provide Migrations in Snapshot Options (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/features/eventsourcing/snapshots.md

This example demonstrates how to provide a list of migration classes (e.g., AddEmailField, RenameOwnerToName) along with the schema version in SnapshotOptions. These migrations are applied sequentially to transform older snapshot schemas to the current one.

```python
from waku.eventsourcing import EventSourcingExtension, SnapshotOptions

EventSourcingExtension().bind_aggregate(
    repository=BankAccountSnapshotRepository,
    event_types=[AccountOpened, MoneyDeposited, MoneyWithdrawn],
    snapshot=SnapshotOptions(
        strategy=EventCountStrategy(threshold=50),
        schema_version=3,
        migrations=[AddEmailField(), RenameOwnerToName()],
    ),
)
```

--------------------------------

### Main Application Execution in Python

Source: https://context7.com/waku-py/waku/llms.txt

The main function demonstrates how to create and run the Waku application. It initializes the app using WakuFactory, obtains a mediator, and sends commands to open an account and deposit funds. It then loads the aggregate to verify the state.

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

    async with app, app.container() as container:
        mediator = await container.get(IMediator)

        # Open account
        result = await mediator.send(
            OpenAccountCommand(account_id='acc-001', owner='Alice'),
        )
        print(f'Opened account: {result.account_id}')

        # Deposit money
        await mediator.send(DepositCommand(account_id='acc-001', amount=100))
        await mediator.send(DepositCommand(account_id='acc-001', amount=50))

        # Load aggregate to check state
        repo = await container.get(BankAccountRepository)
        account = await repo.load('acc-001')
        print(f'Account {account.account_id}: balance={account.balance}, owner={account.owner}')


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

--------------------------------

### Email Service Implementation and User Registration (Python)

Source: https://context7.com/waku-py/waku/llms.txt

Demonstrates an email service interface, a concrete SMTP implementation, and a UserService that utilizes the email service for user registration. Includes dependency injection setup using Waku modules.

```python
import asyncio
from typing import Protocol

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


class IEmailService(Protocol):
    async def send(self, to: str, subject: str) -> bool: ...


class SmtpEmailService(IEmailService):
    async def send(self, to: str, subject: str) -> bool:
        # Real SMTP implementation
        print(f'Sending email to {to}: {subject}')
        return True


class UserService:
    def __init__(self, email: IEmailService) -> None:
        self._email = email

    async def register_user(self, email: str) -> dict:
        await self._email.send(email, 'Welcome!')
        return {'email': email, 'status': 'registered'}


@module(
    providers=[
        singleton(IEmailService, SmtpEmailService),
        scoped(UserService),
    ],
    exports=[UserService, IEmailService],
)
class UsersModule:
    pass


# --- Test Code ---

class FakeEmailService(IEmailService):
    def __init__(self) -> None:
        self.sent_emails: list[tuple[str, str]] = []

    async def send(self, to: str, subject: str) -> bool:
        self.sent_emails.append((to, subject))
        return True


async def test_with_create_test_app() -> None:
    """Test using create_test_app with fake providers."""
    async with create_test_app(
        base=UsersModule,
        providers=[singleton(IEmailService, FakeEmailService)],
    ) as app:
        async with app.container() as container:
            user_service = await container.get(UserService)
            result = await user_service.register_user('test@example.com')
            assert result['status'] == 'registered'

            email_service = await container.get(IEmailService)
            assert isinstance(email_service, FakeEmailService)
            assert ('test@example.com', 'Welcome!') in email_service.sent_emails


async def test_with_override() -> None:
    """Test using override on a long-lived app fixture."""
    app = WakuFactory(
        module(imports=[UsersModule])(type('AppModule', (), {})) # type: ignore
    ).create()

    async with app:
        # Override provider for this test only
        with override(app.container, singleton(IEmailService, FakeEmailService)):
            async with app.container() as container:
                email_service = await container.get(IEmailService)
                assert isinstance(email_service, FakeEmailService)

        # Outside override, original provider is restored
        async with app.container() as container:
            email_service = await container.get(IEmailService)
            assert isinstance(email_service, SmtpEmailService)


if __name__ == '__main__':
    asyncio.run(test_with_create_test_app())
    asyncio.run(test_with_override())
    print('All tests passed!')

```

--------------------------------

### Default Waku Application Setup with Validation

Source: https://github.com/waku-py/waku/blob/master/docs/features/validation.md

Demonstrates how Waku automatically includes ValidationExtension in the default setup when creating an application using WakuFactory. This ensures validation runs without explicit configuration.

```python
from waku import WakuFactory, module

@module()
class AppModule: ...

# ValidationExtension with DependenciesAccessibleRule (strict=True) is applied automatically
app = WakuFactory(AppModule).create()
```

--------------------------------

### waku Module Boundaries and Dependency Injection in Python

Source: https://github.com/waku-py/waku/blob/master/README.md

Illustrates how waku modules manage visibility and dependencies. This example defines an interface `ILogger`, a concrete implementation `ConsoleLogger`, and a `UserService` that depends on `ILogger`. It showcases module imports, exports, and scoped providers.

```python
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())

```

--------------------------------

### Example: Unix Timestamp Serializer for Cross-Language Interop

Source: https://github.com/waku-py/waku/blob/master/docs/features/eventsourcing/schema-evolution.md

Provides an example of a custom event serializer, UnixTimestampEventSerializer, that serializes datetime objects as Unix timestamps. This is useful for cross-language compatibility. The code snippet is a placeholder for the actual implementation.

```python
--8<-- "docs/code/eventsourcing/custom_serializer.py"
```

--------------------------------

### Default MediatorModule Registration

Source: https://github.com/waku-py/waku/blob/master/docs/features/cqrs/index.md

Shows two equivalent ways to register the MediatorModule using default configurations. This is a simpler setup when custom configurations are not needed.

```python
# These two are equivalent:
MediatorModule.register()
MediatorModule.register(MediatorConfig())

```

--------------------------------

### Dynamic Module Configuration in Waku-Py

Source: https://context7.com/waku-py/waku/llms.txt

Illustrates how to create dynamic modules in Waku-Py that accept parameters at import time. This allows for configuration-driven provider registration, making modules more flexible and reusable.

```python
import asyncio
from dataclasses import dataclass
from typing import Literal

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


Environment = Literal['dev', 'staging', 'prod']


@dataclass
class DatabaseConfig:
    host: str
    port: int
    pool_size: int


@dataclass
class AppSettings:
    environment: Environment
    debug: bool
    db: DatabaseConfig


class DatabaseService:
    def __init__(self, settings: AppSettings) -> None:
        self.settings = settings

    def connection_string(self) -> str:
        return f'postgresql://{self.settings.db.host}:{self.settings.db.port}'


@module(is_global=True)
class ConfigModule:
    @classmethod
    def register(cls, env: Environment) -> DynamicModule:
        configs = {
            'dev': DatabaseConfig(host='localhost', port=5432, pool_size=5),
            'staging': DatabaseConfig(host='staging-db', port=5432, pool_size=10),
            'prod': DatabaseConfig(host='prod-db', port=5432, pool_size=50),
        }
        settings = AppSettings(
            environment=env,
            debug=env == 'dev',
            db=configs[env],
        )
        return DynamicModule(
            parent_module=cls,
            providers=[object_(settings)],
        )


@module(providers=[scoped(DatabaseService)])
class DatabaseModule:
    pass


@module(
    imports=[
        ConfigModule.register(env='dev'),  # Pass parameters at import time
        DatabaseModule,
    ],
)
class AppModule:
    pass


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

    async with app, app.container() as container:
        db_service = await container.get(DatabaseService)
        print(f'Connection: {db_service.connection_string()}')
        print(f'Debug mode: {db_service.settings.debug}')
        # Output:
        # Connection: postgresql://localhost:5432
        # Debug mode: True


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

```

--------------------------------

### FastAPI Integration with Waku and Dishka

Source: https://github.com/waku-py/waku/blob/master/docs/fundamentals/integrations.md

This example demonstrates how to integrate a waku application with FastAPI using dishka. It shows the steps to create a waku app, connect its container to FastAPI via setup_dishka(), and use @inject and Injected[Type] in handlers. It also explains managing waku's lifecycle through FastAPI's lifespan.

```python
from dishka.integrations.fastapi import setup_dishka
from fastapi import FastAPI
from waku.di import Injected, inject

# Assume WakuApp is defined elsewhere and configured
# from waku.app import WakuApp

# Placeholder for WakuApp
class WakuApp:
    def __init__(self):
        print("WakuApp initialized")

    async def startup(self):
        print("WakuApp startup")

    async def shutdown(self):
        print("WakuApp shutdown")


app = FastAPI()
waku_app = WakuApp()

# Connect waku's DI container to FastAPI
setup_dishka(app, waku_app.container)

@app.get("/items/{item_id}")
@inject
async def read_item(
    item_id: int,
    waku_dependency: Injected[str]  # Example dependency from waku
):
    return {"item_id": item_id, "waku_data": waku_dependency}

@app.on_event("startup")
async def startup_event():
    await waku_app.startup()

@app.on_event("shutdown")
async def shutdown_event():
    await waku_app.shutdown()

```

--------------------------------

### Mixed Marker and Has Composition

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/di/conditional-providers.md

Shows an example of combining `Marker` and `Has` using the AND operator. This activates a provider only when a specific marker is present *and* a particular type is registered in the container.

```python
from waku.di import Marker, Has

scoped(SomeService, when=Marker('debug') & Has(MetricsService))
```

--------------------------------

### Implement OnModuleConfigure Extension (Python)

Source: https://github.com/waku-py/waku/blob/master/docs/advanced/extensions/custom-extensions.md

Provides an example of an `OnModuleConfigure` extension that modifies module metadata during the `@module` decoration phase. It demonstrates adding providers to the metadata.

```python
from waku.di import scoped
from waku.extensions import OnModuleConfigure
from waku.modules import ModuleMetadata


class HealthCheck:
    async def check(self) -> bool:
        return True


class AutoRegisterHealthCheck(OnModuleConfigure):
    def on_module_configure(self, metadata: ModuleMetadata) -> None:
        metadata.providers.append(scoped(HealthCheck))
```