nuke-di

The simplest dependency injection for async Python projects.

Dependencies are declared with plain type hints. nuke-di builds the dependency tree, creates every client once and drives its async lifecycle: connect() on startup and disconnect() on shutdown, in reverse order.

It was extracted from the DI layer of a production Python microservice framework and has no runtime dependencies.

Installation

pip install nuke-di

Requires Python 3.11+.

Quick start

import asyncio

from nuke_di import DI, Client


class Database(Client):
    async def connect(self) -> None:
        print("database: connected")

    async def disconnect(self) -> None:
        print("database: disconnected")

    async def fetch_user(self, user_id: int) -> str:
        return f"user-{user_id}"


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

    async def greet(self, user_id: int) -> str:
        return f"Hello, {await self._db.fetch_user(user_id)}!"


async def handler(user_id: int, users: UserService) -> str:
    return await users.greet(user_id)


async def main() -> None:
    injected = DI.inject(handler)  # resolves UserService -> Database

    async with DI:  # connect() every client, disconnect() on exit
        print(await injected(42))


asyncio.run(main())

database: connected
Hello, user-42!
database: disconnected

Concepts

Clients

Every dependency is a subclass of one of two base classes:

Override the async connect() / disconnect() methods to open and release resources such as connection pools:

class Redis(Client):
    def __init__(self) -> None:
        self._pool: Pool | None = None

    async def connect(self) -> None:
        self._pool = await create_pool()

    async def disconnect(self) -> None:
        if self._pool is not None:
            await self._pool.close()
            self._pool = None

Composition

A client declares its own dependencies in __init__. Only arguments annotated with a client type are injected; resolution is recursive.

class BusinessLogic(Client):
    def __init__(self, pg: Postgres, grpc: GrpcClient) -> None:
        self._pg = pg
        self._grpc = grpc

Container

Dependencies is the container. DI is a ready-to-use global instance; create your own when you need isolation, e.g. in tests.

resolve, inject, mock and flush only work while the container is disconnected: the whole tree is built before startup.

A failing disconnect() is logged and does not stop the other clients from shutting down.

Dataclass clients

client_dataclass turns a class into a Client and a dataclass at once, so the fields become the injected dependencies:

from nuke_di import client_dataclass


@client_dataclass(frozen=True)
class Checkout:
    pg: Postgres
    payments: PaymentsClient

It accepts the same keyword arguments as dataclasses.dataclass.

Testing

Register mocks before the tree is resolved; every consumer then receives the mock.

from unittest.mock import call

from nuke_di import Dependencies


async def test_greet() -> None:
    deps = Dependencies()
    db = deps.mock(Database)
    db.fetch_user.return_value = "alice"

    users = deps.resolve(UserService)
    async with deps:
        assert await users.greet(1) == "Hello, alice!"

    assert db.fetch_user.await_args_list == [call(1)]

Configuration

The value is read when a Dependencies instance is created. You can also pass it explicitly:

from nuke_di import Dependencies, DependenciesSettings

deps = Dependencies(settings=DependenciesSettings(connect_timeout=5))

Errors

InitializeDependencyError and ConnectError derive from SystemExit: an application whose dependencies cannot start is expected to stop. Catch them explicitly if you need different behavior; the original exception is available as __cause__.

nuke-di logs through the standard logging module under the nuke_di logger.

Development

uv sync
uv run pytest --cov
uv run ruff check . && uv run ruff format --check .
uv run mypy

License

MIT