Validators¶

Every source wraps a validator: a function you write that turns a credential into a user.

The contract¶

A validator receives the credential and returns the user (any object), or raises to refuse it. It can be sync or async.

from fastapi import FastAPI, Security

from fastapi_multiauth import (
    ForbiddenError,
    HTTPBearerAuth,
    UnauthorizedError,
    hash_token,
)

USERS = {
    hash_token("alice-token"): {"name": "alice", "disabled": False},
    hash_token("bob-token"): {"name": "bob", "disabled": True},
}


async def validate_token(token: str) -> dict:
    user = USERS.get(hash_token(token))
    if user is None:
        raise UnauthorizedError()  # 401: unknown or invalid
    if user["disabled"]:
        raise ForbiddenError()  # 403: valid, but not allowed
    return user


bearer = HTTPBearerAuth(validator=validate_token)

app = FastAPI()


@app.get("/me")
async def me(user: dict = Security(bearer)):
    return user

Raise UnauthorizedError (401) when the credential is missing or invalid.
Raise ForbiddenError (403) when it is valid, but not allowed here.

The library adds the WWW-Authenticate header to 401 responses for you.

Extra arguments¶

Keyword arguments given to a source are passed to its validator. require() returns a copy of the source with more of them, to tighten a single route:

from fastapi import FastAPI, Security

from fastapi_multiauth import (
    ForbiddenError,
    HTTPBearerAuth,
    UnauthorizedError,
    hash_token,
)

USERS = {
    hash_token("alice-token"): {"name": "alice", "role": "admin"},
    hash_token("bob-token"): {"name": "bob", "role": "member"},
}


async def validate_token(token: str, *, role: str | None = None) -> dict:
    user = USERS.get(hash_token(token))
    if user is None:
        raise UnauthorizedError()
    if role is not None and user["role"] != role:
        raise ForbiddenError()
    return user


bearer = HTTPBearerAuth(validator=validate_token)

app = FastAPI()


@app.get("/me")
async def me(user: dict = Security(bearer)):
    return user


@app.get("/admin")
async def admin(user: dict = Security(bearer.require(role="admin"))):
    return user

scopes, session_id and request are reserved names: the library fills them in itself.

Reaching the request¶

Declare a request parameter to receive the current request, for example to leave data on request.state for the route:

from fastapi import FastAPI, Request, Security

from fastapi_multiauth import HTTPBearerAuth, UnauthorizedError, hash_token

TOKENS = {hash_token("alice-token"): ("alice", 7)}


async def validate_token(token: str, request: Request | None = None) -> str:
    row = TOKENS.get(hash_token(token))
    if row is None:
        raise UnauthorizedError()
    username, token_id = row
    if request is not None:
        request.state.token_id = token_id
    return username


bearer = HTTPBearerAuth(validator=validate_token)

app = FastAPI()


@app.get("/me")
async def me(request: Request, username: str = Security(bearer)):
    return {"name": username, "token_id": request.state.token_id}

Keep the None default: a validator can also be called outside a request, from a script or a background job.