fastapi-m8

FastAPI application framework for building consumer microservices that integrate with fa-auth-m8. It wires authentication, CORS, health checks, observability, and database lifecycle into a single create_app() call, removing ~90 % of the setup boilerplate from every consumer service.

Summary

fastapi-m8 is a thin application factory layer that sits on top of FastAPI and auth-sdk-m8. You bring a settings object, a router, and optional health checks; the framework wires the rest.

What it provides:

Capability	How
JWT validation	`build_auth_deps()` + `auth-sdk-m8` validator
Role-based access control	`AuthDeps.get_current_active_admin / _superuser`
Token revocation (stateful mode)	`RemoteRevocationClient` → `fa-auth-m8` private API
CORS	Auto-wired from `settings.ALLOWED_ORIGINS`
Metrics middleware	Optional; toggled via `METRICS_ENABLED`
Health endpoint	`GET {API_PREFIX}/health/` with optional detail gating
Service meta + liveness	Auto-mounted `GET {API_PREFIX}/meta` + `GET /ping` (fail-closed at boot)
Database lifecycle	`create_db_engine()` wrapping SQLAlchemy
Startup validation	`startup_validators` list runs before app signals ready
Lifespan management	Auth teardown + DB pool dispose on shutdown

What it is NOT:

Not an auth issuer — that role belongs to fa-auth-m8.
Not a business logic framework — it only provides plumbing and dependency injection.

Architecture & Package Roles

┌───────────────────────────────────────────────────────────────┐
│  Your consumer service  (uses fastapi-m8)                     │
│                                                               │
│  create_app(settings, router,                                 │
│      health=HealthConfig(checks=[...]),                       │
│      lifecycle=AppLifecycle(auth_deps=auth, ...))            │
│  ├─ ConsumerServiceSettings ← auth-sdk-m8 CommonSettings     │
│  ├─ build_auth_deps(settings)                                 │
│  │   ├─ TokenValidator (local JWT check, auth-sdk-m8)        │
│  │   └─ RemoteRevocationClient (stateful only, HTTP)          │
│  └─ auto-wired: CORS · metrics · health · lifespan           │
└────────────────────────┬──────────────────────────────────────┘
                         │ Authorization: Bearer <JWT>
                         │ (stateful) POST /private/v1/jti-status
                         ▼
┌───────────────────────────────────────────────────────────────┐
│  fa-auth-m8  (auth_user_service)                              │
│                                                               │
│  POST /user/login/access-token   → issues JWT pair           │
│  POST /user/login/refresh-token/ → rotates tokens            │
│  POST /private/v1/jti-status     → revocation check          │
│                                                               │
│  Backing stores: MySQL / PostgreSQL · Redis                   │
└───────────────────────────────────────────────────────────────┘

Three packages, three responsibilities:

Package	Role
`fa-auth-m8`	Issues and revokes JWT tokens, manages users and sessions
`auth-sdk-m8`	Shared schemas, JWT validation, settings base classes (read-only)
`fastapi-m8`	Wires `auth-sdk-m8` into a FastAPI consumer service

Installation

# Minimal (no database)
pip install fastapi-m8

# With PostgreSQL
pip install "fastapi-m8[postgres]"

# With MySQL
pip install "fastapi-m8[mysql]"

# With database (driver-agnostic, you choose the driver)
pip install "fastapi-m8[db]"

# Everything
pip install "fastapi-m8[all]"

Runtime requirements: Python 3.11+

Quick Start

1 — Settings

# app/core/config.py
from pathlib import Path
from pydantic_settings import SettingsConfigDict
from fastapi_m8 import ConsumerServiceSettings

class Settings(ConsumerServiceSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

settings = Settings()

2 — Auth & DB dependencies

# app/core/deps.py
from fastapi_m8 import build_auth_deps, create_db_engine
from app.core.config import settings

auth = build_auth_deps(settings)
engine = create_db_engine(settings)

3 — Routes

# app/api/items.py
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlmodel import Session
from app.core.deps import auth, engine

router = APIRouter(prefix="/items", tags=["items"])
SessionDep = Annotated[Session, Depends(engine.session_dep)]

@router.get("/")
async def list_items(user: auth.CurrentUser, session: SessionDep):
    return {"owner": user.email}

4 — App factory

# app/main.py
from fastapi import APIRouter
from fastapi_m8 import (
    AppLifecycle, HealthConfig, create_app, HealthCheckResult, HealthStatus,
)
from sqlmodel import select
from app.core.config import settings
from app.core.deps import auth, engine
from app.api.items import router as items_router

async def check_db() -> HealthCheckResult:
    try:
        with engine.session() as s:
            s.exec(select(1))
        return HealthCheckResult.from_bool("database", True)
    except Exception as exc:
        return HealthCheckResult(name="database", status=HealthStatus.FAIL, error=str(exc))

api_router = APIRouter()
api_router.include_router(items_router)

app = create_app(
    settings,
    api_router,
    service_name="Item Service",
    service_version="1.0.0",
    health=HealthConfig(checks=[check_db]),
    lifecycle=AppLifecycle(auth_deps=auth, db_engine=engine),
)

5 — `.env`

DOMAIN=localhost
ENVIRONMENT=local
PROJECT_NAME=Item Service
STACK_NAME=local
API_PREFIX=/api
AUTH_PREFIX=/auth
BACKEND_HOST=http://localhost:8000
FRONTEND_HOST=http://localhost:3000
BACKEND_CORS_ORIGINS=http://localhost:3000

# Token signing — must match fa-auth-m8 (secure-by-default: RS256 + JWKS)
# ACCESS_TOKEN_ALGORITHM defaults to RS256; supply the issuer's public key.
ACCESS_PUBLIC_KEY_FILE=/opt/keys/access_public.pem
# JWKS_URI=https://auth.example.com/.well-known/jwks.json   # zero-downtime rotation
REFRESH_SECRET_KEY=change-me-refresh-32-chars-min

# Strict iss/aud binding is ON by default — both are required at boot.
TOKEN_ISSUER=https://auth.example.com
TOKEN_AUDIENCE=item-service
# Opt out for single-service/dev deployments (then HS256 + ACCESS_SECRET_KEY is enough):
# TOKEN_STRICT_VALIDATION=false
# ACCESS_TOKEN_ALGORITHM=HS256
# ACCESS_SECRET_KEY=change-me-32-chars-minimum

# Event signing is ON by default — SSE payloads from fa-auth are HMAC-signed.
# DEV-ONLY placeholder — replace with the same value set on fa-auth in staging/production.
EVENT_SIGNING_KEY=DEV-ONLY-do-not-use-event-signing-key-Aa1!
# EVENT_SIGNING_ENABLED=false   # opt out only if signing is also disabled on fa-auth

TOKEN_MODE=stateless
AUTH_SERVICE_ROLE=consumer

# Service/contract metadata — served at {API_PREFIX}/meta for client compat
# checks. REQUIRED: the app fails closed at boot without them. (/ping needs none.)
SERVICE_VERSION=1.0.0
CONTRACT_VERSION=1.0
CONTRACT_RANGE=>=1.0.0 <2.0.0
# API_VERSION=v1                 # default
# CONTRACT_NAME=item-service     # defaults to PROJECT_NAME

# Host validation — set in production to prevent host-header injection
# ALLOWED_HOSTS=api.example.com

# Docs gating — docs are auto-disabled in production (ENVIRONMENT=production)
# unless SERVE_DOCS_IN_PRODUCTION=true (opt-in for public APIs)
# SERVE_DOCS_IN_PRODUCTION=false

# Database
DB_HOST=localhost
DB_PORT=5432
DB_DATABASE=items_db
DB_USER=app_user
DB_PASSWORD=secret

Run with:

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

Configuration Reference

All settings inherit from auth-sdk-m8's CommonSettings. Every field maps 1:1 to an environment variable.

Core / Network

Variable	Required	Default	Description
`DOMAIN`	Yes	—	Public domain, e.g. `localhost`
`ENVIRONMENT`	Yes	—	`local` \| `development` \| `staging` \| `production`
`PROJECT_NAME`	Yes	—	Human-readable service name (shown in docs)
`STACK_NAME`	Yes	—	Docker Compose stack slug
`API_PREFIX`	Yes	—	URL prefix for this service's routes, e.g. `/api`
`AUTH_PREFIX`	No	`/auth`	Auth endpoint prefix (consumer services)
`BACKEND_HOST`	Yes	—	Full backend URL, e.g. `http://127.0.0.1:8000`
`FRONTEND_HOST`	Yes	—	Full frontend URL
`BACKEND_CORS_ORIGINS`	Yes	—	Comma-separated allowed origins

Service Metadata (`/meta`)

create_app auto-mounts the shared service triad from auth-sdk-m8: GET {API_PREFIX}/meta (cacheable service/version/contract identity, read by clients pre-auth to assert compatibility) and a prefix-independent GET /ping (dependency-free liveness → {"status": "ok"}). The /meta values are sourced from these settings, so a consumer fails closed at boot if it doesn't declare its identity. Keep both separate from a dependency-aware /health readiness probe.

Variable	Required	Default	Description
`SERVICE_VERSION`	Yes	—	Service package version, e.g. `1.0.0`
`CONTRACT_VERSION`	Yes	—	Contract version, e.g. `1.0`
`CONTRACT_RANGE`	Yes	—	Compatible contract semver range, e.g. `>=1.0.0 <2.0.0`
`API_VERSION`	No	`v1`	Public API version label
`CONTRACT_NAME`	No	`PROJECT_NAME`	Contract name (defaults to the service name)

Tokens & Cryptography

Variable	Required	Default	Description
`TOKEN_MODE`	No	`stateful`	`stateless` \| `hybrid` \| `stateful` (see Token Modes)
`AUTH_SERVICE_ROLE`	No	`issuer`	Set to `consumer` in all consumer services
`ACCESS_TOKEN_ALGORITHM`	No	`RS256`	`RS256` (default, asymmetric/JWKS) \| `ES256` \| `HS256` (opt-in shared secret)
`ACCESS_PUBLIC_KEY_FILE`	RS256/ES256	—	Path to PEM public key file (consumer validation)
`JWKS_URI`	RS256/ES256 alt	—	JWKS endpoint URL (auto-fetches and caches public keys; zero-downtime rotation)
`JWKS_CACHE_TTL_SECONDS`	No	`300`	JWKS key cache TTL in seconds
`ACCESS_SECRET_KEY`	HS256 only	—	Shared symmetric signing key (≥ 32 chars) — required only when opting into HS256
`REFRESH_SECRET_KEY`	Yes	—	Refresh token signing key (always HS256, internal)
`ACCESS_TOKEN_EXPIRE_MINUTES`	No	`30`	Access token lifetime
`REFRESH_TOKEN_EXPIRE_MINUTES`	No	`120`	Refresh token lifetime
`TOKEN_STRICT_VALIDATION`	No	`true`	Secure-by-default: enforce `iss`/`aud` binding; requires `TOKEN_ISSUER` + `TOKEN_AUDIENCE` at boot. Set `false` for single-service/dev.
`TOKEN_ISSUER`	Yes¹	—	Expected `iss` claim. Required at boot under strict validation.
`TOKEN_AUDIENCE`	Yes¹	—	Expected `aud` claim (this service). Required at boot under strict validation.
`EVENT_SIGNING_ENABLED`	No	`true`	Secure-by-default: HMAC-sign SSE event payloads. Set `false` to disable.
`EVENT_SIGNING_KEY`	Yes²	—	Shared HMAC secret for SSE payload verification. Must match fa-auth. Required at boot unless `EVENT_SIGNING_ENABLED=false`.

¹ Required unless TOKEN_STRICT_VALIDATION=false. ² Required unless EVENT_SIGNING_ENABLED=false.

Secure-by-default (auth-sdk-m8 ≥ 1.0.0): access tokens default to RS256 and validation enforces iss/aud binding, so a factory-built app rejects wrong-audience / wrong-issuer tokens out of the box. Operators who need shared-secret signing opt back in with ACCESS_TOKEN_ALGORITHM=HS256 + ACCESS_SECRET_KEY; those without cross-service boundaries relax binding with TOKEN_STRICT_VALIDATION=false.

Stateful Mode (consumer → auth service)

Required only when TOKEN_MODE=stateful and AUTH_SERVICE_ROLE=consumer.

Variable	Required	Default	Description
`INTROSPECTION_URL`	Yes	—	`POST` endpoint on auth service for JTI revocation checks, e.g. `http://auth_user_service:8000/user/private/v1/jti-status`
`PRIVATE_API_SECRET`	Yes	—	Shared secret for `X-Internal-Token` header (must match auth service)

Auth Event Stream (fa-auth SSE bridge)

An optional, best-effort accelerator for cache eviction. fa-auth-m8 bridges its own auth-state events (session-revoked, user-deleted) to consumers over an authenticated Server-Sent Events stream on the existing private API — the same trust channel (INTROSPECTION_URL + PRIVATE_API_SECRET) already used for JTI checks. No second Redis, no broker: consumers speak HTTPS to fa-auth, which they already do.

The stream is not the revocation authority — the JTI blacklist behind INTROSPECTION_URL is. A consumer that misses every event is still correct, just slower to evict caches. Stream loss is non-fatal; the service keeps running on the HTTP authority path alone.

Wire it in your lifespan with build_event_stream_client, which constructs the SDK's AuthEventStreamClient from your settings (no SDK internals needed):

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi_m8 import build_event_stream_client, AuthStreamEvent


async def on_auth_event(event: AuthStreamEvent) -> None:
    # session-revoked / user-deleted → evict the affected entry from local caches.
    ...


async def on_gap() -> None:
    # Unresumable stream (fa-auth restarted / buffer evicted) → flush ALL caches.
    ...


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = build_event_stream_client(
        settings,
        on_event=on_auth_event,
        on_gap=on_gap,
    )
    client.start()
    try:
        yield
    finally:
        await client.stop()

The client verifies every payload's HMAC signature with EVENT_SIGNING_KEY (must match fa-auth), auto-reconnects with jittered backoff, resumes via Last-Event-ID, and never raises into the host app. Requires TOKEN_MODE=stateful so INTROSPECTION_URL and PRIVATE_API_SECRET are present. Behind a reverse proxy, disable response buffering on the stream endpoint so events and heartbeats pass through promptly.

Variable	Required	Default	Description
`EVENT_STREAM_CONNECT_TIMEOUT`	No	`5`	Seconds to wait for the initial SSE connection (factory arg).
`EVENT_STREAM_READ_TIMEOUT`	No	`60`	Seconds to wait between SSE frames; keep above fa-auth's heartbeat interval (default 15 s).

Database

Variable	Required	Default	Description
`SELECTED_DB`	No	`Mysql`	`Mysql` \| `Postgres`
`DB_HOST`	Yes	—	Database host
`DB_PORT`	Yes	—	Database port
`DB_DATABASE`	Yes	—	Database name
`DB_USER`	Yes	—	Database user
`DB_PASSWORD`	Yes	—	Database password
`TABLES_PREFIX`	No	`app`	Table name prefix

Redis

Required when TOKEN_MODE=stateful or hybrid on the issuer side. Consumer services do not connect to Redis directly.

Variable	Description
`REDIS_HOST`	Redis host
`REDIS_PORT`	Redis port
`REDIS_USER`	Redis username
`REDIS_PASSWORD`	Redis password
`REDIS_SSL`	Enable TLS (`true`/`false`, default `false`)

Observability

Variable	Default	Description
`METRICS_ENABLED`	`false`	Enable Prometheus metrics middleware
`METRICS_GROUPS`	—	Comma-separated groups: `traffic`, `performance`, `reliability`, `health`, `auth`, or `all`

OpenAPI / Docs

Variable	Default	Description
`SET_OPEN_API`	`true`	Expose `/openapi.json` (gated off in production unless `SERVE_DOCS_IN_PRODUCTION=true`)
`SET_DOCS`	`true`	Expose Swagger UI (gated off in production unless `SERVE_DOCS_IN_PRODUCTION=true`)
`SET_REDOC`	`true`	Expose ReDoc (gated off in production unless `SERVE_DOCS_IN_PRODUCTION=true`)
`SERVE_DOCS_IN_PRODUCTION`	`false`	Set `true` to explicitly re-enable docs in production (e.g. public/open-source APIs). Requires `auth-sdk-m8>=0.7.3`.

Production docs gating (secure-by-default): when ENVIRONMENT=production (or STRICT_PRODUCTION_MODE=true), all three doc endpoints are disabled regardless of the SET_* flags, unless SERVE_DOCS_IN_PRODUCTION=true is set. Non-production environments are unaffected.

Security / Host Validation

Variable	Default	Description
`ALLOWED_HOSTS`	`` (empty)	Comma-separated list of allowed `Host` headers, e.g. `api.example.com,localhost`. Empty = no restriction (permissive, suitable for dev). Set in production to prevent host-header injection.

TrustedHostMiddleware: when ALLOWED_HOSTS is non-empty, fastapi-m8 registers Starlette's TrustedHostMiddleware. Requests with a Host header not in the list are rejected with HTTP 400. testserver is automatically added in non-production so pytest's TestClient works without extra configuration.

Response Security Headers

create_app wires the response-hardening layer from auth-sdk-m8 (auth_sdk_m8.security.headers.add_security_headers_middleware, tiered since auth-sdk-m8 ≥ 1.2.1). Headers are applied in three tiers — the browser-persisted HSTS and CSP are now express opt-in rather than inferred from the production gate:

Tier	Headers	When applied
Always-on	`X-Content-Type-Options: nosniff`, `X-Frame-Options: DENY`	Every environment (when `SECURITY_HEADERS_ENABLED`) — safe for Swagger/ReDoc/HMR
Production-gated	`Referrer-Policy`, `Permissions-Policy`	`ENVIRONMENT=production` or `STRICT_PRODUCTION_MODE=true`
Express opt-in	`Strict-Transport-Security`, `Content-Security-Policy`	`HSTS_ENABLED` / `CONTENT_SECURITY_POLICY_ENABLED` — never on `local`, even when opted in

The opt-in tier is decoupled from the production gate, so a TLS-terminated staging stack can enable HSTS/CSP without masquerading as production. They are hard-blocked on ENVIRONMENT=local because HSTS is browser-persisted and would poison the localhost HTTPS cache for HSTS_MAX_AGE seconds (a real risk when a production-configured build is run locally before deploy).

Variable	Default	Description
`SECURITY_HEADERS_ENABLED`	`true`	Master switch; set `false` to suppress every tier
`HSTS_ENABLED`	`false`	Opt in to `Strict-Transport-Security` (never emitted on `local`)
`HSTS_MAX_AGE`	`31536000`	HSTS max-age in seconds; `0` drops the HSTS header even when enabled
`HSTS_INCLUDE_SUBDOMAINS`	`true`	Append `includeSubDomains` to HSTS
`CONTENT_SECURITY_POLICY_ENABLED`	`false`	Opt in to `Content-Security-Policy` (never emitted on `local`)
`CONTENT_SECURITY_POLICY`	(hardened default)	Override the emitted `Content-Security-Policy`
`REFERRER_POLICY`	`strict-origin-when-cross-origin`	`Referrer-Policy` value (production-gated)
`PERMISSIONS_POLICY`	(restrictive default)	`Permissions-Policy` value (production-gated)

Behaviour change (since 1.5.0 / auth-sdk-m8 1.2.1): HSTS and CSP were emitted automatically in production before; they are now off until explicitly enabled. To restore the previous behaviour set HSTS_ENABLED=true and CONTENT_SECURITY_POLICY_ENABLED=true.

These knobs are inherited from CommonSettings; consumer services do not redeclare them. The same layer is shared by fa-auth-m8 and every consumer.

API Reference

`create_app()`

from fastapi_m8 import create_app, HealthConfig, AppLifecycle

app = create_app(
    settings: ConsumerServiceSettings,
    router: APIRouter,
    *,
    service_name: str | None = None,
    service_version: str | None = None,
    health: HealthConfig | None = None,
    lifecycle: AppLifecycle | None = None,
) -> FastAPI

Parameters:

Parameter	Description
`settings`	Service settings object (subclass of `ConsumerServiceSettings`)
`router`	Your domain `APIRouter` (all routes are mounted under this)
`service_name`	Overrides `settings.PROJECT_NAME` in health detail response
`service_version`	Reported in health detail response
`health`	`HealthConfig` dataclass (checks, timeout, policy, detail options, cache TTL)
`lifecycle`	`AppLifecycle` dataclass (auth_deps, db_engine, startup_validators, configure, lifespan_extras)

HealthConfig fields:

Field	Default	Description
`checks`	`None`	List of async callables returning `HealthCheckResult`
`timeout`	`0.5`	Per-check timeout in seconds
`policy`	`LENIENT`	`LENIENT` or `STRICT` — controls when 503 is returned
`detail_public`	`False`	Expose per-check detail without authentication
`detail_authorizer`	`None`	Custom async callable; receives `Request`, returns `bool`
`cache_ttl`	`2.0`	Seconds to cache health results

AppLifecycle fields:

Field	Default	Description
`auth_deps`	`None`	Output of `build_auth_deps()`. Closed on shutdown
`db_engine`	`None`	Output of `create_db_engine()`. Disposed on shutdown
`startup_validators`	`None`	Async callables run before app signals ready; raise to abort
`configure`	`None`	Callback receiving the raw `FastAPI` instance for custom middleware
`lifespan_extras`	`None`	Async context manager run inside the managed lifespan

Lifespan sequence:

Run lifecycle.startup_validators — raise any exception to prevent ready signal.
Enter lifecycle.lifespan_extras context (if provided).
Set app.state.service_ready = True.
(app serves traffic)
Exit lifecycle.lifespan_extras.
Call lifecycle.auth_deps.close() (closes revocation HTTP client).
Call lifecycle.db_engine.dispose() (closes connection pool).

`ConsumerServiceSettings`

from fastapi_m8 import ConsumerServiceSettings

Base settings class. Subclass it and configure model_config for your .env file.

from pydantic_settings import SettingsConfigDict
from fastapi_m8 import ConsumerServiceSettings

class Settings(ConsumerServiceSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

settings = Settings()

# Useful computed properties (inherited from auth-sdk-m8)
settings.is_stateless        # bool
settings.is_stateful         # bool
settings.ALLOWED_ORIGINS     # list[str] — derived from BACKEND_CORS_ORIGINS
settings.SQLALCHEMY_DATABASE_URI  # str — assembled from DB_* fields

`build_auth_deps()`

from fastapi_m8 import build_auth_deps, AuthDeps

auth: AuthDeps = build_auth_deps(settings)

Returns a frozen dataclass with everything needed for route protection.

Field	Type	Description
`auth.CurrentUser`	`Annotated[UserModel, Depends(...)]`	Inject authenticated user into routes
`auth.get_current_user`	`async Callable`	FastAPI dependency; validates JWT, checks revocation
`auth.get_current_active_admin`	`Callable`	Raises 403 unless user has ADMIN or SUPERADMIN role
`auth.get_current_active_superuser`	`Callable`	Raises 403 unless user has SUPERADMIN role and `is_superuser=True`
`auth.revocation_client`	`RemoteRevocationClient \| None`	Present only in stateful mode

UserModel fields available in routes:

Field	Type	Description
`id`	`uuid.UUID`	User primary key
`email`	`str`	User email
`full_name`	`str \| None`	Display name
`role`	`RoleType`	`USER` \| `READER` \| `WRITER` \| `ADMIN` \| `SUPERADMIN`
`is_active`	`bool`	Account active flag
`is_superuser`	`bool`	Superuser flag
`email_verified`	`bool`	Email verification status

`create_db_engine()`

from fastapi_m8 import create_db_engine, DbEngine

engine: DbEngine = create_db_engine(settings)

Wraps SQLAlchemy engine assembled from settings.SQLALCHEMY_DATABASE_URI.

Method	Description
`engine.session()`	Context manager yielding a `Session`
`engine.session_dep()`	FastAPI dependency (use with `Depends`)
`engine.dispose()`	Closes connection pool (called automatically on shutdown)

from typing import Annotated
from fastapi import Depends
from sqlmodel import Session

SessionDep = Annotated[Session, Depends(engine.session_dep)]

@router.post("/items")
async def create_item(session: SessionDep, item: ItemCreate):
    session.add(Item.model_validate(item))
    session.commit()

Health Checks

Implement the HealthCheck protocol — any async callable returning HealthCheckResult.

from fastapi_m8 import HealthCheck, HealthCheckResult, HealthStatus

# Function-based
async def check_database() -> HealthCheckResult:
    try:
        with engine.session() as s:
            s.exec(select(1))
        return HealthCheckResult.from_bool("database", True)
    except Exception as exc:
        return HealthCheckResult(name="database", status=HealthStatus.FAIL, error=str(exc))

# Class-based (useful when state is needed)
class RedisCheck:
    def __init__(self, client):
        self._client = client

    async def __call__(self) -> HealthCheckResult:
        try:
            await self._client.ping()
            return HealthCheckResult.from_bool("redis", True)
        except Exception as exc:
            return HealthCheckResult(name="redis", status=HealthStatus.FAIL, error=str(exc))

HealthCheckResult fields:

Field	Type	Description
`name`	`str`	Check identifier
`status`	`HealthStatus`	`ok` \| `degraded` \| `fail` \| `unknown`
`latency_ms`	`float \| None`	Auto-populated by the health subsystem
`error`	`str \| None`	Error message (credentials automatically scrubbed)
`meta`	`dict \| None`	Arbitrary metadata (sensitive keys auto-redacted)
`ok`	`bool`	Computed: `True` when status is `ok`

HealthAggregatePolicy:

Value	HTTP 503 when
`LENIENT` (default)	Any check is `fail`
`STRICT`	Any check is `fail` or `unknown`

Authentication

Token Modes

Configured via TOKEN_MODE on both the auth service and all consumer services. The value must match across the stack.

Mode	Access token revocation	Requires Redis (issuer)	Google OAuth
`stateless`	None (waits for expiry)	No	No
`hybrid`	None for access; refresh is allowlisted	Yes	Yes
`stateful`	Immediate, via JTI introspection	Yes	Yes

Stateless — maximum scalability, simplest setup. Logout does not invalidate in-flight access tokens; they expire naturally.

Stateful — highest security. On each request a consumer performs an HTTP call to fa-auth-m8 to verify the JWT's JTI has not been revoked. Requires INTROSPECTION_URL and PRIVATE_API_SECRET in consumer settings.

Algorithm options:

Algorithm	Key config	Use case
`RS256` (default)	`ACCESS_PUBLIC_KEY_FILE` or `JWKS_URI`	Multi-service; consumers need only the public key
`ES256`	`ACCESS_PUBLIC_KEY_FILE` or `JWKS_URI`	Same as RS256, smaller keys
`HS256` (opt-in)	`ACCESS_SECRET_KEY` (symmetric, shared)	Simple single-service or trusted internal network

Since auth-sdk-m8 ≥ 1.0.0, RS256 is the default; choose HS256 explicitly via ACCESS_TOKEN_ALGORITHM=HS256. With JWKS_URI set, the consumer fetches and caches the public key automatically, refreshing on unknown kid headers.

Role System

Roles are hierarchical. Higher roles include all permissions of lower roles.

SUPERADMIN > ADMIN > WRITER > READER > USER

Role	Typical use
`SUPERADMIN`	Full platform access, user management
`ADMIN`	Administrative operations within a service
`WRITER`	Create and update resources
`READER`	Read-only access
`USER`	Base authenticated user

Protecting Routes

from fastapi import APIRouter, Depends
from typing import Annotated
from app.core.deps import auth

router = APIRouter()

# Any authenticated user
@router.get("/profile")
async def get_profile(user: auth.CurrentUser):
    return {"id": user.id, "email": user.email, "role": user.role}

# ADMIN or SUPERADMIN
@router.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: Annotated[UserModel, Depends(auth.get_current_active_admin)],
):
    ...

# SUPERADMIN only
@router.post("/admin/bootstrap")
async def bootstrap(
    su: Annotated[UserModel, Depends(auth.get_current_active_superuser)],
):
    ...

Unauthorized requests receive:

401 Unauthorized — missing or invalid token
403 Forbidden — valid token but insufficient role

Health Endpoint

Mounted automatically at GET {API_PREFIX}/health/ (e.g. /api/health/).

Before app is ready (during startup validators):

HTTP 503
{"status": "initializing", "ready": false}

After ready — public response:

HTTP 200   (or 503 if any check is "fail")
{"status": "ok"}

After ready — authorized response (with X-Internal-Token header or custom authorizer):

HTTP 200
{
  "status": "ok",
  "checks": [
    {"name": "database", "status": "ok", "latency_ms": 3.2, "error": null, "ok": true}
  ],
  "service": "Item Service",
  "version": "1.0.0",
  "fastapi_m8": "1.3.0",
  "auth_sdk_m8": "1.1.x"
}

Authorization options:

from fastapi_m8 import create_app, HealthConfig

# Option A — built-in X-Internal-Token (requires PRIVATE_API_SECRET in settings)
app = create_app(settings, router, health=HealthConfig(checks=[check_db]))
# Pass header: X-Internal-Token: <PRIVATE_API_SECRET>

# Option B — always public
app = create_app(settings, router, health=HealthConfig(checks=[check_db], detail_public=True))

# Option C — custom authorizer
async def is_internal(request: Request) -> bool:
    return request.client.host == "10.0.0.1"

app = create_app(
    settings,
    router,
    health=HealthConfig(checks=[check_db], detail_authorizer=is_internal),
)

Database Integration

Install the appropriate extra:

pip install "fastapi-m8[postgres]"   # psycopg2-binary
pip install "fastapi-m8[mysql]"      # pymysql

Configure in .env:

SELECTED_DB=Postgres
DB_HOST=db
DB_PORT=5432
DB_DATABASE=my_app
DB_USER=app
DB_PASSWORD=secret
TABLES_PREFIX=app

SQLALCHEMY_DATABASE_URI is assembled automatically. You can also set it directly to override the assembly.

Define models with TimestampMixin from auth-sdk-m8 (adds created_at / updated_at UTC columns):

import uuid
from sqlmodel import SQLModel, Field
from auth_sdk_m8.models.shared import TimestampMixin

class Item(TimestampMixin, SQLModel, table=True):
    __tablename__ = "app_items"

    id: int | None = Field(default=None, primary_key=True)
    name: str
    owner_id: uuid.UUID   # references the authenticated user's UUID id

Pre-Start Script

A CLI script that blocks until the database is reachable. Use it as a container init step to prevent your app from starting before the database is ready.

# Installed entry point
fastapi-m8-prestart

# Or directly
python -m fastapi_m8.scripts.pre_start

The script expects app.core.deps.engine to be a DbEngine instance. It retries SELECT 1 up to 300 times with 5-second intervals, then exits. If the module is not found or engine is not a DbEngine, it exits gracefully.

Dockerfile:

RUN pip install "fastapi-m8[postgres]"
CMD fastapi-m8-prestart && uvicorn app.main:app --host 0.0.0.0 --port 8000

Complete Example

my_service/
├── app/
│   ├── core/
│   │   ├── config.py      # Settings subclass
│   │   └── deps.py        # auth + engine singletons
│   ├── api/
│   │   └── items.py       # Domain router
│   └── main.py            # create_app() entry point
├── .env
└── pyproject.toml

app/core/config.py

from pydantic_settings import SettingsConfigDict
from fastapi_m8 import ConsumerServiceSettings

class Settings(ConsumerServiceSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

settings = Settings()

app/core/deps.py

from fastapi_m8 import build_auth_deps, create_db_engine
from app.core.config import settings

auth = build_auth_deps(settings)
engine = create_db_engine(settings)

app/api/items.py

from typing import Annotated
from fastapi import APIRouter, Depends
from sqlmodel import Session, select
from app.core.deps import auth, engine

router = APIRouter(prefix="/items", tags=["items"])
SessionDep = Annotated[Session, Depends(engine.session_dep)]

@router.get("/")
async def list_items(user: auth.CurrentUser, session: SessionDep):
    return {"owner": user.email}

@router.delete("/{item_id}/admin")
async def delete_item(
    item_id: int,
    admin: Annotated[object, Depends(auth.get_current_active_admin)],
    session: SessionDep,
):
    return {"deleted": item_id}

app/main.py

from fastapi import APIRouter
from sqlmodel import select
from fastapi_m8 import (
    AppLifecycle, HealthConfig, create_app, HealthCheckResult, HealthStatus,
)
from app.core.config import settings
from app.core.deps import auth, engine
from app.api.items import router as items_router

async def check_db() -> HealthCheckResult:
    try:
        with engine.session() as s:
            s.exec(select(1))
        return HealthCheckResult.from_bool("database", True)
    except Exception as exc:
        return HealthCheckResult(name="database", status=HealthStatus.FAIL, error=str(exc))

api_router = APIRouter()
api_router.include_router(items_router)

app = create_app(
    settings,
    api_router,
    service_name="Item Service",
    service_version="1.0.0",
    health=HealthConfig(checks=[check_db]),
    lifecycle=AppLifecycle(auth_deps=auth, db_engine=engine),
)

Testing

Override settings to avoid reading .env files in tests:

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from pydantic_settings import SettingsConfigDict
from fastapi_m8 import ConsumerServiceSettings, create_app

class TestSettings(ConsumerServiceSettings):
    model_config = SettingsConfigDict(env_file=None)  # no file — all from kwargs

@pytest.fixture()
def settings():
    return TestSettings(
        DOMAIN="localhost",
        ENVIRONMENT="local",
        PROJECT_NAME="test",
        STACK_NAME="test",
        API_PREFIX="/api",
        BACKEND_HOST="http://localhost:8000",
        FRONTEND_HOST="http://localhost:3000",
        BACKEND_CORS_ORIGINS="http://localhost:3000",
        ACCESS_SECRET_KEY="x" * 32,
        REFRESH_SECRET_KEY="y" * 32,
        TOKEN_MODE="stateless",
        AUTH_SERVICE_ROLE="consumer",
        DB_HOST="localhost",
        DB_PORT=5432,
        DB_DATABASE="test",
        DB_USER="test",
        DB_PASSWORD="test",
    )

@pytest.fixture()
def client(settings):
    from fastapi import APIRouter
    router = APIRouter()
    app = create_app(settings, router)
    return TestClient(app)

Use anyio for async tests (required by CLAUDE.md):

import pytest
import anyio

@pytest.mark.anyio
async def test_health(client):
    response = client.get("/api/health/")
    assert response.status_code == 200

Compatibility

`fastapi-m8`	`auth-sdk-m8`	Python
`1.3.0`	`>=1.1.0, <2.0.0`	3.11, 3.12, 3.13
`1.2.0`	`>=1.0.0, <2.0.0`	3.11, 3.12, 3.13
`1.1.4`	`>=0.7.3, <0.8.0`	3.11, 3.12, 3.13
`1.1.0–1.1.3`	`>=0.7.1, <0.8.0`	3.11, 3.12, 3.13
`1.0.x`	`>=0.7.0, <0.8.0`	3.11, 3.12, 3.13

The compatibility matrix is enforced at startup via COMPAT_MATRIX. A RuntimeError is raised immediately if the installed auth-sdk-m8 version is outside the supported range.

Check at runtime:

from fastapi_m8 import CAPABILITIES, __version__

print(__version__)          # "1.3.0"
print(CAPABILITIES)         # {"async": False, "db_optional": True, ...}

create_async_app() is a planned API stub for v2.0. Calling it raises NotImplementedError.

Name		Name	Last commit message	Last commit date
Latest commit History 54 Commits
.github		.github
fastapi_m8		fastapi_m8
tests		tests
.codacy.yml		.codacy.yml
.env.example		.env.example
.gitattributes		.gitattributes
.gitignore		.gitignore
.pydocstyle		.pydocstyle
CHANGELOG.md		CHANGELOG.md
LICENSE		LICENSE
README.md		README.md
pyproject.toml		pyproject.toml

Folders and files

Latest commit

History

Repository files navigation

fastapi-m8

Table of Contents

Summary

Architecture & Package Roles

Installation

Quick Start

1 — Settings

2 — Auth & DB dependencies

3 — Routes

4 — App factory

5 — .env

Configuration Reference

Core / Network

Service Metadata (/meta)

Tokens & Cryptography

Stateful Mode (consumer → auth service)

Auth Event Stream (fa-auth SSE bridge)

Database

Redis

Observability

OpenAPI / Docs

Security / Host Validation

Response Security Headers

API Reference

create_app()

ConsumerServiceSettings

build_auth_deps()

create_db_engine()

Health Checks

Authentication

Token Modes

Role System

Protecting Routes

Health Endpoint

Database Integration

Pre-Start Script

Complete Example

Testing

Compatibility

About

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases 12

Packages 0

Uh oh!

Uh oh!

Contributors

Uh oh!

Languages

5 — `.env`

Service Metadata (`/meta`)

`create_app()`

`ConsumerServiceSettings`

`build_auth_deps()`

`create_db_engine()`

Packages