fastapi-service（FastAPI service 测试）

FastAPI Service

Build production-grade FastAPI services with JWT authentication, structured logging via structlog, and Prometheus metrics.

Quick Start

import os
import sys
import logging
from contextlib import asynccontextmanager
from typing import Dict

import structlog
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from prometheus_fastapi_instrumentator import Instrumentator

# -------------------------
# Logging configuration
# -------------------------
def configure_logging() -> None:
    log_level_name = os.getenv("LOG_LEVEL", "INFO").upper()
    log_level = getattr(logging, log_level_name, logging.INFO)
    json_logs = os.getenv("LOG_JSON", "false").lower() == "true"
    service_name = os.getenv("SERVICE_NAME", "fastapi-app")

    logging.basicConfig(level=log_level, stream=sys.stdout)

    processors = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
    ]
    if json_logs:
        processors.append(structlog.processors.JSONRenderer())
    else:
        processors.append(structlog.dev.ConsoleRenderer())

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger(log_level),
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

    structlog.contextvars.bind_contextvars(service=service_name)

configure_logging()
log = structlog.get_logger()

# -------------------------
# Auth configuration
# -------------------------
JWT_SECRET = os.getenv("AUTH_SECRET", "change-me")
JWT_ALG = os.getenv("JWT_ALG", "HS256")
security = HTTPBearer(auto_error=True)

def require_claims(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> Dict:
    token = credentials.credentials
    try:
        claims = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG])
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return claims

# -------------------------
# App & instrumentation
# -------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
    log.info("service.startup")
    try:
        yield
    finally:
        log.info("service.shutdown")

app = FastAPI(title=os.getenv("SERVICE_NAME", "fastapi-app"), lifespan=lifespan)

# Prometheus: exposes /metrics by default
Instrumentator().instrument(app).expose(app)

@app.get("/healthz", tags=["health"])
def healthz():
    return {"status": "ok"}

@app.get("/whoami", tags=["auth"])
def whoami(claims: Dict = Depends(require_claims)):
    return {"claims": claims}

if __name__ == "__main__":
    import uvicorn

    host = os.getenv("HOST", "0.0.0.0")
    port = int(os.getenv("PORT", "8000"))
    uvicorn.run("main:app", host=host, port=port, reload=False)

Dependencies

uv add fastapi "uvicorn[standard]" "python-jose[cryptography]" prometheus-fastapi-instrumentator structlog

Environment

Create .env.server:

export AUTH_SECRET="your-strong-shared-secret"
export LOG_LEVEL=INFO
export LOG_JSON=false

Makefile

SHELL := /bin/bash

define setup_env
        $(eval ENV_FILE := $(1))
        $(eval include $(1))
        $(eval export)
endef

run-server:
    $(call setup_env, .env.server)
    uv run uvicorn main:app --host 0.0.0.0 --port 8000

Testing Auth

Generate a test token:

uv run python - <<'PY'
from jose import jwt
print(jwt.encode({"sub":"alice","role":"admin"}, "your-strong-shared-secret", algorithm="HS256"))
PY

Call the authenticated endpoint:

curl -H "Authorization: Bearer <token>" http://localhost:8000/whoami

Key Patterns

• Lifespan: Use @asynccontextmanager for startup/shutdown, not deprecated on_event
• Dependency injection: Use Depends() for auth, DB connections, and shared resources
• Structured logging: Configure structlog once at module level; bind context vars per-request
• Metrics: prometheus-fastapi-instrumentator auto-instruments all routes
• Health check: Always expose /healthz for k8s readiness/liveness probes

Bookkeeping

After modifying the server, update the adjacent README.md to reflect the current API surface.