Skip to content

Providers

Introduction

A provider tells waku how to create and deliver a dependency — what type it satisfies, how long it lives, and what it needs in order to be constructed. Providers are the building blocks of the module system: each module declares which providers it owns and which it exposes to other modules.

Dependency Injection

waku provides a module system that lets you organize providers into cohesive, self-contained units with explicit import/export boundaries. At bootstrap, waku collects providers from all modules, resolves the module dependency graph, and hands the result to the Dishka IoC container, which handles dependency resolution and lifecycle management.

What is Dependency Injection?

Dependency Injection (DI) is a design pattern that addresses the issue of tightly coupled code by decoupling the creation and management of dependencies from the classes that rely on them. In traditional approaches, classes directly instantiate their dependencies, resulting in rigid, hard-to-maintain code. DI solves this problem by enabling dependencies to be supplied externally, typically through mechanisms like constructor or setter injection.

By shifting the responsibility of dependency management outside the class, DI promotes loose coupling, allowing classes to focus on their core functionality rather than how dependencies are created. This separation enhances maintainability, testability, and flexibility, as dependencies can be easily swapped or modified without altering the class's code. Ultimately, DI improves system design by reducing interdependencies and making code more modular and scalable.

See also: Dishka — Introduction to DI

Manual DI Example
from abc import ABC, abstractmethod


# Use an interface to define contract for clients
# This allows us injecting different implementations
class IClient(ABC):
    @abstractmethod
    def request(self, url: str) -> str:
        pass


# Regular implementation
class RealClient(IClient):
    def request(self, url: str) -> str:
        # Some HTTP requesting logic
        return f'"{url}" call result'


# Implementation for tests
class MockClient(IClient):
    def __init__(self, return_data: str) -> None:
        self._return_data = return_data

    def request(self, url: str) -> str:
        # Mocked behavior for testing
        return f'{self._return_data} from "{url}"'


class Service:
    # Accepts any IClient implementation
    def __init__(self, client: IClient) -> None:
        self._client = client

    def do_something(self) -> str:
        return self._client.request('https://example.com')


# Usage in regular code
real_client = RealClient()
service = Service(real_client)
print(service.do_something())  # Output: "https://example.com" call result

# Usage in tests
mocked_client = MockClient('mocked data')
service = Service(mocked_client)
print(service.do_something())  # Output: mocked data from "https://example.com"

Here, a MockClient is injected into Service, making it easy to test Service in isolation without relying on a real client implementation.

What is IoC-container?

An IoC container is a framework that automates object creation and dependency management based on the Inversion of Control (IoC) principle. It centralizes the configuration and instantiation of components, reducing tight coupling and simplifying code maintenance. By handling dependency resolution, an IoC container promotes modular, testable, and scalable application design.

With the power of an IoC container, you can leverage all the benefits of DI without manually managing dependencies.

See also: Dishka — Key Concepts

Providers

A Provider is a class that holds dependency metadata, such as its type, lifetime scope and factory.

In waku, there are five provider helpers:

Each provider (except object_() and contextual()) accepts two positional arguments:

  • interface_or_source: the type to register — or the interface type when a separate implementation is provided.
  • implementation (optional): the implementation type or factory. When given, the first argument is treated as the interface.

Source types

Every provider helper accepts classes and callables as the source of a dependency. The container inspects the source's type hints to determine what to inject.

Classes

The most common source — the container calls the constructor and injects parameters:

1
2
3
4
5
6
7
8
9
from waku.di import scoped


class UserRepository:
    def __init__(self, db: Database) -> None:
        self._db = db


scoped(IUserRepository, UserRepository)

Factory functions

Any callable with a return type annotation works as a source. The container injects the function's parameters and uses the return value:

1
2
3
4
5
6
7
8
from waku.di import scoped


def create_repository(db: Database) -> UserRepository:
    return UserRepository(db)


scoped(IUserRepository, create_repository)

Async factory functions

Async factories work the same way — the container awaits them during resolution:

1
2
3
4
5
6
7
8
from waku.di import scoped


async def create_repository(db: Database) -> UserRepository:
    return UserRepository(db)


scoped(IUserRepository, create_repository)

Generator factories

Generator factories support resource cleanup. Yield the dependency; code after yield runs when the scope exits:

from collections.abc import Iterator

from waku.di import scoped


def create_session(pool: ConnectionPool) -> Iterator[Session]:
    session = pool.acquire()
    yield session
    session.close()


scoped(ISession, create_session)

Async generators follow the same pattern with AsyncIterator:

from collections.abc import AsyncIterator

from waku.di import singleton


async def create_pool() -> AsyncIterator[DatabasePool]:
    pool = DatabasePool()
    yield pool
    await pool.close()


singleton(IDatabasePool, create_pool)
Working with context managers

Do not apply @contextmanager or @asynccontextmanager to generator factories — Dishka manages the generator lifecycle directly and these decorators interfere with that mechanism. Use plain generators as shown above.

To wrap an existing context manager class, enter it inside a generator factory:

from collections.abc import AsyncIterator

from waku.di import scoped


async def create_connection(config: DbConfig) -> AsyncIterator[Connection]:
    async with Connection(config.dsn) as conn:
        yield conn


scoped(IConnection, create_connection)

For automatic enter/exit of all context manager dependencies without writing individual wrappers, see dishka#457 for a @decorate-based recipe.

Scopes

Provider helper names are inspired by .NET Core's service lifetimes. Under the hood, each helper maps to a Dishka scope that determines the dependency's lifetime. Dishka uses two primary scopes:

  • APP — the dependency lives for the entire application lifetime.
  • REQUEST — the dependency lives for a single scope entry (e.g., one HTTP request).

Dependencies are lazy — they are created when first requested. Within a scope, the same instance is returned by default (configurable per helper). When a scope exits, all its dependencies are finalized in reverse creation order.

For more details, see the Dishka scopes documentation.

Transient

Registers the dependency in Scope.REQUEST with caching disabled. A new instance is created every time the dependency is requested, even within the same scope.

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


@module(providers=[transient(list)])
class AppModule:
    pass


async def main() -> None:
    application = WakuFactory(AppModule).create()
    async with application:
        async with application.container() as request_container:
            obj_1 = await request_container.get(list)
            obj_2 = await request_container.get(list)
            assert obj_1 is not obj_2

        # Providers are disposed at this point

Scoped

Registers the dependency in Scope.REQUEST with caching enabled. The instance is created once per scope entry and reused for all subsequent requests within that scope. Finalized when the scope exits.

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


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


async def main() -> None:
    application = WakuFactory(AppModule).create()
    async with application:
        async with application.container() as request_container:
            obj_1 = await request_container.get(list)
            obj_2 = await request_container.get(list)
            assert obj_1 is obj_2

        # Providers are disposed at this point

Singleton

Registers the dependency in Scope.APP with caching enabled. The instance is created once on first request and reused across all scopes for the entire application lifetime. Finalized when the application shuts down.

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


@module(providers=[singleton(list)])
class AppModule:
    pass


async def main() -> None:
    application = WakuFactory(AppModule).create()
    async with application:
        async with application.container() as request_container:
            obj_1 = await request_container.get(list)

        async with application.container():
            obj_2 = await request_container.get(list)

        assert obj_1 is obj_2

    # Providers are disposed at this point

Object

Registers a pre-created instance in Scope.APP. Unlike singleton(), you provide the instance directly — its lifecycle is managed by you, not the container.

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

some_object = (1, 2, 3)


@module(
    providers=[
        object_(some_object, provided_type=tuple),
    ],
)
class AppModule:
    pass


async def main() -> None:
    application = WakuFactory(AppModule).create()
    async with application, application.container() as request_container:
        obj = await request_container.get(tuple)
        assert obj is some_object

    # Providers are not disposed at this point automatically
    # because they are not part of the application container lifecycle

Contextual

contextual() injects external objects — values that originate outside the DI container and have their own lifecycle (HTTP requests, database transactions, event data).

When to use:

  • Framework integration: HTTP request objects, user sessions, authentication contexts
  • Event-driven scenarios: Queue messages, webhooks, callback data
  • External resources: Database transactions, file handles, network connections managed by external systems
  • Per-request data: Any data that varies per request/operation and originates from outside your application

How it works:

  1. Declare the contextual dependency using the contextual provider in your module
  2. Use the dependency in other providers just like any regular dependency
  3. Provide the actual value when entering the container scope using the context= parameter. Context can be provided at APP level via WakuFactory(context=...) or at REQUEST level via app.container(context=...) — see Application — Container Access for details

contextual() accepts two arguments:

  • provided_type: the type of the dependency to be injected.
  • scope: the scope where the context is available (defaults to Scope.REQUEST).
from waku import WakuFactory, module
from waku.di import contextual, Scope

some_object = (1, 2, 3)


@module(
    providers=[
        contextual(provided_type=tuple, scope=Scope.REQUEST),
    ],
)
class AppModule:
    pass


async def main() -> None:
    application = WakuFactory(AppModule).create()
    async with (
        application,
        application.container(
            context={tuple: some_object},
        ) as request_container,
    ):
        obj = await request_container.get(tuple)
        assert obj is some_object

    # Providers are not disposed at this point automatically
    # because they are not part of the application container lifecycle

FastAPI example — injecting the current request into your service layer:

from fastapi import FastAPI, Request
from waku import WakuFactory, module
from waku.di import contextual, scoped, Scope


class UserService:
    """Service that uses the current HTTP request for user-specific operations."""

    def __init__(self, request: Request) -> None:
        self.request = request

    def get_user_info(self) -> dict[str, str]:
        """Extract user information from the request headers."""
        return {
            'user_id': self.request.headers.get('user-id', 'anonymous'),
            'session_id': self.request.headers.get('session-id', 'none'),
            'user_agent': self.request.headers.get('user-agent', 'unknown'),
        }


@module(
    providers=[
        contextual(provided_type=Request, scope=Scope.REQUEST),
        scoped(UserService),
    ],
)
class AppModule:
    pass


# FastAPI application setup
app = FastAPI()
application = WakuFactory(AppModule).create()


@app.get('/user-info')
async def get_user_info(request: Request) -> dict[str, str]:
    """Endpoint that uses contextual dependency injection."""
    async with (
        application,
        application.container(
            context={Request: request},
        ) as request_container,
    ):
        # UserService automatically receives the current HTTP request
        user_service = await request_container.get(UserService)
        return user_service.get_user_info()


# Example usage:
# curl -H "user-id: john123" -H "session-id: abc456" http://localhost:8000/user-info

Important

In this example, the contextual provider and waku itself are used to manually inject the current request into the UserService. However, in real-world applications, you should use the Dishka FastAPI integration to inject the request automatically.

Where and how to inject dependencies?

To inject dependencies with waku, you need to:

  1. Register them as providers with the desired scope in modules.
  2. Identify your application entrypoints and decorate them with the @inject decorator for your framework. Consult the Dishka integrations section for your framework to find out how to do this.
  3. Add dependencies as arguments to your entrypoint signature using the Injected type hint.

Further reading