fastapi-m8 1.3.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
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_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
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-bus signing is ON by default (inherited from auth-sdk-m8>=1.0.0).
EVENT_SIGNING_KEY=change-me-event-signing-32-chars
# EVENT_SIGNING_ENABLED=false   # opt out only if you do not use the event bus

TOKEN_MODE=stateless
AUTH_SERVICE_ROLE=consumer

# 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

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 event-bus payloads. Set false to disable.
EVENT_SIGNING_KEY	Yes²	—	Shared HMAC secret for the event bus. 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)

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, since auth-sdk-m8 ≥ 1.1.0). The headers are emitted only in production — when ENVIRONMENT=production or STRICT_PRODUCTION_MODE=true — so local/dev keeps Swagger, ReDoc, and HMR working unrestricted. In production the response carries X-Frame-Options: DENY, X-Content-Type-Options: nosniff, a Content-Security-Policy with frame-ancestors 'none', Referrer-Policy, Permissions-Policy, and HSTS.

Variable	Default	Description
SECURITY_HEADERS_ENABLED	true	Master switch; set false to suppress the layer even in production
HSTS_MAX_AGE	31536000	Strict-Transport-Security max-age in seconds; 0 drops the HSTS header only
HSTS_INCLUDE_SUBDOMAINS	true	Append includeSubDomains to HSTS
CONTENT_SECURITY_POLICY	(hardened default)	Override the emitted Content-Security-Policy
REFERRER_POLICY	strict-origin-when-cross-origin	Referrer-Policy value
PERMISSIONS_POLICY	(restrictive default)	Permissions-Policy value

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:

1. Run lifecycle.startup_validators — raise any exception to prevent ready signal.
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_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.