fastapi-service

FastAPI サービス

JWT 認証、structlog による構造化ログ、Prometheus メトリクスを備えた、本番環境向けの FastAPI サービスを構築します。

クイックスタート

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

主なパターン

• ライフスパン：非推奨の on_event ではなく、起動／シャットダウンに @asynccontextmanager を使用します
• 依存性注入：認証、DB 接続、共有リソースには Depends() を使用します
• 構造化ログ：structlog はモジュールレベルで一度設定し、リクエストごとにコンテキスト変数をバインドします
• メトリクス：prometheus-fastapi-instrumentator がすべてのルートを自動で計測します
• ヘルスチェック：k8s のレディネス／ライブネスプローブ用に、常に /healthz を公開します

管理

サーバーを変更した後は、現在の API サーフェスを反映するよう、隣接する README.md を更新してください。