Protecting routes¶

auth.current_user() builds a FastAPI dependency that authenticates the request and returns a Principal. Every authorization rule is a keyword on that one factory, so you compose access control instead of writing it.

from fastapi import Depends
from crudauth import Principal

@app.get("/dashboard")
async def dashboard(user: Principal = Depends(auth.current_user())):
    return {"id": user.user_id}

With no keywords it just requires a valid credential: an authenticated request gets the Principal, an anonymous one gets 401 Unauthorized.

The Principal¶

Whatever transport authenticated the request, the dependency yields the same object:

@app.get("/me")
async def me(user: Principal = Depends(auth.current_user())):
    return {"id": user.user_id, "via": user.transport, "email": user.user.email}

Authorization keywords¶

Each keyword adds a check. They stack, and a failed check raises the matching status.

auth.current_user()                          # required, 401 if anonymous
auth.current_user(optional=True)             # returns None instead of raising
auth.current_user(superuser=True)            # 403 unless is_superuser
auth.current_user(verified=True)             # 403 unless email_verified
auth.current_user(scopes=["reports:read"])   # 403 unless the credential's scopes cover these
auth.current_user(transport="bearer")        # only accept this transport (or a list)

scopes is a superset check: the credential must carry every scope you list. transport narrows which credentials are accepted, which is useful when an endpoint should be API-only or browser-only.

Combining keywords¶

Keywords compose, so a route can require several things at once:

@app.get("/reports")
async def reports(
    user: Principal = Depends(auth.current_user(superuser=True, verified=True, scopes=["reports:read"])),
):
    ...

Custom checks¶

check= runs your own predicate on the resolved principal, after the built-in keywords. It can be sync or async. Returning False denies with 403; to deny with a different status or message, raise your own exception from inside the check.

def owns_team(user: Principal) -> bool:
    return user.user.team_id is not None

@app.get("/team")
async def team(user: Principal = Depends(auth.current_user(check=owns_team))):
    ...

Returning anything that isn't False (including None) allows the request, so a raise-to-deny callback that simply returns nothing on success also works.

Optional authentication¶

Pass optional=True to make a route public but personalized. The dependency returns None for anonymous callers instead of raising:

@app.get("/products")
async def products(user: Principal | None = Depends(auth.current_user(optional=True))):
    if user is not None:
        ...  # personalize for the logged-in user
    return ...

A present but invalid credential (for example a session cookie that fails its CSRF check on a mutation) still raises, even under optional=True. A tampered credential is treated as an attack signal, not as "anonymous".

Resolving outside dependency injection¶

Middleware and other request-level code can resolve the same principal directly:

@app.middleware("http")
async def audit(request: Request, call_next):
    principal = await auth.resolve_principal(request)
    if principal is not None:
        logger.info("request from user %s", principal.user_id)
    return await call_next(request)

This method returns None for anonymous or invalid credentials, does not enforce CSRF, and does not slide session activity by default. Pass update_activity=True when middleware should count the request as session activity.

A later current_user() dependency in the same request reuses the result: it reloads the user through its own DB session, and still applies its CSRF check, the session activity update and its authorization gates. Invalid credentials aren't cached, so that dependency still rejects them.

resolve_principal opens its own DB session by calling your session dependency directly, so FastAPI's dependency_overrides don't apply to it.

Protecting a whole router¶

Attach the dependency at the router level to gate every route under it:

from fastapi import APIRouter

admin = APIRouter(prefix="/admin", dependencies=[Depends(auth.current_user(superuser=True))])

@admin.get("/stats")
async def stats():
    ...  # already gated; reached only by superusers

Router-level dependencies enforce access but don't inject a value into the handler. If a route needs the Principal, add current_user() to that route as well.

Composing with rate limits and sudo¶

current_user() is one dependency among several. Combine it with throttling or re-authentication on the same route:

from crudauth.ratelimit import RateLimit

@app.post(
    "/account/close",
    dependencies=[Depends(auth.rate_limit("account_close", RateLimit(3, 3600)))],
)
async def close_account(
    user: Principal = Depends(auth.current_user(superuser=False)),
    _: Principal = Depends(auth.require_sudo()),
):
    ...

See the rate limiting and sudo reference pages for those pieces.

Next: Sessions →