IT & ライフハックブログ｜学びと実践のためのアイデア集

毎日に役立つライフハックをサクッと紹介 Support by AI

システム開発

爆速＆堅牢！FastAPIのキャッシュ戦略完全ガイド――Redis・HTTPキャッシュ・ETag・レート制限・圧縮まで

投稿者 greeden #FastAPI, #Python, #システム開発
green snake
Photo by Pixabay on Pexels.com

爆速＆堅牢！FastAPIのキャッシュ戦略完全ガイド――Redis・HTTPキャッシュ・ETag・レート制限・圧縮まで

✅ まずは要約（インバーテッドピラミッド）

  • この記事でできること
    FastAPIで体感速度を上げるキャッシュ設計を、アプリ層（Redis）とHTTP層（Cache-Control/ETag/304）で二枚看板にして実装できます。さらにレート制限やGZip圧縮も合わせて導入し、パフォーマンスと安定性の両立を図ります。
  • 主なトピック
    1. キャッシュの基礎と「どこで効かせるか」（アプリ・HTTP・データ）
    2. **Redis（非同期）**を使ったキー設計・TTL・無効化（インバリデーション）
    3. HTTPキャッシュ：ETag／If-None-Match／Cache-Controlで304応答
    4. FastAPI実装例：ページネーション結果のキャッシュ、タグ無効化、部分更新と整合性
    5. レート制限（トークンバケット）とGZip圧縮の同時導入
  • 得られる効果
    • 同時接続やピークトラフィックに強く、コスト削減＆応答時間短縮
    • バックエンド（DB/外部API）への負荷を平準化
    • 仕様としての再現性（304/ヘッダー）により、フロントやCDNと協調しやすくなる

🎯 誰が読んで得をする？（具体像）

  • 個人開発者Aさん（学部4年）
    学内向けAPIが「閲覧が集中すると急に遅い…」という悩み。Redis＋HTTPキャッシュでリスト系エンドポイントを爆速化したい。
  • 受託開発Bさん（3名チーム）
    顧客のダッシュボードAPIに秒間スパイク。レート制限で守りながら、304応答でフロントキャッシュを活かしたい。
  • SaaS Cさん（スタートアップ）
    外部APIへの依存が大きく、料金やレート上限が厳しい。フェッチ結果を賢くキャッシュして、コストと障害の両方を抑えたい。

♿ アクセシビリティ評価（本記事の読みやすさ）

  • レベル：AA相当。見出し・箇条書きで構造化し、専門語は初出で短く定義。コードはコメント入りで固定幅表示。
  • 配慮：章頭に「要点まとめ」を置き、スクリーンリーダーでも流れを追いやすく設計。要所で実行可能な最小サンプルを提示。
  • 対象：初学者には「どこから手を付けるか」を明確化し、中級者にはキー設計・整合性・無効化まで踏み込みます。

1. キャッシュを「どこで」効かせるか

キャッシュは大きく3層に分けて考えます。

  1. HTTP層（ブラウザ／CDN）
    • Cache-Control / ETag / If-None-Match などのプロトコル機能で効かせる。
    • 変更が少ない一覧表示や詳細取得と相性抜群。304 Not Modifiedで帯域節約。
  2. アプリ層（RedisなどのKVS）
    • APIサーバ側で計算結果や外部APIレスポンスを保存。
    • TTL（有効期限）・タグ無効化・再計算の制御が柔軟。
  3. データ層（DBのMaterialized View等）
    • 集計結果の前計算・更新頻度の設計。記事では主にアプリ/HTTP層を扱います。

要点まとめ

  • HTTPキャッシュは“無料の最適化”：ヘッダを返すだけでブラウザ/CDNが仕事をしてくれます。
  • Redisは“自由度の高い武器”：TTLやタグで無効化ロジックを自前で組めます。
  • まずはHTTPヘッダ→Redisの順で導入するのがおすすめ♡

2. サンプル構成と依存（前提）

2.1 依存ライブラリ（例）

fastapi
uvicorn[standard]
redis>=5.0  # asyncio対応の公式クライアント
pydantic
pydantic-settings

2.2 設定クラス

# app/core/settings.py
from pydantic import Field
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI Cache Guide"
    redis_url: str = "redis://localhost:6379/0"
    default_ttl_seconds: int = 60  # デフォルトTTL

    class Config:
        env_file = ".env"
        extra = "ignore"

def get_settings() -> Settings:
    return Settings()

2.3 Redisクライアント（async）

# app/core/redis_client.py
from redis import asyncio as aioredis
from app.core.settings import get_settings

redis: aioredis.Redis | None = None

async def connect_redis():
    global redis
    settings = get_settings()
    redis = aioredis.from_url(
        settings.redis_url,
        encoding="utf-8",
        decode_responses=True,
    )

async def disconnect_redis():
    global redis
    if redis:
        await redis.close()
        redis = None

要点まとめ

  • 公式redisパッケージのasync APIを使用。
  • アプリ起動/終了時に接続プールを開閉します。

3. HTTPキャッシュの基本：ETag と 304

ETag（エンティティタグ）はレスポンス本文の「指紋」。クライアントは次回のリクエストで If-None-Match: <etag> を送り、サーバ側が内容不変と判断すれば304で本文を返さずに済みます。

3.1 ETagの付与と条件付きGET

# app/utils/http_cache.py
import hashlib
from fastapi import Request, Response

def make_etag(payload: bytes) -> str:
    # 本文のハッシュをETagに
    return '"' + hashlib.sha256(payload).hexdigest() + '"'

def conditional_response(request: Request, response: Response, body_bytes: bytes):
    etag = make_etag(body_bytes)
    client_etag = request.headers.get("if-none-match")
    response.headers["ETag"] = etag
    # 一致すれば304で本文省略
    if client_etag == etag:
        response.status_code = 304
        response.body = b""
    else:
        response.body = body_bytes

3.2 Cache-Control の設計

  • 公開可（CDNや共有キャッシュに乗せたい）：Cache-Control: public, max-age=60
  • ユーザー別（認証含む応答）：Cache-Control: private, max-age=0, no-cache（基本はETag中心）
  • 変更が滅多にない：max-age を長めに。バックエンドが更新されたらETagが変わるようにする。

要点まとめ

  • ETag＋304は帯域と時間を節約。
  • Cache-Controlは公開/非公開と**寿命（max-age）**を明確化。
  • 認証が絡む応答はprivateが基本（共有キャッシュに載せない）。

4. アプリ層キャッシュ：Redisで“作らない”を作る

4.1 キー設計（失敗しない型）

  • キーの粒度：<resource>:<path_hash> でURL＋クエリの組み合わせを一意化。
  • バージョン：v1: を先頭に付けると、互換性崩れのとき一発無効化できます。
  • ユーザー別：認証がある場合はuser:<id>もキーに含める（プライバシー保護）。
  • タグ：tag:<category> のセットにキーを登録しておくと、タグ単位で一括削除が可能。

4.2 実装：読み・書き・TTL・タグ登録

# app/utils/app_cache.py
import json, hashlib
from typing import Any
from fastapi import Request
from app.core.redis_client import redis
from app.core.settings import get_settings

def cache_key_from_request(request: Request, prefix="v1:list") -> str:
    # パス＋クエリをハッシュ化
    raw = request.url.path + "?" + "&".join(sorted(request.query_params.multi_items()))
    return f"{prefix}:{hashlib.sha256(raw.encode()).hexdigest()}"

async def cache_get(key: str) -> bytes | None:
    if not redis:
        return None
    return await redis.get(key)

async def cache_set(key: str, value: bytes, ttl: int | None = None):
    if not redis:
        return
    settings = get_settings()
    await redis.set(key, value, ex=ttl or settings.default_ttl_seconds)

async def tag_add(tag: str, key: str):
    # タグ→キーの集合に登録
    if redis:
        await redis.sadd(f"tag:{tag}", key)

async def tag_invalidate(tag: str):
    if not redis:
        return 0
    tag_key = f"tag:{tag}"
    members = await redis.smembers(tag_key)
    if members:
        await redis.delete(*members)   # まとめて削除
    await redis.delete(tag_key)
    return len(members or [])

要点まとめ

  • キー=関数の入力を意識（パス・クエリ・ユーザー）。
  • タグで「カテゴリ一括無効化」を実現。
  • TTLはまず60秒程度で効果を体感→調整が吉。

5. ページネーションAPIのキャッシュ実装（実例）

5.1 ルーターの用意

# app/main.py
from fastapi import FastAPI, Request, Response, Query, HTTPException
from fastapi.middleware.gzip import GZipMiddleware
from app.core.settings import get_settings
from app.core.redis_client import connect_redis, disconnect_redis
from app.utils.http_cache import conditional_response
from app.utils.app_cache import cache_key_from_request, cache_get, cache_set, tag_add, tag_invalidate
import json

app = FastAPI(title="FastAPI Cache Demo")
app.add_middleware(GZipMiddleware, minimum_size=600)  # 圧縮で帯域削減（後述）

@app.on_event("startup")
async def on_startup():
    await connect_redis()

@app.on_event("shutdown")
async def on_shutdown():
    await disconnect_redis()

# 疑似データソース（本番はDB）
FAKE_POSTS = [
    {"id": i, "title": f"post-{i}", "category": "tech" if i % 2 == 0 else "life"}
    for i in range(1, 501)
]

@app.get("/posts")
async def list_posts(
    request: Request,
    response: Response,
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0),
    category: str | None = Query(None)
):
    # 1) まずアプリ層キャッシュを見る
    key = cache_key_from_request(request, prefix="v1:posts")
    cached = await cache_get(key)
    if cached:
        # 2) キャッシュ命中：ETag/304判定を行い、帯域節約
        conditional_response(request, response, cached)
        response.headers["Cache-Control"] = "public, max-age=30"
        return Response(content=response.body, media_type="application/json", headers=response.headers)

    # 3) データを作る（本番はDBクエリ）
    items = [p for p in FAKE_POSTS if (category is None or p["category"] == category)]
    total = len(items)
    page = items[offset: offset + limit]
    body = json.dumps({"total": total, "limit": limit, "offset": offset, "items": page}).encode()

    # 4) 保存＋タグ登録（カテゴリで一括無効化できるように）
    await cache_set(key, body, ttl=60)
    if category:
        await tag_add(f"posts:cat:{category}", key)

    # 5) HTTPキャッシュ（ETag/304）とヘッダ
    conditional_response(request, response, body)
    response.headers["Cache-Control"] = "public, max-age=30"
    return Response(content=response.body, media_type="application/json", headers=response.headers)

5.2 無効化（例：管理者が投稿を更新した時）

@app.post("/admin/posts/invalidate")
async def invalidate_posts(category: str):
    # カテゴリ単位でキャッシュを一括削除
    deleted = await tag_invalidate(f"posts:cat:{category}")
    return {"invalidated": deleted}

要点まとめ

  • まずRedis→命中ならETagで304判定→帯域も処理も軽く。
  • キーはURL＋クエリで一意化、カテゴリタグで「まとめて無効化」。
  • GZipMiddleware で圧縮も同時に行い、ネットワークを節約。

6. レート制限（守りのキャッシュ設計）

トークンバケット方式で、一定間隔でトークンを補充し、超えたら429を返す仕組みです。Redisの原子的操作で安全に実装できます。

# app/utils/rate_limit.py
import time
from fastapi import HTTPException, status

# 例：ユーザー/IPごとに 1分間に 60 リクエスト
async def token_bucket(redis, bucket_key: str, limit=60, refill_seconds=60):
    now = int(time.time())
    # バケットに「期限」を設け、期間切り替わりでリセット
    window = now // refill_seconds
    key = f"ratelimit:{bucket_key}:{window}"
    current = await redis.incr(key)
    if current == 1:
        await redis.expire(key, refill_seconds)
    if current > limit:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded. Please try again later."
        )

適用例（IP単位）：

from app.utils.rate_limit import token_bucket
from app.core.redis_client import redis

@app.get("/protected")
async def protected(request: Request):
    ip = request.client.host
    await token_bucket(redis, bucket_key=f"ip:{ip}", limit=120, refill_seconds=60)
    return {"ok": True}

要点まとめ

  • 乱暴なアクセスからバックエンドを保護。
  • バケットキーをIP／ユーザー／APIキーで切り替え。
  • 業務要件に合わせてエンドポイント単位で制御。

7. 一部更新と整合性：キャッシュ“破棄”の設計

**インバリデーション（無効化）**はキャッシュの難所です。実運用での指針は以下です。

  1. 最小限の無効化：変更範囲のみ削除（例：カテゴリ更新→カテゴリタグの集合）
  2. バージョン先頭付け：v1: → 互換性が崩れた時に切り替えやすい
  3. 短寿命＋上書き：TTLを短くし、多少の遅延容認で運用を簡素化
  4. 書き込み時無効化：POST/PUT/PATCH/DELETEのハンドラで該当タグを必ず消す
  5. ジョブ化：大量無効化は非同期ジョブ（Celery等）で

要点まとめ

  • 「何が影響するか」をタグで表現→ピンポイント削除。
  • TTL短め＋タグ無効化＝現実解でトラブルが少ない♡

8. 認証応答とキャッシュ：private／Vary

  • 認証ヘッダ（Authorization）が付くレスポンスは共有キャッシュに載せないのが基本。
  • Cache-Control: private, no-store も検討（セキュリティ最優先）。
  • ユーザー別キャッシュが必要なら、キーにユーザーIDを含め、Vary: Authorization で下流に意図を示す。

要点まとめ

  • 認証つきはprivate、できればno-store。
  • どうしてもキャッシュするならユーザー単位で分離。

9. 圧縮（GZip）とヘッダ設計

  • GZipMiddleware(minimum_size=600) で本文が大きい時だけ圧縮。
  • JSONは圧縮効率が高く、回線の遅さに効きます。
  • ETagは圧縮後の本文を基に生成すると、実際に配るバイト列と一致します（本記事の実装はこの形）。

要点まとめ

  • 圧縮はセットで入れる。転送時間の短縮が体感速度に直結。
  • ETagは「何に対するタグか」をチームで統一。

10. よくある落とし穴と回避策

症状 原因 回避策
いつまでも古いデータが出る 無効化漏れ／長すぎるTTL タグ無効化を徹底、TTLは短めから調整
304にならない ETag不一致 同一レスポンスで必ず同じETag。圧縮の前後に注意
ユーザーが混ざる 共有キャッシュに載ってしまった Cache-Control: private、必要なら no-store
レート制限が効かない バケットキー設計ミス IP/ユーザー/APIキーなど要件に応じて切替
Redisに依存しすぎ 障害時に全体遅延 フォールバック（Redis無効時は直計算）を実装

要点まとめ

  • 「古い」「混ざる」「効かない」の三大落とし穴をキー設計・タグ・ヘッダで回避。

11. テストの考え方（信頼できるキャッシュへ）

  1. 機能テスト：
    • 初回→ミス→再取得の一連の流れで、miss/hitが切り替わることを確認。
    • If-None-Match を送って304になることを検証。
  2. 無効化テスト：
    • 投稿更新→該当タグ無効化→再取得で新データが返るか。
  3. 負荷テスト：
    • 一覧APIに対してスパイクをかけ、Redis命中率とバックエンド負荷を観察。
  4. フォールバック：
    • Redis接続を切ってもアプリは動く（直計算）ことを確認。
  5. レート制限：
    • 規定回数を超えると429になるか／ウィンドウ切り替わりで回復するか。

要点まとめ

  • 機能・無効化・負荷・フォールバック・制限の5本柱で安心♡

12. ミニサンプル：詳細APIにETagだけ足す（最短の一歩）

from fastapi import Request, Response, HTTPException
from app.utils.http_cache import conditional_response

POSTS = {1: {"id": 1, "title": "hello"}, 2: {"id": 2, "title": "world"}}

@app.get("/posts/{pid}")
async def get_post(pid: int, request: Request, response: Response):
    post = POSTS.get(pid)
    if not post:
        raise HTTPException(404)
    body = json.dumps(post).encode()
    conditional_response(request, response, body)
    response.headers["Cache-Control"] = "public, max-age=60"
    return Response(content=response.body, media_type="application/json", headers=response.headers)

ここから開始して、効果を感じたらRedisを追加しましょう。

13. 段階的な導入ロードマップ

  1. Step 1：詳細・一覧にETag＋Cache-Controlを追加。
  2. Step 2：読み取り多い一覧にRedisキャッシュ（60秒）を導入。
  3. Step 3：カテゴリやIDごとにタグ無効化を導入。
  4. Step 4：レート制限とGZipで守りと帯域最適化。
  5. Step 5：負荷計測→TTL/キー/タグの最適化、フォールバック確認。

要点まとめ

  • 小さく始めて徐々に拡張。いきなり複雑にしないのが成功のコツです。

14. 読者別インパクト（より具体的に）

  • 個人開発者：ETag実装だけでも大きな体感改善。Redis追加でバックエンドの負荷が目に見えて減ります。
  • 小規模チーム：タグ無効化とレート制限で運用事故が減り、夜間対応の頻度が下がります。
  • 成長中SaaS：CDNとHTTPキャッシュが噛み合うと帯域コストが削減。ピーク時の可用性が上がります。

まとめ（今日から“速くて軽い”APIへ♡）

  • HTTPキャッシュ（ETag/Cache-Control）はまずやる価値が大。実装コストが低く、すぐ効きます。
  • Redisキャッシュは効果の大きい箇所（一覧や外部API呼び出し）から導入。キーはURL＋クエリ＋ユーザー単位で設計し、タグ無効化で整合性を保ちましょう。
  • レート制限とGZip圧縮を併せて入れると、守りと帯域の最適化が進みます。
  • まずはこの記事の最小サンプルを試し、メトリクスを見ながらTTLとタグを磨いてください。小さな改善の積み重ねが、速くて壊れにくいAPIを育てます。わたしもずっと応援していますね♡

投稿者 greeden

関連投稿

システム開発

【Laravel×Blade×Alpine.js】アクセシブルなインタラクティブUIパターン実装大全――モーダル／ドロップダウン／タブ／アコーディオンを“正しく”作る

greeden
システム開発

【徹底解説】Qwen3-Next——“アクティブ3B”の超効率MoE×超ロング文脈で再設計された次世代Qwen／前世代Qwen3との違い、ChatGPT（GPT-5）・Gemini 2.5・Claude 3.7・DeepSeekとの比較まで

greeden
システム開発

【完全解説】Seedream 4.0——ByteDanceの次世代画像生成・編集“統合”モデルを仕事で活かす：4K生成・一括入出力・知識駆動の編集まで（実務サンプルつき）

greeden

コメントを残す

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です

日本語が含まれない投稿は無視されますのでご注意ください。（スパム対策）

見逃しています

システム開発

【Laravel×Blade×Alpine.js】アクセシブルなインタラクティブUIパターン実装大全――モーダル／ドロップダウン／タブ／アコーディオンを“正しく”作る

システム開発

爆速＆堅牢！FastAPIのキャッシュ戦略完全ガイド――Redis・HTTPキャッシュ・ETag・レート制限・圧縮まで

Business

【世界の9月16日総まとめ】FRB利下げ観測が世界市場を牽引／ガザ市で地上作戦拡大／米中はTikTok枠組み合意へ接近—金は史上高値、エネルギーは地政学で高止まり

Web Accessibility

ARIA実践ガイド：最小ARIAで堅牢なUIをつくる——APG準拠のコンポーネント設計と落とし穴