app.middlewares.auth_context module

class app.middlewares.auth_context.AuthContext(api_key, oauth_user_id, is_admin, token_data)[source]

Bases: object

Per-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: object

Per-request audit logger bound to a module name and AuthContext.

Parameters:
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 success(message, details=None)[source]

Write a SUCCESS-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 error(message, details=None)[source]

Write an ERROR-level audit log row for this request.

Parameters:
  • message (str) – Human-readable description of the failure.

  • 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:
Return type:

AuthContext

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 AuthContext and binds it, together with module, to a fresh AuditLogger.

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