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. Independent clients start concurrently, layer by layer, from the deepest dependencies up.

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

Layers

Clients connect concurrently in layers. Clients without dependencies form layer 0; every other client sits one layer above its highest dependency. A layer starts only after the previous one has connected, so a client never connects before its own dependencies. disconnect() walks the layers in reverse.

Checkout(pg: Postgres, payments: Payments)    layer 2
Payments(pg: Postgres)                        layer 1
Postgres, Redis                               layer 0  <- connect together

Only dependencies declared in __init__ are ordered. If a client needs another one to be connected first, declare it as a dependency.

If a client fails to connect, the rest of its layer is cancelled and the next layers never start. The clients that already connected are disconnected, layers in reverse, and the container is left disconnected and empty; the same happens when connect() itself is cancelled. Mocked clients are not connected and do not affect the layers.

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 or hanging disconnect() is logged and does not stop the other clients from shutting down: each one is bounded by DISCONNECT_TIMEOUT_SECONDS.

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.

Workers and jobs

An async function becomes the main program of a process with one decorator:

# app/jobs/sync.py
from nuke_di import job

from app.clients import Postgres, Warehouse


@job
async def sync(pg: Postgres, warehouse: Warehouse) -> None:
    for batch in await warehouse.changed_batches():
        await pg.upsert(batch)

python -m app.jobs.sync

Every argument is a client, injected from the global DI container. The process resolves the clients, connects them, runs the function, disconnects them and exits with an exit code. Scheduling is not part of the library: a Kubernetes CronJob, a systemd timer or crontab decides when a job runs.

When the module is run as __main__, the decorator runs the function right away and the process exits there, so code below the decorated function never runs: keep one entrypoint per module and define it last. On a normal import the decorator returns the function unchanged, so a test calls it directly with mocks:

from unittest.mock import AsyncMock


async def test_sync() -> None:
    pg, warehouse = AsyncMock(), AsyncMock()
    warehouse.changed_batches.return_value = [batch]

    await sync(pg, warehouse)

    pg.upsert.assert_awaited_once_with(batch)

The decorated function must be declared with async def, otherwise TypeError is raised on import.

Shutdown

On the first SIGTERM or SIGINT the Shutdown client is set. An entrypoint that depends on it can finish its current piece of work and return:

from nuke_di import Shutdown, worker


@worker
async def consumer(queue: Queue, shutdown: Shutdown) -> None:
    while not shutdown.is_set():
        message = await queue.get()
        await message.process()
        await message.ack()

await shutdown.wait() blocks until the Shutdown begins. If the entrypoint is still running after SHUTDOWN_GRACE_SECONDS, it is cancelled; a second signal cancels it immediately. A signal that arrives while the clients are connecting stops the startup, and the clients that already connected are disconnected.

A worker that returns or raises on its own also ends the process: restarting it is the orchestrator's job.

In the worst case a process stops in SHUTDOWN_GRACE_SECONDS + DISCONNECT_TIMEOUT_SECONDS × layers. With the defaults, a tree of two layers takes the whole Kubernetes default terminationGracePeriodSeconds of 30 seconds, so lower the timeouts or raise the grace period for deeper trees.

On Windows only SIGINT (Ctrl+C) is handled; SIGTERM keeps its default behavior.

Background tasks

BackgroundTasks is a client that supervises coroutines running alongside the entrypoint:

@worker
async def indexer(tasks: BackgroundTasks, search: Search, shutdown: Shutdown) -> None:
    tasks.spawn(search.refresh_loop(), name="refresh")
    await shutdown.wait()

A failing task is logged with its traceback, and inside a worker or a job it fails the whole process: the entrypoint is cancelled and the exit code is 1. When the process stops, the tasks are cancelled and awaited before any client disconnects, so they never run against closed clients. Outside a worker or a job, e.g. under a plain async with DI, failures are only logged and the tasks are cancelled on disconnect().

Exit codes

The first matching rule wins:

SIGTERM gives 143 and SIGINT gives 130. A job that sees a Shutdown and returns cleanly still exits with 128 + signum: its work was interrupted, and a scheduler must not count it as complete.

Hooks

Hooks observe every run, e.g. to push metrics or open a tracing span:

from nuke_di import Run, job


class Metrics:
    async def on_start(self, run: Run) -> None:
        print(f"{run.kind} {run.name} started at {run.started_at}")

    async def on_finish(self, run: Run) -> None:
        print(f"{run.name} exited with {run.exit_code}, error: {run.error!r}")


@job(hooks=[Metrics()])
async def sync(pg: Postgres) -> None: ...

on_start is called in list order before the clients are resolved; on_finish in reverse order after they have disconnected, so it sees the final exit_code, error, signal and finished_at, connect failures included. Hooks are plain objects, not clients: they manage their own resources. An exception in a hook is logged and does not change the exit code.

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 container settings are read when a Dependencies instance is created. You can also pass them explicitly:

from nuke_di import Dependencies, DependenciesSettings

deps = Dependencies(settings=DependenciesSettings(connect_timeout=5, disconnect_timeout=5, connect_concurrency=4))

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

make install   # uv sync --locked
make check     # ruff, mypy and tests, as in CI
make cov       # tests with a coverage report (terminal + htmlcov/)
make test-all  # tests on Python 3.11-3.14

Line and branch coverage is 100%, and CI fails if it drops below that (fail_under = 100 in pyproject.toml).

License

MIT