Scoped context on contextvars

nodrill

Provide a value once, use it anywhere below

• Python 3.10–3.14
• Zero dependencies
• Typed · py.typed

Set the database handle or the request id once, at the boundary, and read it ten frames down. The functions in between keep the signatures they already have, and every lookup stays thread-safe and asyncio-task-safe.

Get started View on GitHub

$ pip install nodrill Copy Copied

checkout.py — drilled

def handle(request, db, tenant):
    return place_order(request.json, db, tenant)

def place_order(payload, db, tenant):  # uses neither
    return charge(payload, db, tenant)

def charge(payload, db, tenant):
    db.bill(tenant, payload["total"])

checkout.py — with nodrill

def handle(request, db, tenant):
    with provider("scope", db=db, tenant=tenant):
        return place_order(request.json)

def place_order(payload):       # just the payload
    return charge(payload)

def charge(payload):
    scope = use("scope")
    scope.db.bill(scope.tenant, payload["total"])

Not a container, not a global

It carries a value you already made down the stack, and does nothing else.

Nothing is constructed

No dependency graph, no lifetimes, no cache. use() returns exactly what a provider upstream put there, and the lookup is one dict read rather than a graph walk.

Nothing outlives its block

Values live in contextvars, scoped to the with that set them and restored on the way out, even when the block raises. One request cannot read what the last one left.

It composes with what you have

Let dishka or your own factories build things, and FastAPI's Depends own the endpoint edge. nodrill carries the per-request scope past the layers neither of them reaches.

The whole boundary, down to what is deliberately left out, is written up under what nodrill is for.

What the library gives you

Four pieces, each one a thin layer over the standard library.

Typed lookups

Classes are the keys

A string namespace is quick to write; a class is the one your editor completes. Provide an instance and it registers under its own class, so the lookup below hands it back with the static type intact, under mypy and pyright alike.

• use(Scope) is inferred as Scope, not Any
• An inner block overrides one tenant, the outer scope returns on exit
• A miss tries set_default, then default=, then raises

scope.py

@dataclass
class Scope:
    db: Session
    tenant: str

with provider(Scope(db=session, tenant="acme")):
    place_order(payload)

def charge(total: int) -> None:
    scope = use(Scope)          # -> Scope
    scope.db.bill(scope.tenant, total)

Injection

Keep the dependency in the signature

When a function should say out loud what it needs, annotate the parameter and @inject fills it from the active context at call time. An argument you pass explicitly always wins, which is what makes the test on the last line a one-liner.

• Explicit arguments win, so tests need no providers
• The wrapper compiles once, at decoration, mirroring your signature
• Works on async def, methods and classmethods

billing.py

@inject
def refund(order, gw: FromCtx[Gateway] = injected) -> None:
    gw.credit(order.id, order.total)

with provider(StripeGateway(key=SECRET)):
    refund(order)                 # from the context

refund(order, gw=FakeGateway())   # a test passes one

Concurrency

The scope survives the fan-out

asyncio inherits the context by itself, so a task created inside a provider block already sees it. Plain threads do not, and that is where a request id usually disappears from the logs. Two helpers close the gap.

• wrap binds a callable to the context active right now
• Executor is a ThreadPoolExecutor, so submit and map are unchanged
• A worker's writes stay in the worker, and siblings stay isolated

upload.py

with provider("request", id=request_id):
    with Executor(max_workers=4) as pool:
        pool.map(thumbnail, upload.pages)

    Thread(target=wrap(warm_cache)).start()

def thumbnail(page):
    log.info("resizing", request_id=use("request").id)

Testing

Nothing to mock, nothing to reset

Injected code under test is still a function you can call with arguments, so the common case needs no fixture at all. For the tests that do exercise providers, isolate() hands each one a clean slate and puts the outer state back afterwards.

• No patching of module globals, because there are none
• isolate() clears providers and rolls back set_default
• active() is a read-only view of what is provided right now

test_billing.py

@pytest.fixture(autouse=True)
def _context():
    with nodrill.isolate():      # fresh state per test
        yield

def test_refund_credits_the_order():
    gw = FakeGateway()
    refund(order, gw=gw)          # no provider needed
    assert gw.credits == [(order.id, order.total)]

frozen=True

Hand consumers a read-only view while the block keeps the writable object. A callee that assigns gets FrozenContextError instead of a bug three layers up.

set_default

Declare the fallback for a class once, and code that also has to run outside any provider keeps working. The factory runs per miss, so it is never a cached singleton.

nodrill.context

An ambient attribute namespace for the values that are not shaped like a scope, on its own ContextVar so it cannot shadow anything a provider set.

Errors that help

A miss names the key, lists what is active, and suggests the close match. Under debug() it goes further and names the thread, the task and the line the provider is actually open on.

NoProviderError: use('datbase'): no active provider for 'datbase'. Active providers: 'cache', 'database'. Did you mean 'database'?

Ten minutes to the whole library

The public surface is small enough to read in one sitting, and the tutorial builds one small application around the names you reach for first.

Read the tutorial API reference