app.middlewares.auth_context module¶
- class app.middlewares.auth_context.AuthContext(api_key, oauth_user_id, is_admin, token_data)[source]¶
Bases:
objectPer-request auth context resolved by get_auth_context.
- Parameters:
api_key (str | None)
oauth_user_id (str | None)
is_admin (bool)
token_data (Dict[str, Any] | None)
- class app.middlewares.auth_context.AuditLogger(module, service_log, auth)[source]¶
Bases:
objectPer-request audit logger bound to a module name and AuthContext.
- Parameters:
module (str)
service_log (LogsService)
auth (AuthContext)
- async info(message, details=None)[source]¶
Write an
INFO-level audit log row for this request.- Parameters:
message (str) – Human-readable description of the event.
details (Optional[Dict[str, Any]]) – Structured context to attach; defaults to an empty dict.
- Return type:
None
- async app.middlewares.auth_context.get_auth_context(request, token=Depends(CustomOAuth2AuthorizationCodeBearer), api_key_header=Depends(get_api_key_header), service_oauth=Depends(Provide), service_log=Depends(Provide))[source]¶
- Resolves the per-request auth context from Authorization: Bearer
and/or X-API-Key.
- When a bearer token is present:
validates it via valid_access_token,
extracts sub and admin role,
bootstraps a Keycloak OAuth user record if missing (and writes a single auth / OAuth user bootstrapped audit entry).
- Parameters:
request (Request)
token (str | None)
service_oauth (OAuthUsersService)
service_log (LogsService)
- Return type:
- app.middlewares.auth_context.audit_log(module)[source]¶
Build a FastAPI dependency that yields a request-scoped
AuditLogger.The returned dependency resolves the request’s
AuthContextand binds it, together withmodule, to a freshAuditLogger.- Parameters:
module (str) – Logical module name stamped on every emitted log row.
- Returns:
Callable – A FastAPI dependency returning an
AuditLogger.- Return type:
Callable
Example
@router.get("/foo", dependencies=[Depends(auth_api_key_or_oauth2)]) async def foo(audit: AuditLogger = Depends(audit_log("users"))): await audit.info("Foo invoked", {...}) try: ... except Exception as exc: await audit.error("Foo failed", {"error": str(exc)}) raise