fastapi-service

FastAPI 服务

构建生产级 FastAPI 服务，支持 JWT 身份验证、通过 structlog 实现结构化日志记录，以及 Prometheus 指标。

快速开始

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)

依赖项

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

环境配置

创建 .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

测试身份验证

生成测试令牌：

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

调用经过身份验证的端点：

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

关键模式

• 生命周期：使用 @asynccontextmanager 处理启动和关闭，不要使用已弃用的 on_event
• 依赖注入：使用 Depends() 处理身份验证、数据库连接和共享资源
• 结构化日志记录：在模块级别配置一次 structlog；为每个请求绑定上下文变量
• 指标：prometheus-fastapi-instrumentator 会自动为所有路由添加监控
• 健康检查：始终公开 /healthz，供 k8s 就绪和存活探针使用

维护记录

修改服务后，更新相邻的 README.md，以反映当前的 API 接口。