FastAPIのキャッシュ設計ガイド：Redis、ETag、レート制限、GZipの実装手順

FastAPIのAPIが遅くなったとき、Redisを置くだけでは原因ごとの対策を選べません。

キャッシュとは、取得や計算に時間がかかる結果を一時保存し、同じ処理の繰り返しを減らす仕組みです。

HTTPキャッシュはブラウザやCDNとの再転送を減らし、Redisによるアプリケーションキャッシュはデータベース照会や外部API呼び出しなどの再実行を減らします。

この記事では、両者の役割を分けたうえで、ETag、無効化、認証応答、レート制限、GZipを段階的に組み合わせます。

キャッシュを置く三つの層

キャッシュは、どこに置くかによって減らせる処理と管理上の注意が変わります。

TTLは、キャッシュを保持する有効期間です。

インバリデーションは、元データの更新に合わせて古いキャッシュを無効にする処理です。

TTLだけに任せると期限まで古い結果が残り、無効化だけに頼ると処理漏れが古い結果につながります。

実装では、許容できる鮮度に合わせたTTLと、更新処理に連動する無効化を組み合わせます。

実装前に決める四つのこと

1. 許容できる鮮度：更新から何秒まで古い結果を返せるかを、エンドポイントごとに決めます。
2. キーに含める入力：パス、クエリ、言語、権限、利用者など、レスポンスを変える条件を洗い出します。
3. 無効化の起点：POST、PUT、PATCH、DELETEのどの処理が、どのキャッシュへ影響するかを対応づけます。
4. 障害時の動作：Redisを使えないときに直計算へ戻すか、処理を止めるかを用途別に決めます。

同じURLでも利用者によって内容が変わるなら、公開キャッシュとして扱うことはできません。

まずレスポンスの共有範囲を決め、その後でTTLやRedisの構成を選びます。

サンプルの依存関係と設定

以下の例では、非同期Redisクライアントを使います。

fastapi
uvicorn[standard]
redis>=5.0
pydantic-settings

# ​app/core/settings.py
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

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

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

60秒は説明用の初期値です。

本番では、データの更新頻度と古い結果を許容できる時間を測って調整します。

Redis接続を一か所で管理する

Redisの接続は、アプリケーションの開始時に作成し、終了時に閉じます。

共有リソースの扱いを詳しく確認したい場合は、FastAPIのlifespanによる共有リソース管理も参照してください。

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

_client: ​redis_async.Redis ​| ​None ​= ​None

async ​def ​connect_redis() ​-> ​None:
 ​ ​ ​ ​global ​_client
 ​ ​ ​ ​settings ​= ​get_settings()
 ​ ​ ​ ​_client ​= ​redis_async.from_url(
 ​ ​ ​ ​ ​ ​ ​ ​settings.redis_url,
 ​ ​ ​ ​ ​ ​ ​ ​decode_responses=False,
 ​ ​ ​ ​)

def ​get_redis() ​-> ​redis_async.Redis ​| ​None:
 ​ ​ ​ ​return ​_client

async ​def ​disconnect_redis() ​-> ​None:
 ​ ​ ​ ​global ​_client
 ​ ​ ​ ​if ​_client ​is ​not ​None:
 ​ ​ ​ ​ ​ ​ ​ ​await ​_client.aclose()
 ​ ​ ​ ​ ​ ​ ​ ​_client ​= ​None

# ​app/main.py
from ​contextlib ​import ​asynccontextmanager
from ​fastapi ​import ​FastAPI
from ​app.core.redis_client ​import ​connect_redis, ​disconnect_redis

@asynccontextmanager
async ​def ​lifespan(app: ​FastAPI):
 ​ ​ ​ ​await ​connect_redis()
 ​ ​ ​ ​try:
 ​ ​ ​ ​ ​ ​ ​ ​yield
 ​ ​ ​ ​finally:
 ​ ​ ​ ​ ​ ​ ​ ​await ​disconnect_redis()

app ​= ​FastAPI(title="FastAPI ​Cache ​Demo", ​lifespan=lifespan)

各モジュールはRedis変数そのものを取り込まず、利用時にget_redis()を呼びます。

この形なら、起動時に設定された現在のクライアントを参照できます。

Redisキャッシュのキーと無効化

キャッシュキーは、レスポンスを決める入力を表す必要があります。

次の例は、パスと並べ替えたクエリ文字列をハッシュ化し、用途を表す接頭辞を付けます。

# ​app/utils/app_cache.py
import ​hashlib
from ​urllib.parse ​import ​urlencode
from ​fastapi ​import ​Request
from ​app.core.redis_client ​import ​get_redis
from ​app.core.settings ​import ​get_settings

def ​cache_key_from_request(request: ​Request, ​prefix: ​str) ​-> ​str:
 ​ ​ ​ ​query ​= ​urlencode(sorted(request.query_params.multi_items()))
 ​ ​ ​ ​raw ​= ​f"{request.url.path}?{query}"
 ​ ​ ​ ​digest ​= ​hashlib.sha256(raw.encode("utf-8")).hexdigest()
 ​ ​ ​ ​return ​f"{prefix}:{digest}"

async ​def ​cache_get(key: ​str) ​-> ​bytes ​| ​None:
 ​ ​ ​ ​client ​= ​get_redis()
 ​ ​ ​ ​if ​client ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​return ​None
 ​ ​ ​ ​return ​await ​client.get(key)

async ​def ​cache_set(key: ​str, ​value: ​bytes, ​ttl: ​int ​| ​None ​= ​None) ​-> ​None:
 ​ ​ ​ ​client ​= ​get_redis()
 ​ ​ ​ ​if ​client ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​return
 ​ ​ ​ ​settings ​= ​get_settings()
 ​ ​ ​ ​seconds ​= ​ttl ​if ​ttl ​is ​not ​None ​else ​settings.default_ttl_seconds
 ​ ​ ​ ​await ​client.set(key, ​value, ​ex=seconds)

async ​def ​tag_add(tag: ​str, ​key: ​str) ​-> ​None:
 ​ ​ ​ ​client ​= ​get_redis()
 ​ ​ ​ ​if ​client ​is ​not ​None:
 ​ ​ ​ ​ ​ ​ ​ ​await ​client.sadd(f"tag:{tag}", ​key)

async ​def ​invalidate_tags(*tags: ​str) ​-> ​int:
 ​ ​ ​ ​client ​= ​get_redis()
 ​ ​ ​ ​if ​client ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​return ​0

 ​ ​ ​ ​tag_keys ​= ​[f"tag:{tag}" ​for ​tag ​in ​tags]
 ​ ​ ​ ​members: ​set[bytes] ​= ​set()
 ​ ​ ​ ​for ​tag_key ​in ​tag_keys:
 ​ ​ ​ ​ ​ ​ ​ ​members.update(await ​client.smembers(tag_key))

 ​ ​ ​ ​if ​members:
 ​ ​ ​ ​ ​ ​ ​ ​await ​client.delete(*members)
 ​ ​ ​ ​if ​tag_keys:
 ​ ​ ​ ​ ​ ​ ​ ​await ​client.delete(*tag_keys)
 ​ ​ ​ ​return ​len(members)

接頭辞にv1:のようなバージョンを含めると、レスポンス形式を変えたときに旧キーと新キーを分けられます。

認証済み利用者ごとに結果が違う場合は、利用者IDなどの分離条件もキーへ含めます。

ただし、個人情報そのものをキーへ入れる必要はありません。

ETagで条件付きGETを返す

ETagは、レスポンスの版を識別する値です。

クライアントが次のGETでIf-None-Matchを送り、現在のETagと一致すれば、サーバーは本文のない304 ​Not ​Modifiedを返せます。

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

def ​make_etag(payload: ​bytes) ​-> ​str:
 ​ ​ ​ ​digest ​= ​hashlib.sha256(payload).hexdigest()
 ​ ​ ​ ​return ​f'"{digest}"'

def ​conditional_json_response(
 ​ ​ ​ ​request: ​Request,
 ​ ​ ​ ​body: ​bytes,
 ​ ​ ​ ​cache_control: ​str,
) ​-> ​Response:
 ​ ​ ​ ​etag ​= ​make_etag(body)
 ​ ​ ​ ​headers ​= ​{
 ​ ​ ​ ​ ​ ​ ​ ​"ETag": ​etag,
 ​ ​ ​ ​ ​ ​ ​ ​"Cache-Control": ​cache_control,
 ​ ​ ​ ​}

 ​ ​ ​ ​if ​request.headers.get("if-none-match") ​== ​etag:
 ​ ​ ​ ​ ​ ​ ​ ​return ​Response(status_code=304, ​headers=headers)

 ​ ​ ​ ​return ​Response(
 ​ ​ ​ ​ ​ ​ ​ ​content=body,
 ​ ​ ​ ​ ​ ​ ​ ​media_type="application/json",
 ​ ​ ​ ​ ​ ​ ​ ​headers=headers,
 ​ ​ ​ ​)

304用のResponseをそのまま返すため、ステータスコードが後から200へ戻ることはありません。

この最小例は単一のETagを完全一致で比較します。

複数値、弱いETag、ワイルドカードまで扱う実装では、HTTP仕様に沿った比較処理が別途必要です。

一覧APIへRedisとETagを組み込む

次の例は、公開一覧APIの結果をRedisへ保存し、同じ本文からETagを作ります。

# ​app/main.pyのルーター部分
import ​json
from ​fastapi ​import ​Query, ​Request, ​Response
from ​app.utils.app_cache ​import ​cache_get, ​cache_key_from_request, ​cache_set, ​tag_add
from ​app.utils.http_cache ​import ​conditional_json_response

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,
 ​ ​ ​ ​limit: ​int ​= ​Query(20, ​ge=1, ​le=100),
 ​ ​ ​ ​offset: ​int ​= ​Query(0, ​ge=0),
 ​ ​ ​ ​category: ​str ​| ​None ​= ​Query(None),
) ​-> ​Response:
 ​ ​ ​ ​key ​= ​cache_key_from_request(request, ​prefix="v1:posts")
 ​ ​ ​ ​cached ​= ​await ​cache_get(key)
 ​ ​ ​ ​if ​cached ​is ​not ​None:
 ​ ​ ​ ​ ​ ​ ​ ​return ​conditional_json_response(
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​request,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​cached,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​cache_control="public, ​max-age=30",
 ​ ​ ​ ​ ​ ​ ​ ​)

 ​ ​ ​ ​items ​= ​[
 ​ ​ ​ ​ ​ ​ ​ ​post ​for ​post ​in ​FAKE_POSTS
 ​ ​ ​ ​ ​ ​ ​ ​if ​category ​is ​None ​or ​post["category"] ​== ​category
 ​ ​ ​ ​]
 ​ ​ ​ ​page ​= ​items[offset:offset ​+ ​limit]
 ​ ​ ​ ​body ​= ​json.dumps(
 ​ ​ ​ ​ ​ ​ ​ ​{
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"total": ​len(items),
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"limit": ​limit,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"offset": ​offset,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​"items": ​page,
 ​ ​ ​ ​ ​ ​ ​ ​},
 ​ ​ ​ ​ ​ ​ ​ ​ensure_ascii=False,
 ​ ​ ​ ​ ​ ​ ​ ​separators=(",", ​":"),
 ​ ​ ​ ​).encode("utf-8")

 ​ ​ ​ ​await ​cache_set(key, ​body, ​ttl=60)
 ​ ​ ​ ​await ​tag_add("posts:all", ​key)
 ​ ​ ​ ​if ​category ​is ​not ​None:
 ​ ​ ​ ​ ​ ​ ​ ​await ​tag_add(f"posts:cat:{category}", ​key)

 ​ ​ ​ ​return ​conditional_json_response(
 ​ ​ ​ ​ ​ ​ ​ ​request,
 ​ ​ ​ ​ ​ ​ ​ ​body,
 ​ ​ ​ ​ ​ ​ ​ ​cache_control="public, ​max-age=30",
 ​ ​ ​ ​)

posts:allは絞り込みのない一覧も含めて無効化するためのタグです。

カテゴリタグだけを登録すると、投稿更新後も全件一覧の古いキャッシュが残るため、影響範囲を両方へ登録します。

書き込み時に関連キャッシュを消す

元データを書き換えた処理の中で、影響するタグを無効化します。

from ​app.utils.app_cache ​import ​invalidate_tags

@app.post("/admin/posts/invalidate")
async ​def ​invalidate_posts(category: ​str):
 ​ ​ ​ ​deleted ​= ​await ​invalidate_tags(
 ​ ​ ​ ​ ​ ​ ​ ​"posts:all",
 ​ ​ ​ ​ ​ ​ ​ ​f"posts:cat:{category}",
 ​ ​ ​ ​)
 ​ ​ ​ ​return ​{"invalidated": ​deleted}

このコードは無効化の流れだけを示しており、管理者認証を省略しています。

本番で無効化用エンドポイントを公開するときは、更新権限を確認する仕組みと組み合わせます。

対象が多く、書き込み処理から分離したい場合は、FastAPIの処理をCeleryとRedisへ分離する設計が参考になります。

認証応答を公開キャッシュへ載せない

認証情報を含むレスポンスへpublicを付けると、共有キャッシュで別の利用者へ再利用される設計になり得ます。

• 全利用者で同じ公開情報には、要件に合うpublicとmax-ageを設定します。
• 利用者ごとの情報をブラウザ内だけで再利用する場合は、privateを検討します。
• 保存させたくない応答には、no-storeを設定します。

アプリケーション層で利用者別レスポンスを保存する場合も、利用者を分ける値をキャッシュキーへ含めます。

Vary: ​Authorizationを使う場合でも、Cache-Controlによる共有可否の指定は省けません。

固定窓レート制限でAPIを保護する

原文のカウンター実装は、一定時間ごとに回数を数え直す固定窓方式です。

トークンを継続的に補充するトークンバケット方式とは動作が異なるため、名称を分けます。

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

async ​def ​enforce_fixed_window(
 ​ ​ ​ ​client,
 ​ ​ ​ ​bucket_key: ​str,
 ​ ​ ​ ​limit: ​int ​= ​60,
 ​ ​ ​ ​window_seconds: ​int ​= ​60,
) ​-> ​None:
 ​ ​ ​ ​window ​= ​int(time.time()) ​// ​window_seconds
 ​ ​ ​ ​key ​= ​f"ratelimit:{bucket_key}:{window}"
 ​ ​ ​ ​current ​= ​await ​client.incr(key)

 ​ ​ ​ ​if ​current ​== ​1:
 ​ ​ ​ ​ ​ ​ ​ ​await ​client.expire(key, ​window_seconds)

 ​ ​ ​ ​if ​current ​> ​limit:
 ​ ​ ​ ​ ​ ​ ​ ​raise ​HTTPException(
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​status_code=status.HTTP_429_TOO_MANY_REQUESTS,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​detail="Rate ​limit ​exceeded. ​Please ​try ​again ​later.",
 ​ ​ ​ ​ ​ ​ ​ ​)

バケットキーは、IPアドレス、利用者ID、APIキーなど、制限したい単位に合わせます。

Redis障害時に制限を通すか、503などで止めるかは、保護対象と業務要件に応じて事前に決めます。

GZipは転送量対策として分けて考える

GZipはキャッシュではなく、レスポンス本文を圧縮して転送量を抑える仕組みです。

from ​fastapi.middleware.gzip ​import ​GZipMiddleware

app.add_middleware(GZipMiddleware, ​minimum_size=600)

minimum_size=600は原文の例であり、実際のレスポンスサイズと処理負荷を測って調整します。

この記事のETag例は、GZipミドルウェアへ渡す前のJSON本文から値を生成します。

圧縮後の本文を基にしているわけではないため、CDNやプロキシを含む本番経路では、実際のETag、Content-Encoding、Varyを確認します。

動作確認のチェックリスト

1. 初回のGETが200と本文を返し、ETagとCache-Controlを含むことを確認します。
2. 同じETagをIf-None-Matchへ指定したGETが、本文なしの304を返すことを確認します。
3. クエリの順序が違っても同じ条件なら同じRedisキーになり、値が違えば別キーになることを確認します。
4. 投稿更新後に全件一覧と該当カテゴリのキャッシュが消え、新しい本文が返ることを確認します。
5. 二人の利用者でレスポンスが混ざらず、認証応答が共有キャッシュへ保存されないことを確認します。
6. Redisを停止し、通常の取得処理とレート制限が決めた方針どおりに動くことを確認します。
7. 固定窓内で上限を超えると429になり、次の窓で回復することを確認します。
8. 一定サイズを超える本文でGZipが適用され、ETagによる再検証も成立することを確認します。

よくある不具合の切り分け

段階的な導入手順

1. 対象APIの応答時間、呼び出し回数、バックエンド処理を測ります。
2. 変更の少ない公開GETで、ETagとCache-Controlによる再検証を試します。
3. 繰り返し取得される一覧や集計へRedisを導入し、キーとTTLを記録します。
4. 同じ変更で影響を受けるキャッシュをタグにまとめ、書き込み処理から無効化します。
5. 認証応答の分離、Redis障害時の動作、固定窓レート制限をテストします。
6. 大きなレスポンスへGZipを適用し、計測結果を見ながら設定値を調整します。

FastAPIのキャッシュ設計を運用へつなげる

HTTPキャッシュとRedisは、同じ高速化機能ではありません。

HTTP層では再転送を減らし、アプリケーション層では取得や計算の繰り返しを減らします。

どちらも、公開範囲、キー、TTL、無効化、障害時の動作まで決めて初めて運用できます。

まず一つの公開GETで200と304を確認し、その後にRedisと無効化を加えると、問題が起きた層を切り分けやすくなります。