fastapi-m8 4.1.0

FastAPI application framework for m8 consumer microservices.

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.

---

Table of Contents

1. Summary
2. Architecture & Package Roles
3. Installation
4. Quick Start
5. Configuration Reference
6. API Reference
   • create_app()
   • ConsumerServiceSettings
   • build_auth_deps()
   • create_db_engine()
   • Health Checks
7. Authentication
   • Token Modes
   • Role System
   • Protecting Routes
   • API-key routes (remote principal)
8. Health Endpoint
9. Database Integration
10. Pre-Start Script
11. Complete Example
12. Testing
13. Compatibility

---

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_writer / _admin / _superuser
API-key routes (optional)	AuthDeps.get_current_api_key_reader / _writer → live owner-role resolution via fa-auth-m8 introspection; never admin/superuser
Token revocation (stateful mode)	RemoteRevocationClient → fa-auth-m8 private API (optional short-TTL cache via REVOCATION_CACHE_TTL_SECONDS)
Auth event stream (optional)	build_event_stream_client() → fa-auth SSE bridge for best-effort cache eviction
CORS	Auto-wired from settings.ALLOWED_ORIGINS
Metrics middleware + /metrics	Optional; toggled via METRICS_ENABLED, scrape-gated via METRICS_SCRAPE_CREDENTIAL
Response security headers	Tiered hardening from auth-sdk-m8 (HSTS/CSP express opt-in)
Health endpoint	GET {API_PREFIX}/health/ with optional detail gating
Service meta + liveness	Auto-mounted GET {API_PREFIX}/meta + GET {API_PREFIX}/ping (fail-closed at boot; single-mount at the effective prefix)
Database lifecycle	create_db_engine() wrapping SQLAlchemy
Startup validation	Auto-run check_config_health() + caller startup_validators 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 dependency-free GET {API_PREFIX}/ping liveness probe (→ {"status": "ok"}). /ping is mounted once at the effective prefix (single-mount since auth-sdk-m8 2.0.0): when a prefix is set, only {API_PREFIX}/ping exists so liveness stays reachable behind a prefix-routing reverse proxy (Traefik forwards only PathPrefix({API_PREFIX})); when no prefix is set, /ping is at the root. The single mount always appears in the OpenAPI schema. 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.

⚠️ Breaking change (3.0.0 / auth-sdk-m8 2.0.0): root GET /ping no longer exists when API_PREFIX is set. Update container livenessProbe / sidecar healthcheck URLs from /ping to {API_PREFIX}/ping (e.g. /api/ping).

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	—	The credential for private calls. In legacy mode it is the single shared secret sent as X-Internal-Token (must match auth service); in per-consumer mode (see below) it carries this consumer's bootstrap secret.
ACCESS_REVOCATION_FAILURE_MODE	No	fail_closed	fail_closed (default, secure — reject tokens when the check is unverifiable, returning 503) or fail_open (accept on network/HTTP error — the opt-out is logged loudly and counted as revocation_check_failures_total{mode="fail_open"}).
REVOCATION_CACHE_TTL_SECONDS	No	0	Short-TTL positive validation cache. 0 (default) disables it — every request calls fa-auth. Set to e.g. 30 to trust an active=True result for 30 s, skipping the HTTP round-trip. Entries are tagged with the auth_generation fa-auth returned; stream events (session-revoked/user-deleted) evict affected entries and an unresumable gap flushes all (requires the event-stream client). This TTL is the staleness ceiling — the stream is an accelerator, not the guarantee.

Per-consumer internal auth (item 9.1)

By default a consumer authenticates private calls with the single shared PRIVATE_API_SECRET (legacy mode). Legacy mode is development-only. fa-auth-m8 has retired the issuer's legacy single-shared-secret fallback — its PRIVATE_API_CONSUMERS registry is required in production/strict (item 11.2a) — so a production/strict consumer left in legacy mode is guaranteed to fail against a hardened issuer. Set INTERNAL_CLIENT_ID to switch to the per-consumer model: PRIVATE_API_SECRET becomes this service's bootstrap secret, fa-auth authorizes each private route by the credential's granted scope (deny-by-default), and the blast radius collapses from the whole fleet to one consumer. Optionally exchange the bootstrap credential for short-TTL Authorization: Bearer service tokens so rotation comes for free from the token TTL. Selection is purely by config — the home lab keeps working untouched.

Fatal under production/strict (item 11.2b). When ENVIRONMENT=production or STRICT_PRODUCTION_MODE=true, settings validation fails at construction if INTROSPECTION_URL is set but INTERNAL_CLIENT_ID is unset. Two coherence rules apply in every environment: SERVICE_TOKEN_EXCHANGE_ENABLED=true requires INTERNAL_CLIENT_ID, and setting INTERNAL_CLIENT_ID requires PRIVATE_API_SECRET.

Variable	Required	Default	Description
INTERNAL_CLIENT_ID	No (Yes in prod/strict when INTROSPECTION_URL is set)	—	This consumer's X-Internal-Client id. Unset = legacy single-secret mode (development-only; rejected under production/strict once INTROSPECTION_URL is set); set = per-consumer bootstrap mode.
SERVICE_TOKEN_EXCHANGE_ENABLED	No	false	Exchange the bootstrap credential for short-TTL Bearer service tokens at {issuer}/private/v1/service-token (requires INTERNAL_CLIENT_ID).
SERVICE_TOKEN_SCOPES	No	["introspection"]	Scopes requested when minting a service token; fa-auth narrows to the subset the bootstrap credential was granted.
SERVICE_TOKEN_REFRESH_LEEWAY_SECONDS	No	30	Refresh a cached service token this many seconds before its exp so a call never races expiry.

Remote API-Key Principal

Resolves a user API key presented by an external client to its owner's current authority through the issuer's private introspection endpoint. Off by default; see API-key routes for the route surface.

This group is deliberately self-contained and has no fail-open option. It never consults ACCESS_REVOCATION_FAILURE_MODE — that knob's fail_open value must not reach this path, because an unconfirmable API-key principal is always a denial rather than a downgrade to bare key validity.

Startup fails, not the first request. When API_KEY_INTROSPECTION_ENABLED=true, settings validation fails at construction if there is no endpoint (neither API_KEY_INTROSPECTION_URL nor INTROSPECTION_URL to derive one from), no INTERNAL_CLIENT_ID, no PRIVATE_API_SECRET, or a schema version this auth-sdk-m8 release cannot speak. A half-configured guard never serves traffic.

The consumer's registered identity (INTERNAL_CLIENT_ID) is what fixes the audience the issuer evaluates — it is derived from your credential, never from the request — and the issuer echoes it back so the client can verify it. An active principal carrying a different audience means the registry mapping or the credential is wrong: that is a 503, never a 401.

Variable	Required	Default	Description
API_KEY_INTROSPECTION_ENABLED	No	false	Enable the remote API-key principal dependencies. Requires INTERNAL_CLIENT_ID, PRIVATE_API_SECRET, and an endpoint.
API_KEY_INTROSPECTION_URL	No	derived	POST endpoint, e.g. http://auth_user_service:8000/user/private/v1/api-keys/introspect. Unset = derived from INTROSPECTION_URL.
API_KEY_INTROSPECTION_SCHEMA_VERSION	No	"1"	The contract version this consumer speaks. An unknown version fails at startup; a response declaring one fails closed (503).
API_KEY_INTROSPECTION_CONNECT_TIMEOUT	No	2.0	Connection-establishment timeout, seconds.
API_KEY_INTROSPECTION_READ_TIMEOUT	No	3.0	Response-read timeout, seconds.
API_KEY_INTROSPECTION_POOL_TIMEOUT	No	2.0	How long a request waits for concurrency capacity before being shed with a 503 instead of queueing onto a saturated issuer.
API_KEY_INTROSPECTION_MAX_CONCURRENCY	No	20	Ceiling on in-flight introspection calls.
API_KEY_INTROSPECTION_MAX_RESPONSE_BYTES	No	8192	Hard cap on the issuer response body; a larger reply is abandoned mid-read.
API_KEY_INTROSPECTION_CIRCUIT_FAILURE_THRESHOLD	No	5	Consecutive issuer failures that open the circuit.
API_KEY_INTROSPECTION_CIRCUIT_RESET_SECONDS	No	30.0	How long the circuit stays open before one trial call is admitted.

The issuer must grant this consumer the dedicated api-key-introspection scope. It is deliberately separate from introspection, so a consumer that checks JTI status is not implicitly able to introspect user API keys — grant it explicitly.

The key is never cached and never retried after transmission. Every capability-bearing request introspects the issuer (connection pooling is the only reuse), so a downgrade lands immediately and a transport failure can never be answered from a stale success. Because introspection consumes the key's quota, only a failure known to have occurred before transmission (connection establishment) is retried; a read timeout or 5xx is never replayed.

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). The factory uses the same internal-auth provider as the revocation client — the stream credential is therefore a single config knob shared with JTI introspection: legacy X-Internal-Token, per-consumer bootstrap, or short-TTL Authorization: Bearer service token, selected by INTERNAL_CLIENT_ID / SERVICE_TOKEN_EXCHANGE_ENABLED (see Per-consumer internal auth above).

Pass auth.handle_auth_event as on_event: it applies each event to the validation cache under the rules below, so your service never re-derives them.

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi_m8 import build_event_stream_client

from app.core.deps import auth  # your single build_auth_deps(...) result


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


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

handle_auth_event accepts both v1 and v2 session-revoked events, so it keeps working across the interval where the issuer already emits v2 and consumers have not all upgraded. A v2 event carries the owner's auth_generation and a durable event_id, which together make replay and reorder safe:

Event generation vs. the user's watermark	Outcome
Lower	Already superseded — ignored.
Equal	Applied, unless that exact event_id was already applied at this generation.
Higher	Applied; the watermark advances.

A user-wide v2 event (no jti) evicts only the entries older than the revoking generation, so a session minted by a login that has already re-authenticated survives. A v1 event carries no generation to compare against and therefore evicts the whole user; an unusable payload flushes the cache. The durable event_id — not the SSE transport id, which resets when fa-auth restarts — is the dedup key, so several JTI events sharing one generation stay distinguishable.

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 is set. 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)

Health Detail Gating (items 9.3 + 9.4)

Variable	Default	Description
HEALTH_DETAIL_CREDENTIAL	—	Optional credential for the /health detail body gate. When unset, /health returns a constant {"status":"ok"} to all callers (fail-closed liveness response). When set, callers must present X-Internal-Token: <value> (constant-time match) to receive the real aggregate status, per-check breakdown, and the correct HTTP code (200 or 503). Must not equal PRIVATE_API_SECRET — accidental reuse is a fatal startup misconfiguration. Supports _FILE mount: set HEALTH_DETAIL_CREDENTIAL_FILE=/run/secrets/health_cred.txt.

Constant ungated body (item 9.4 Design B): ungated callers always receive 200 {"status":"ok"} regardless of the real aggregate health (including degraded or fail). This prevents operational-state leakage on public-HTTPS stacks: a degraded response is a timing oracle that signals fail-open degradation is active. Only callers presenting a valid HEALTH_DETAIL_CREDENTIAL see the real status, the per-check detail, and the 503 HTTP code when checks fail. Liveness probes and {API_PREFIX}/ping are unaffected.

No-reuse enforcement (item 9.3): at startup, create_app asserts that neither HEALTH_DETAIL_CREDENTIAL nor METRICS_SCRAPE_CREDENTIAL equals PRIVATE_API_SECRET. This ensures each credential is independently rotatable — exposing one value cannot open multiple surfaces. The check raises ConfigurationError (fail-closed), aborting startup.

Observability

Variable	Default	Description
METRICS_ENABLED	false	Enable Prometheus metrics middleware and the /metrics route
METRICS_GROUPS	—	Comma-separated groups: traffic, performance, reliability, health, auth, or all
METRICS_SCRAPE_CREDENTIAL	—	Optional static bearer credential for the /metrics scrape endpoint. When set, requests must present Authorization: Bearer <value> (constant-time match). When unset, /metrics relies on network isolation only. Must not equal PRIVATE_API_SECRET.

/metrics route: when METRICS_ENABLED=true, create_app also registers a GET /metrics endpoint (hidden from the schema) rendering the Prometheus registry. Set METRICS_SCRAPE_CREDENTIAL to gate scrapes with a bearer credential — configure Prometheus scrape_configs.authorization.credentials to match. The revocation cache (when enabled) also emits revocation_cache_lookups_total{result="hit"|"miss"} and a revocation_cache_ttl_seconds gauge on the same registry; no JTI, user ID, or secret is ever used as a label or value.

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.

Defaults by layer (consumer)

How the security-critical settings behave specifically in the fastapi-m8 consumer layer. This is the consumer column of the fleet-wide defaults-by-layer table (the SDK and fa-auth-m8 READMEs hold the issuer/SDK columns). The throughline: secure-by-default, inherited from auth-sdk-m8, and made fatal only when pointed at production.

Control	Consumer behaviour	Becomes fatal when
Config-health gate	create_app auto-runs check_config_health in its lifespan before the app signals ready — no manual wiring	any fatal check trips (raises ConfigurationError, app refuses to boot)
ACCESS_REVOCATION_FAILURE_MODE	fail_closed by default — an unverifiable revocation check returns 503; fail_open is a conscious opt-out, logged loudly and counted (revocation_check_failures_total{mode="fail_open"})	never fatal (it is the degradation policy)
ALLOWED_HOSTS	inherited from CommonSettings; unset = no host check (dev)	unset under STRICT_PRODUCTION_MODE (strict fatal); wildcard * under strict
EVENT_SIGNING_ENABLED / EVENT_SIGNING_ACCEPT_UNSIGNED	inherited secure defaults (signing on, unsigned rejected); the gate fires through the consumer's auto-run config-health (item 7.x.1)	ENABLED=false under strict; ACCEPT_UNSIGNED=true under production or strict
METRICS_SCRAPE_CREDENTIAL	unset = /metrics relies on network isolation only; set = constant-time Authorization: Bearer gate. Must not equal PRIVATE_API_SECRET (fatal reuse check at startup).	never fatal (network-isolation is a valid posture); reuse of PRIVATE_API_SECRET is fatal
HEALTH_DETAIL_CREDENTIAL (items 9.3 + 9.4)	unset = /health returns a constant {"status":"ok"} to all callers (fail-closed liveness; real status never leaked); set = X-Internal-Token must match to receive the real aggregate status, per-check breakdown, and correct HTTP code. Must not equal PRIVATE_API_SECRET (fatal reuse check at startup).	never fatal (no credential = constant liveness response, which is a valid posture); reuse of PRIVATE_API_SECRET is fatal
INTERNAL_CLIENT_ID (items 9.1 + 11.2b)	unset = legacy single PRIVATE_API_SECRET (development-only); set = per-consumer bootstrap / service-token auth on private calls (must be coordinated with the issuer's PRIVATE_API_CONSUMERS)	fatal under production/strict when INTROSPECTION_URL is set but INTERNAL_CLIENT_ID is unset; also fatal (any env) when SERVICE_TOKEN_EXCHANGE_ENABLED=true without INTERNAL_CLIENT_ID, or INTERNAL_CLIENT_ID set without PRIVATE_API_SECRET
_FILE secret mounts	inherited from CommonSettings — every secret (PRIVATE_API_SECRET_FILE, DB_PASSWORD_FILE, METRICS_SCRAPE_CREDENTIAL_FILE, HEALTH_DETAIL_CREDENTIAL_FILE, …) can be sourced from /run/secrets/* with no code change	a referenced <FIELD>_FILE path is missing (fails closed at construction)

---

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:

1. Run the auto-prepended check_config_health() validator (from auth-sdk-m8), then lifecycle.startup_validators — raise any exception to prevent the ready signal. The config-health check runs first, so a fatal misconfiguration (e.g. production localhost CORS origins, a wildcard ALLOWED_HOSTS under strict mode) aborts startup with ConfigurationError before any caller validators run.
2. Enter lifecycle.lifespan_extras context (if provided).
3. Set app.state.service_ready = True.
4. (app serves traffic)
5. Exit lifecycle.lifespan_extras.
6. Call lifecycle.auth_deps.close() (closes revocation HTTP client).
7. 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_writer	Callable	Raises 403 unless user has WRITER, ADMIN or SUPERADMIN role
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 (both are required; neither claim grants privilege on its own)
auth.revocation_client	RemoteRevocationClient | None	Present only in stateful mode
auth.handle_auth_event	Callable	Pass as on_event to build_event_stream_client; applies v1/v2 session-revoked and user-deleted events to the validation cache
auth.get_current_api_key_principal	async Callable | None	Resolves an X-API-Key to its owner's current authority via the issuer. Proves a live owner only — never writer capability
auth.require_api_key_role	Callable | None	Factory: require_api_key_role(RoleType.WRITER) builds a capability dependency. Rejects any role above WRITER
auth.get_current_api_key_reader	async Callable | None	Raises 403 unless the key's owner is at least READER
auth.get_current_api_key_writer	async Callable | None	Raises 403 unless the key's owner is at least WRITER and the key is read_write
auth.api_key_client	ApiKeyIntrospectionClient | None	Present only when API_KEY_INTROSPECTION_ENABLED

The four API-key members are None unless API_KEY_INTROSPECTION_ENABLED=true.

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
tenant_id	uuid.UUID | None	Tenant claim (populated when the token carries tenant_id; None for untenanted/legacy tokens). Requires auth-sdk-m8 ≥ 1.3.0.

---

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}

# WRITER, ADMIN or SUPERADMIN
@router.post("/items")
async def create_item(
    writer: Annotated[UserModel, Depends(auth.get_current_active_writer)],
):
    ...

# 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

---

API-key routes (remote principal)

An API key is an opaque pointer to its owner, not a credential with capabilities of its own — it stores no role. A downstream service does not share the issuer's database, so it resolves the owner's current role by calling the issuer on every capability-bearing request. That is what makes a role downgrade take effect on the key's next request, in every token mode.

Enable it with API_KEY_INTROSPECTION_ENABLED=true (see Remote API-Key Principal); the dependencies are None until you do. The key travels on the X-API-Key header — the same header the issuer reads, so a key behaves identically at either end.

# Reads only — the owner must be at least READER.
@router.get("/items")
async def list_items(
    principal: Annotated[ApiKeyPrincipal, Depends(auth.get_current_api_key_reader)],
):
    ...

# Writes — the owner must be at least WRITER *and* the key must be read_write.
@router.post("/items")
async def create_item(
    principal: Annotated[ApiKeyPrincipal, Depends(auth.get_current_api_key_writer)],
):
    ...

Effective authority is an intersection, and every dimension can only narrow it:

owner's current role  ∩  key access_mode  ∩  matching audience  ∩  route policy

So a read_only key owned by a writer cannot write; a read_write key owned by a reader cannot write; and deactivating an owner denies every key they hold.

API keys never carry administrative or superuser authority — not from the owner's role, not from is_superuser, and not by configuration. There is no get_current_api_key_admin/_superuser dependency and no setting that creates one; require_api_key_role() raises ApiKeyCapabilityCeilingError at wiring time for anything above WRITER. Keep those operations on the JWT guards.

Depending on get_current_api_key_principal never implies write capability. It proves only that the key resolves to a live owner. Every capability-bearing route must use a role dependency.

Responses:

• 401 Invalid or expired API key — one generic denial covering an unknown, revoked, or expired key and every owner-state cause, so a caller cannot probe another account's state
• 403 Forbidden — the owner's role or the key's access mode is insufficient
• 429 — the key's quota is exhausted; the issuer's Retry-After is relayed
• 503 — the principal could not be confirmed (issuer outage, timeout, open circuit, shed load, or a response this release cannot interpret). Always a denial — there is no fallback to a cached principal or to bare key validity

---

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": "3.0.0",
  "auth_sdk_m8": "2.0.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
4.0.0	>=3.0.0, <4.0.0	3.11, 3.12, 3.13, 3.14
3.3.0	>=2.1.1, <3.0.0	3.11, 3.12, 3.13, 3.14
3.2.0	>=2.1.0, <3.0.0	3.11, 3.12, 3.13, 3.14
3.1.0	>=2.1.0, <3.0.0	3.11, 3.12, 3.13, 3.14
3.0.0	>=2.0.1, <3.0.0	3.11, 3.12, 3.13, 3.14
2.1.0	>=1.5.0, <2.0.0	3.11, 3.12, 3.13
2.0.0	>=1.4.0, <2.0.0	3.11, 3.12, 3.13
1.6.0	>=1.3.0, <2.0.0	3.11, 3.12, 3.13
1.5.0	>=1.2.1, <2.0.0	3.11, 3.12, 3.13
1.4.0	>=1.2.0, <2.0.0	3.11, 3.12, 3.13
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__)          # "3.0.0"
print(CAPABILITIES)         # {"async": False, "plugin_system": False,
                            #  "trace_context": False, "db_optional": True,
                            #  "health_detail_gating": True}

create_async_app() is a reserved stub for a future async app surface. Calling it raises NotImplementedError; check CAPABILITIES["async"] before using it.