nodstar

nodnod integration for Litestar. Declare dependency lifetimes on nodes, inject into handlers by type.

Install

uv add nodstar

Requires Python 3.14+.

Usage

from nodnod import scalar_node
from litestar import Litestar, get
from nodstar import NodstarPlugin, Node, global_node, request

Define nodes

Decorate with a lifetime (@global_node, @request, @per_call) and @scalar_node:

@global_node
@scalar_node
class DatabasePool:
    @classmethod
    async def __compose__(cls) -> AsyncEngine:
        engine = create_async_engine(DATABASE_URL)
        yield engine
        await engine.dispose()


@request
@scalar_node
class DbSession:
    @classmethod
    async def __compose__(cls, pool: DatabasePool) -> AsyncSession:
        async with AsyncSession(pool) as session:
            yield session

Inject into handlers

Annotate a handler parameter with a node type — nodnod resolves the dependency graph, Litestar injects the value. Injection is by type, so the parameter can be named anything:

@get("/users")
async def get_users(session: DbSession) -> list[User]:
    return await session.scalars(select(User))

Optionally wrap the type in Node[T] for precise static typing — it resolves to T for the type checker (nodes are otherwise seen as type[T]):

@get("/users")
async def get_users(session: Node[DbSession]) -> list[User]:
    return await session.scalars(select(User))

Both forms are equivalent at runtime.

Wire up

app = Litestar(
    route_handlers=[get_users],
    plugins=[NodstarPlugin()],
)

That's it. No dependencies={...}, no manual Provide(), no container configuration.

Lifetimes

Nodes declare their own lifetime. The dependency graph is resolved automatically — a @request node can depend on a @global_node, and nodnod will pull the value from the parent scope.

How it works

1. Lifetime decorators register nodes in a global registry
2. On app init, NodstarPlugin walks every route handler (including those on Controllers and Routers) and inspects its type hints
3. For each parameter whose type is a registered node, the plugin binds a Provide to that handler under the parameter's own name and marks it skip_validation=True, so matching is by type, not by parameter name
4. On startup, @global_node nodes are composed into an app-wide scope
5. Per request, a child scope is created and @request/@per_call nodes are composed; the provider pulls the unboxed value from that scope
6. Node[T] is an optional type-level alias that resolves to T for type checkers; at runtime it is just Annotated[T, Dependency(skip_validation=True)] and is treated identically to a bare T annotation

Generator lifecycle

Use yield in __compose__ for setup/teardown:

@request
@scalar_node
class DbSession:
    @classmethod
    async def __compose__(cls, pool: DatabasePool) -> AsyncSession:
        async with AsyncSession(pool) as session:
            yield session
            # teardown runs when request scope closes

License

MIT