FastAPI

FastAPI + SQLAlchemy性能改善プレイブック: 遅いAPIを計測ベースで高速化する FastAPIの初期実装は非常に快適です。しかし運用フェーズに入ると、次のような症状が出てきます。 一覧APIのレスポンスが急に遅くなる 同時接続が増えるとp95が跳ねる CPUは余っているのにタイムアウトが増える DB接続数が上限に張り付く こうした問題の多くは「Pythonが遅い」のではなく、SQLAlchemyの使い方とDBアクセス設計 に起因します。 本記事では、FastAPI + SQLAlchemy + PostgreSQL構成を前提に、実際の改善手順を計測ベースで整理します。 1. 最初に測るべき指標 最適化は、体感ではなく数値で進めます。最低限、以下を可視化します。 APIのp50/p95/p99レイテンシ エンドポイント別SQL発行回数 1リクエストあたりのDB滞在時間 connection pool待ち時間 slow query件数（200ms以上など） OpenTelemetryやNew Relicを使っているなら、アプリspanとDB spanを必ず紐付けてください。これだけでボトルネック特定速度が上がります。 2. N+1問題を最優先で潰す 最も頻出するのがN+1です。例えばユーザー一覧でプロフィールを参照すると、ユーザー数分の追加クエリが発行されます。 2.1 悪い例 1 2 3 4 5 6 7 8 users = session.query(User).limit(100).all() result = [] for u in users: result.append({ "id": u.id, "name": u.name, "profile": u.profile.bio, }) 2.2 改善例（joinedload/selectinload） 1 2 3 4 5 6 7 8 from sqlalchemy.orm import selectinload users = ( session.query(User) .options(selectinload(User.profile)) .limit(100) .all() ) joinedload と selectinload はデータ量で使い分けます。 ...

FastAPI + Celery信頼性設計: 非同期ジョブを本番で壊さないための実装パターン FastAPIでAPIを作ると、重い処理はすぐに非同期ジョブへ逃がしたくなります。画像変換、レポート生成、外部API連携、メール配信など、Celeryは非常に便利です。ですが、本番で問題になるのは「動くかどうか」ではなく、失敗したときに壊れないか です。 同じジョブが二重実行される 一時障害で永遠にリトライしてキューが詰まる ワーカー再起動で中途半端な状態が残る 完了通知が先に飛んで実データがない 本記事では FastAPI + Celery + Redis 構成を前提に、再実行安全性（idempotency）と運用信頼性を上げる実装手順をまとめます。 1. まず守るべき設計原則 非同期基盤の事故は、ほぼ次の4原則で防げます。 At-least-once前提（同一タスク再実行は必ず起こる） 副作用は冪等化（何回実行されても結果が壊れない） 状態遷移を明示（PENDING/RUNNING/SUCCEEDED/FAILED） 失敗を可観測化（リトライ回数・死活・滞留時間を計測） この原則を外すと、障害時に「何が完了して何が未完了か」が追えなくなります。 2. 参照アーキテクチャ API: FastAPI Queue Broker: Redis Worker: Celery Result Store: PostgreSQL（業務状態） Monitoring: Flower + Prometheus + Sentry ポイントは、業務上重要な状態はRedis結果バックエンドに依存しない ことです。Redisは一時的に使い、真実の状態はRDBに持たせます。 3. 実装の土台: タスク受付API 3.1 受け付け時に idempotency_key を必須化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 from fastapi import FastAPI, HTTPException from pydantic import BaseModel from sqlalchemy import select app = FastAPI() class JobRequest(BaseModel): idempotency_key: str report_type: str user_id: str @app.post("/reports") def create_report(req: JobRequest): existing = find_job_by_key(req.idempotency_key) if existing: return {"job_id": existing.id, "status": existing.status} job = create_job_record( idempotency_key=req.idempotency_key, status="PENDING", report_type=req.report_type, user_id=req.user_id, ) generate_report.delay(job.id) return {"job_id": job.id, "status": "PENDING"} これでクライアント再送が来てもジョブ多重作成を防げます。 ...

FastAPI認証・認可の本番設計：JWT運用、権限制御、監査ログまで含めた実装パターン FastAPI は実装が速い反面、認証・認可を最小構成のまま本番に出してしまい、後からセキュリティ事故に発展するケースが少なくありません。特に「JWT を入れたから安全」という誤解は危険です。 本記事では、開発速度を落とさずに本番で耐える認証基盤を作るための設計を、コード例と運用手順込みで解説します。 1. 認証と認可を分離して設計する 最初に押さえるべきは責務分離です。 認証（Authentication）: 誰かを確認する 認可（Authorization）: 何をしてよいか判定する この2つを混ぜると、実装も監査も破綻します。FastAPI では dependency を分け、get_current_user と require_permission を独立させるのが基本です。 2. JWT は「短命 + リフレッシュ + 失効管理」で使う アクセストークンを長寿命にすると、漏えい時の被害が大きくなります。実運用では以下が標準です。 Access Token: 5〜15分 Refresh Token: 7〜30日 Refresh Token は DB 保存し、ローテーション時に旧トークンを失効 sub だけでなく、jti（トークンID）や scope を持たせると管理しやすくなります。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from datetime import datetime, timedelta, timezone import jwt ALGORITHM = "HS256" def create_access_token(user_id: str, scopes: list[str], secret: str) -> str: now = datetime.now(timezone.utc) payload = { "sub": user_id, "scope": " ".join(scopes), "iat": int(now.timestamp()), "exp": int((now + timedelta(minutes=10)).timestamp()), "jti": "generated-uuid" } return jwt.encode(payload, secret, algorithm=ALGORITHM) 3. 鍵管理とローテーション 秘密鍵を .env に固定して数年運用するのは典型的な事故パターンです。最低限、次を実施します。 ...

FastAPI本番運用ハードニング完全ガイド：セキュリティ・性能・障害対応を実装で固める FastAPI は開発速度が高く、PoC から本番まで一気に進めやすいフレームワークです。しかし、早く作れることと安全に運用できることは別問題です。実際の障害は、コードの正しさよりも運用の隙から発生します。 本記事では、FastAPI を本番で安心して運用するためのハードニング手順を、実装可能な形でまとめます。対象は「すでにAPIが動いているが、運用強度を上げたい」チームです。 1. 入口防御：TLS、ヘッダ、レート制限 TLS終端とForwardedヘッダ ロードバランサ配下で動かす場合、X-Forwarded-For と X-Forwarded-Proto の扱いを明確にします。誤るとクライアントIPが取れず、監査も制限も機能しません。 1 2 3 4 5 from fastapi import FastAPI from starlette.middleware.trustedhost import TrustedHostMiddleware app = FastAPI() app.add_middleware(TrustedHostMiddleware, allowed_hosts=["api.example.com", "*.example.com"]) allowed_hosts をワイルドにしすぎると Host Header Injection の温床になります。 セキュリティヘッダ 最低限次を返します。 Strict-Transport-Security X-Content-Type-Options: nosniff X-Frame-Options: DENY Referrer-Policy APIでも無関係ではありません。管理画面やドキュメントUIを守る意味があります。 レート制限 ブルートフォースと突発負荷に備え、IPまたはAPIキー単位でレート制限を設定します。 認証系: 5 req/min 通常API: 60 req/min 高負荷検索: 20 req/min Redis バックエンド方式にして、アプリ再起動でカウンタが失われないようにします。 ...