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.

Nodes that need the request

A node can depend on the live connection with Connection, so authentication and anything else derived from headers is an ordinary node rather than a hand-written Provide:

from nodstar import Connection

@request
@scalar_node
class Principal:
    @classmethod
    async def __compose__(cls, connection: Connection, session: DbSession) -> Principal:
        token = connection.headers.get("Authorization")
        ...


@get("/me")
async def me(principal: Node[Principal]) -> UserRead: ...

The injected value is Litestar's own connection object. Composing such a node outside a request raises, since there is nothing to inject.

Reaching nodes from guards and hooks

Litestar calls guards, before_request hooks and exception handlers with a fixed signature, so they cannot take Node[...] parameters. from_connection is the way in:

from nodstar import from_connection

async def authenticated(connection: ASGIConnection, _: BaseRouteHandler) -> None:
    principal = await from_connection(connection, Principal)
    if not principal.allows(...):
        raise PermissionDeniedException

Guards run before handler DI, but composition is shared per request: the handler's Node[Principal] reuses what the guard already composed.

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.

Composition is lazy: a request pays only for the nodes it actually asks for, so a handler that touches no node never opens a session, and an authentication node never runs on a public route. Within one request each @request node is composed once and shared by every consumer, including concurrently resolved parameters.

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, and an agent is built (not run) per request-scoped node so a broken graph fails at startup
5. Per request, a middleware creates a child scope but composes nothing. A provider composes its node's subtree on first use, into the child scope for @request and into a throwaway grandchild for @per_call; both are closed after the response
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