Request Context#

The request context holds values, such as the id of the user who sends the request, that any code in the request can read. Bind a value once, then read it wherever you need it:

import fastapi_restly as fr

class Current(fr.ContextNamespace):
    user_id: fr.ContextParam[int]

with Current.bind(user_id=42):
    print(Current.user_id())  # 42

Current is a class that you declare in your application, usually in app/current.py, as New Projects shows. Its base, fr.ContextNamespace, gives it the binding methods. Each annotation with fr.ContextParam declares one value, a member. Import the same Current wherever you use it.

There are two ways to bind a member:

For each request, a FastAPI dependency binds it from your authentication: Bind from a FastAPI dependency.
Around any block of code, such as a script or a job, Current.bind(...) binds it: Bind in a script or job.

Call the member to read it: Current.user_id() returns the bound value. Without a binding, it raises LookupError. A binding lasts until its with block exits, or until the request ends. Current.user_id == 42 and if Current.user_id: raise TypeError, because without the call they would test the member, not the bound value.

Pass the member without calling it where the value is read later. In a query clause, the member is a placeholder that gets the bound value each time the clause is applied. In a column default, SQLAlchemy calls the member when it writes the row.

Use Current for values that code across the application needs, such as the user who acts. Keep ordinary function inputs as arguments. A function that reads Current needs a binding, even though its signature does not show it.

Bind from a FastAPI dependency#

Pass your existing authentication dependency to Current.depends. Most apps have one that loads the user of the request, such as the CurrentUser alias of the FastAPI full-stack template. Pass the alias as it is:

from fastapi import APIRouter, FastAPI

import fastapi_restly as fr
from app.api.deps import CurrentUser, get_current_user  # your authentication
from app.models import User

class Current(fr.ContextNamespace):
    user: fr.ContextParam[User]

SetCurrentContextDep = Current.depends(user=CurrentUser)

router = APIRouter(dependencies=[SetCurrentContextDep])

@router.get("/me")
def me() -> dict[str, str]:
    return {"email": Current.user().email}

app = FastAPI()
app.include_router(router)

FastAPI calls get_current_user before each route of the router. The generated dependency binds the user for the request, so code in the request can read it, such as Current.user().is_superuser. Both def and async def endpoints can read it. Keep the routes that a client calls before it has a token, such as login and sign-up, out of this router: get_current_user answers them with 401.

Binding only supplies the value. It does not check who the caller is, and it does not decide what the caller may do: that is authorize.

Bind the whole user, as here, when your app already loads it for each request. Bind scalars when a worker must bind without loading a row, or when a column default or a query clause should use the member directly. The other examples on this page bind a scalar, user_id: fr.ContextParam[int].

Details

SetCurrentContextDep is already a Depends object. Add it directly to dependencies, without another Depends(...) around it.

Attach the dependency to the app, as in FastAPI(dependencies=[SetCurrentContextDep]), only when every route needs a user. A view’s dependencies also take it, and fr.include_view registers a view on a router as it does on the app. Each source can be a callable, a Depends(...) object, or an Annotated dependency alias.

One Current.depends(...) call can bind several members, each from its own dependency. The SaaS example binds four scalar members from its authentication in app/current.py.

Read the current user in a column default#

A SQLAlchemy column default can read Current. Pass the member itself, without parentheses:

created_by_id: Mapped[int] = mapped_column(
    init=False, insert_default=Current.user_id
)

SQLAlchemy calls the member when it inserts the row, usually during a flush. So keep the binding active until the flush finishes. Current.user_id() with parentheses would read the value once, when Python defines the model class. No request has bound a value then, so the import fails with LookupError.

A member that holds the whole user needs a lambda. SQLAlchemy calls the lambda when it inserts the row, and the lambda reads the member then:

created_by_id: Mapped[int] = mapped_column(
    init=False, insert_default=lambda: Current.user().id
)

Stamp the fields the server owns shows the whole recipe, with onupdate and fr.ReadOnly on the schema.

Bind in a script or job#

Outside FastAPI, use Current.bind around the code that reads the values. This standalone example writes a row with a column default that reads Current.user_id:

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

import fastapi_restly as fr


class Current(fr.ContextNamespace):
    user_id: fr.ContextParam[int]


class Base(fr.ModelDefaults, DeclarativeBase):
    pass


class Note(fr.IDMixin, Base):
    title: Mapped[str]
    created_by_id: Mapped[int] = mapped_column(
        init=False, insert_default=Current.user_id
    )


engine = create_engine("sqlite://")
Base.metadata.create_all(engine)

with Current.bind(user_id=42), Session(engine) as session:
    note = Note(title="Meeting notes")
    session.add(note)
    session.commit()
    assert note.created_by_id == 42

A worker receives its input in the job payload. Bind those values in the worker before it calls code that reads Current.

How long a binding lasts#

An inner binding replaces an outer value until its block exits. Then the previous value comes back, also when an exception leaves the block:

with Current.bind(user_id=42):
    with Current.bind(user_id=7):
        assert Current.user_id() == 7
    assert Current.user_id() == 42

Bindings use contextvars.ContextVar, not class attributes that the whole process shares. So two requests that run at the same time each see their own values. An async task gets a copy of the context in which it is created. When the parent’s binding ends, the task keeps its copy. Bind explicitly when you start work for a different user.

Put the database session in Current#

You can bind a SQLAlchemy Session or AsyncSession to a Current.session member. This works, but we do not recommend it. Pass the session as an argument to the functions that use the database.

Coroutines that you start with asyncio.create_task() or asyncio.gather() get the current bindings. If Current.session is bound, they get the same session object, not separate sessions. Give database tasks that run at the same time separate sessions. SQLAlchemy’s async scoped-session guidance also recommends passing sessions as arguments in new asyncio code.

Details

Current.session works like SQLAlchemy’s scoped_session: code finds the session in the context instead of receiving it as an argument. Current only holds the session that you bind. It does not create sessions, and it does not manage their transactions.

When code finds the session in the context, it is harder to see which operations share one transaction. Helpers that look independent can use the same session. A commit then includes the pending changes of all those helpers. Keep it clear which code creates, commits, rolls back and closes the session. See SQLAlchemy’s session lifecycle guidance.

Use another user in a test#

FastAPI’s app.dependency_overrides still works on the source dependency. Override the source, and the generated dependency binds the value that the override returns:

from fastapi.testclient import TestClient

app.dependency_overrides[get_current_user] = lambda: User(email="ann@example.com")
try:
    with TestClient(app) as client:
        assert client.get("/me").json() == {"email": "ann@example.com"}
finally:
    app.dependency_overrides.pop(get_current_user)

A test that calls application code directly, without a request, binds the values with Current.bind(...) around the call. That does not test the request dependency, so keep a test that goes through a request too. Testing covers the test setup.

See what is bound#

Current.explain lists each member with its value, and the file and line that bound it. A member without a binding shows as UNBOUND:

# app/example.py
with Current.bind(user_id=42):
    print(Current.explain())

print(Current.explain())

Output:

Current
└─ user_id = 42   bound at app/example.py:2 (<module>)
Current
└─ user_id: UNBOUND

For a generated dependency, the line is the Current.depends(...) call. Read this output when a value is missing or not what you expect.