CRUDAuth¶

crudauth.CRUDAuth ¶

CRUDAuth(
    *,
    session: Callable[..., Any],
    user_model: type[Any],
    SECRET_KEY: str,
    transports: Sequence[Transport] | None = None,
    column_map: dict[str, str] | None = None,
    oauth: dict[str, Any] | None = None,
    email: Any = None,
    hooks: AuthHooks | None = None,
    redirect_base_url: str | None = None,
    algorithm: str = DEFAULT_ALGORITHM,
    cookies: CookieConfig | None = None,
    register_schema: type[BaseModel] | None = None,
    register_extra_fields: set[str] | None = None,
    rate_limiter: RateLimiterBackend | None = None,
    rate_limits: dict[str, RateLimit] | None = None,
    trusted_proxy_hops: int = 0,
    sudo: SudoConfig | None = None,
    warn_on_memory_backend: bool = True,
)

Composition root: configure transports, mount routers, gate routes.

Construct one per auth surface. It owns the user repository, the shared AuthRuntime, the rate-limiter backend, and the assembled routers. Session auth is the default; add bearer/oauth/email by passing transports=, oauth=, email=.

Example

auth = CRUDAuth(session=get_session, user_model=User, SECRET_KEY="change-me")
app.include_router(auth.router)

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

Configure the auth surface.

Parameters:

Name	Type	Description	Default
`session`	`Callable[..., Any]`	FastAPI dependency that yields an `AsyncSession` (your `get_session`); every route and the `current_user` dependency acquire the DB through it.	required
`user_model`	`type[Any]`	Your SQLAlchemy user model (typically inheriting AuthUserMixin).	required
`SECRET_KEY`	`str`	Secret used to sign session/JWT and email tokens.	required
`transports`	`Sequence[Transport] \| None`	Ordered auth channels to enable; defaults to a single SessionTransport. Order is the first-wins precedence.	`None`
`column_map`	`dict[str, str] \| None`	Maps crudauth logical field names to your model's actual column names when they differ (e.g. `{"hashed_password": "pw_hash"}`).	`None`
`oauth`	`dict[str, Any] \| None`	`{provider_name: OAuthCredentials}` to enable OAuth login; requires `redirect_base_url` and a session transport.	`None`
`email`	`Any`	An EmailConfig to enable verify/reset/change flows; `None` disables them.	`None`
`hooks`	`AuthHooks \| None`	Lifecycle callbacks (AuthHooks).	`None`
`redirect_base_url`	`str \| None`	Public base URL used to build OAuth redirect URIs and the post-login redirect default.	`None`
`algorithm`	`str`	JWT signing algorithm (default `"HS256"`).	`DEFAULT_ALGORITHM`
`cookies`	`CookieConfig \| None`	App-wide CookieConfig (`secure` / `samesite` / `path`); transports may override per-instance.	`None`
`register_schema`	`type[BaseModel] \| None`	Custom Pydantic body for `/register`. By default only `email`/`username` are persisted; any other field is dropped unless its name is listed in `register_extra_fields`.	`None`
`register_extra_fields`	`set[str] \| None`	App-defined model columns that `/register` is allowed to set (e.g. `{"full_name", "locale"}`). Registration is an allowlist: without opting a column in here it is dropped, so adding a column to your model never silently becomes settable at signup. crudauth's privileged fields (`is_superuser`, `email_verified`, ...) can never be opted in.	`None`
`rate_limiter`	`RateLimiterBackend \| None`	Backend for lockout/throttles; defaults to an in-process MemoryRateLimiterBackend. Use `redis_rate_limiter(...)` in production.	`None`
`rate_limits`	`dict[str, RateLimit] \| None`	Per-action overrides merged over :data:`~crudauth.ratelimit.DEFAULT_RATE_LIMITS`.	`None`
`trusted_proxy_hops`	`int`	Number of trusted reverse proxies in front of the app. `0` (default) ignores `X-Forwarded-For` and keys per-IP rate limits / lockout on the socket peer; set to the count of proxies you control (e.g. `1` behind a single nginx/Caddy) so the real client IP is read without trusting attacker-supplied header values. See get_client_ip.	`0`
`sudo`	`SudoConfig \| None`	Enable sudo mode (short-lived re-authentication for sensitive actions) with this SudoConfig. Requires a session transport - elevation is stamped on the server-side session. Exposes `auth.sudo` and `auth.require_sudo()`.	`None`
`warn_on_memory_backend`	`bool`	Log a startup warning when an in-memory backend is active (the zero-config default). In-memory state is per-process, so under multiple workers it silently breaks; set `False` to silence once you've accepted that (e.g. single-worker dev).	`True`

Raises:

Type	Description
`ValueError`	If `SECRET_KEY` is empty; if `oauth` or `sudo` is set without a session transport (and `oauth` also needs `redirect_base_url`); or if a configured OAuth provider has no `{provider}_id` column on the user model.

sessions `property` ¶

sessions

The SessionManager of the configured session transport.

router `property` ¶

router: APIRouter

The full router to mount: shared (/register, /me) plus every transport's routes, plus OAuth and email routes when configured.

Returns:

Type	Description
`APIRouter`	An `APIRouter` to pass to `app.include_router`.

Example

app.include_router(auth.router)

session_router `property` ¶

session_router: APIRouter

Only the session transport's routes (/login, /logout).

Raises:

Type	Description
`RuntimeError`	If no SessionTransport is configured.

bearer_router `property` ¶

bearer_router: APIRouter

Only the bearer transport's routes (/token, /refresh).

Raises:

Type	Description
`RuntimeError`	If no BearerTransport is configured.

current_user ¶

current_user(
    *,
    optional: bool = False,
    superuser: bool = False,
    verified: bool = False,
    scopes: list[str] | None = None,
    transport: str | list[str] | None = None,
    check: Callable[[Principal], Any] | None = None,
) -> Callable[..., Any]

Build a FastAPI dependency that authenticates and authorizes a request.

Every gate is a keyword: optional, superuser, verified, scopes, transport (narrow to one/some transports), and check.

Note

check is a predicate (sync or async) run last on the resolved principal. Returning False denies the request with 403. To deny with a custom status/message, raise your own exception from inside check. Returning None (or anything that isn't False) allows - so both styles work: a boolean predicate (check=lambda p: p.is_superuser) and a raise-to-deny callback that simply returns nothing on success.

Note

Transports are tried in order, first credential wins. A transport returns None when its credential is absent (move to the next), but RAISES for a present-but-invalid one (e.g. a session cookie that fails the CSRF header check on a mutation). That hard-fail propagates even under optional=True - a tampered credential is an attack signal, not "treat me as anonymous".

Returns:

Type	Description
`Callable[..., Any]`	An async dependency yielding the Principal (or `None` when
`Callable[..., Any]`	`optional` and no credential is present).

Example

@app.get("/admin")
async def admin(_: Principal = Depends(auth.current_user(superuser=True))):
    ...

require_sudo ¶

require_sudo() -> Callable[..., Any]

Build a dependency that requires a current sudo elevation.

Authenticates like current_user (reusing the per-request principal cache) and then demands an unexpired sudo stamp, raising 403 otherwise. Compose it with current_user gates on the same route to also enforce identity/role:

Example

@app.post("/account/close")
async def close(
    user: Principal = Depends(auth.current_user(superuser=True)),
    _: Principal = Depends(auth.require_sudo()),
):
    ...

Raises:

Type	Description
`RuntimeError`	If sudo isn't configured (pass `sudo=SudoConfig()`).

rate_limit ¶

rate_limit(
    action: str,
    limit: RateLimit | None = None,
    *,
    key: KeyBy | Callable[[Request], str] = IP,
) -> Callable[..., Any]

Build a FastAPI dependency that throttles an endpoint.

Resolves the limit (explicit limit → rate_limits= override → :data:~crudauth.ratelimit.DEFAULT_RATE_LIMITS), keys by IP, user, or a custom function, writes X-RateLimit-* headers, and raises RateLimitException (429) when the caller exceeds the window.

Example

@app.post("/contact", dependencies=[Depends(auth.rate_limit("contact", RateLimit(5, 60)))])
async def contact(...): ...

initialize `async` ¶

initialize() -> None

Open storage/limiter connections; call from your app's lifespan startup.

Idempotent per component. Required for server-side backends (redis); a no-op for the in-memory defaults.

Example

@asynccontextmanager
async def lifespan(app):
    await auth.initialize()
    yield
    await auth.shutdown()

shutdown `async` ¶

shutdown() -> None

Close connections. Call in lifespan teardown.

CRUDAuth¶

crudauth.CRUDAuth ¶

sessions property ¶

router property ¶

session_router property ¶

bearer_router property ¶

current_user ¶

require_sudo ¶

rate_limit ¶

initialize async ¶

shutdown async ¶

sessions `property` ¶

router `property` ¶

session_router `property` ¶

bearer_router `property` ¶

initialize `async` ¶

shutdown `async` ¶