Flaskのパフォーマンス最適化実践：Flask-Caching（Redis）でキャッシュ、Flask-Limiterでレート制限、N+1とコネクションプールまで

導入：パフォーマンス最適化は「速くすること」ではなく「計測すること」から始まる

Flask アプリが「遅い」と言われたとき、最も多い失敗は 計測せずに手を動かすことです。リスト内包表記を map に書き換える、文字列結合を join にする——こうした micro 最適化は、ほとんどの Web アプリでは誤差です。本番の Flask アプリで体感速度を決めるのは、ほぼ例外なく データベース（N+1・索引欠落・コネクションプールの枯渇） と キャッシュの有無、そして ワーカー設計です。ここを外したまま小手先を磨いても、p95 は 1 ミリも縮みません。

本記事は、Flask 本番運用ガイド（ピラー） のパフォーマンス領域を本番品質まで深掘りするスポークです。扱うのは Flask 3.1 系（現行安定版）を前提に、公式ドキュメント最新版に忠実な最適化設計だけです。具体的には、(1) 計測のマインドセット、(2) Flask-Caching（Redis）によるキャッシュ、(3) Flask-Limiter によるレート制限、(4) DB パフォーマンス（N+1・プール・ページネーション）、(5) レスポンス層（gzip・async）、(6) これらがそのまま効いてくるコスト効率、までを実コードで通します。

筆者は、経済産業大臣賞を受賞した B2B SaaS のバックエンドを Python / Flask / SQLAlchemy / PostgreSQL で設計・実装し、API Gateway → ALB → ECS(Fargate) 上で 221 エンドポイントを本番運用してきました。読み取りが支配的な業務 SaaS で「速さ」と「コスト」を両立させるために必要だったのは、派手なアルゴリズムではなく、計測 → DB → キャッシュ → レート制限 → ワーカーという地味な順序を守ることでした。本記事はその順序をそのまま辿ります。

💡 この記事で扱うバージョン：Flask 3.1 系を前提に、Flask-Caching 2.4.0・Flask-Limiter 4.1.1 を使います。いずれも Flask の依存ではなく別途インストールする拡張です（pip install Flask-Caching==2.4.0 Flask-Limiter==4.1.1 redis）。ストレージには Redis を前提とします。本稿に出てくる数値（レイテンシ・接続数）はすべて説明のための例であり、特定環境のベンチマーク値ではありません。

1. 計測ファースト：どこが遅いのかを先に知る

1.1 「Flask は同期」という前提を体に入れる

最適化の前に、Flask の実行モデルを正確に押さえます。Flask は WSGI（同期）フレームワークで、各リクエストは 1 つの Gunicorn ワーカーを最後まで占有します。公式も async def ビューについて「各リクエストは依然 1 ワーカーを占有する。async にしても同時に捌けるリクエスト数は変わらない」と明言しています。

ここから、Flask のスループットを決めるのが何かが見えてきます。

• 同時に捌けるリクエスト数 ≒ ワーカー数（同期 sync ワーカーの場合）。
• 1 リクエストの処理時間が長い（＝ DB や外部 API を待っている）ほど、その間ワーカーは塞がり、他のリクエストが待たされる。
• だから 「1 リクエストの処理時間を縮める」（キャッシュ・DB 最適化） ことと、「ワーカーを適切に並べる」（デプロイ設計） ことは、同じコインの裏表です。

ワーカー数の設計（CPU × 2 を出発点に、sync か gevent か、メモリとのトレードオフ）は 本番デプロイガイド に集約しています。本記事は「1 リクエストを速くする／無駄なリクエストを弾く」側を担当します。両者はセットで効きます。

1.2 何を計測するか：平均ではなく p95 / p99

レイテンシは平均で見てはいけません。平均は外れ値に鈍感で、「ほとんどのユーザーは速いが、一部が致命的に遅い」状態を隠してしまいます。見るべきはパーセンタイルです。

p99 が悪いエンドポイントは、そのワーカーを長時間占有します。同期 Flask では、遅いエンドポイント 1 つが他の全リクエストの待ち時間に波及するため、「最も遅い一部」を潰すことがスループット全体に効きます。

1.3 どう計測するか：3 つの粒度

計測は粗いものから細かいものへ、3 段階で十分です。

(a) エンドポイント単位の所要時間ログ——最初の一歩。before_request / after_request で各リクエストの所要時間を記録します。

import time

from flask import g, request, current_app


@app.before_request
def _start_timer():
    g._t0 = time.perf_counter()


@app.after_request
def _log_latency(response):
    elapsed_ms = (time.perf_counter() - g._t0) * 1000
    # 構造化ログに乗せる（リクエストID付与は可観測性ガイド参照）
    current_app.logger.info(
        "request_latency",
        extra={"path": request.path, "method": request.method,
               "status": response.status_code, "elapsed_ms": round(elapsed_ms, 1)},
    )
    return response

このログを集計すれば「どのパスが p95/p99 を引っ張っているか」が分かります。ログの構造化・リクエスト ID の相関は エラー処理・可観測性ガイド に委ねます。Gunicorn のアクセスログにも所要時間を出せます。

(b) SQL クエリの可視化——遅い原因の大半は DB です。SQLAlchemy のエンジンログを開発時だけ ON にすると、1 リクエストで何本のクエリが飛んでいるかが見えます（N+1 がここで露見します）。詳細は §4 と Flask-SQLAlchemy 実践ガイド。

(c) プロファイラでホットスポット特定——どうしても CPU バウンドな処理が疑わしいときだけ、cProfile や Werkzeug の ProfilerMiddleware でビュー内のホットスポットを特定します。ただしここに来る前に DB とキャッシュを疑うのが順序です。

⚠️ 推測で最適化しない。「たぶんここが遅い」で書き換えるのは、技術的負債を増やすだけの賭けです。Knuth の「早すぎる最適化は諸悪の根源」は今も正しい。必ず計測でボトルネックを特定し、効果を計測で確認する——このループを回せる仕組み（所要時間ログ・SQL ログ）を、最適化に着手する前に用意してください。これが最も投資対効果の高い一手です。

💡 筆者の経験：木材流通 SaaS で「一覧画面が遅い」という報告を受けたとき、最初に見たのは SQL ログでした。原因はアルゴリズムではなく、明細を 1 件ずつ引く N+1（§4.1）と、マスタデータを毎回 DB から読んでいたこと（§2 のキャッシュ対象）でした。計測せずに「Python が遅い」と決めつけていたら、永遠に直らなかった問題です。

2. Flask-Caching：読み取りをメモリ速度にする

2.1 何をキャッシュすべきか（そして、すべきでないか）

キャッシュは万能薬ではありません。間違った対象をキャッシュすると、古いデータを配り続ける障害になります。キャッシュに向くのは、次の 3 条件を満たすデータです。

1. 読み取りが多い（read-heavy）：何度も同じものが要求される。
2. 生成コストが高い（expensive）：DB の重い集計、外部 API 呼び出し、複雑なレンダリング。
3. 多少の鮮度のズレを許容できる（tolerant of staleness）：数十秒〜数分古くても業務上問題ない。

逆に、書き込み直後に即座に正確であることが必須なデータ（決済残高、在庫の確定値）は、安易にキャッシュしてはいけません。そして「遅い処理を非同期で逃がす」べきケース——たとえば重い帳票生成やメール送信——は、キャッシュではなくバックグラウンドジョブの領分です。キャッシュは read-heavy 用、Celery は write/slow work 用と棲み分けます（Flask + Celery + Redis バックグラウンドタスクガイド）。

2.2 セットアップ：RedisCache をファクトリで束ねる

Flask-Caching は拡張なので、ピラー §3 の init_app パターンに従います。Cache() を裸で生成し、init_app で束縛します。

# extensions.py — どのアプリにも束縛されていない「裸」の拡張
from flask_caching import Cache

cache = Cache()

# __init__.py（create_app 内・抜粋）
from .extensions import cache


def create_app(test_config=None):
    app = Flask(__name__, instance_relative_config=True)
    # ...設定読み込み...
    cache.init_app(app, config={
        "CACHE_TYPE": "RedisCache",
        "CACHE_REDIS_URL": "redis://localhost:6379/0",
    })
    return app

Cache(app) のように生成時に渡すこともできますが、ファクトリ構成では init_app 一択です。CACHE_TYPE には用途別に複数の値があります。

⚠️ SimpleCache を本番で使わない。SimpleCache はプロセスローカルのメモリ辞書です。Gunicorn は複数ワーカー（＝複数プロセス）を立てるため、ワーカー A がキャッシュした値はワーカー B からは見えません。さらにワーカーが再起動すれば消えます。これは Flask-Limiter の memory:// 問題（§3.3）とまったく同じ構造の罠です。マルチワーカー/マルチコンテナの本番では、必ず Redis のような共有バックエンドを使ってください。

2.3 @cache.cached：ビュー単位でキャッシュする

最も手軽なのが、ビュー関数のレスポンスまるごとをキャッシュする @cache.cached です。公式の最小形がこれです。

@app.route("/")
@cache.cached(timeout=50)
def index():
    return render_template('index.html')

timeout=50 は 50 秒の TTL（生存時間）です。同じパスへの 2 回目以降のリクエストは、ビューを実行せず Redis から返ります。

デコレータの順序に注意：@app.route が外側、@cache.cached が内側（ビュー関数に近い側）です。逆にすると正しく機能しません。

@cache.cached の主要パラメータは次の通りです。

query_string=True は地味に重要です。既定では request.path だけがキーなので、/search?q=flask と /search?q=django が同じキャッシュを共有してしまいます。これを query_string=True にすると、クエリパラメータごとに別キャッシュになり、しかもパラメータの順序（?a=1&b=2 と ?b=2&a=1）を正規化して同一視してくれるため、キャッシュが無駄に分裂しません。

@app.route("/search")
@cache.cached(timeout=60, query_string=True)
def search():
    q = request.args.get("q", "")
    return jsonify(results=run_expensive_search(q))

⚠️ ユーザー固有のレスポンスを @cache.cached でキャッシュしない。既定のキーは request.path（＋クエリ）だけで、ログイン中のユーザーを区別しません。認証済みユーザーのダッシュボードのような「人によって中身が違う」レスポンスを素の @cache.cached でキャッシュすると、A さんのデータが B さんに見える重大な情報漏えいになります。ユーザー別にするなら key_prefix を callable にしてユーザー ID をキーに織り込むか、そもそもこの層でキャッシュしない判断をします。unless で「認証済みならバイパス」も有効です。

2.4 @cache.memoize：関数を「引数ごと」にキャッシュする

ビューではなく、内部の重い関数の戻り値をキャッシュしたいなら @cache.memoize です。@cache.cached との決定的な違いは、引数の値ごとに別々にキャッシュする点です。

@cache.memoize(50)
def get_product_summary(product_id: int) -> dict:
    # 重い集計クエリ。同じ product_id なら 50 秒間は再計算しない
    return expensive_aggregate(product_id)

get_product_summary(1) と get_product_summary(2) は別のキャッシュエントリになります。@cache.cached がリクエストパスを見るのに対し、@cache.memoize は関数の引数を見るので、「同じ計算を同じ引数で何度もやる」典型的な重複作業に効きます。マスタデータの参照、設定値の解決、外部 API のレスポンス変換などが好例です。

2.5 キャッシュ無効化：本当に難しいのはここ

「コンピュータサイエンスの 2 大難問はキャッシュ無効化と命名だ」という有名な箴言の通り、キャッシュの本当の難しさは、いつ・どう古い値を捨てるかです。戦略は大きく 2 つ。

(a) TTL（時間ベース）の自動失効——timeout で「N 秒後に勝手に消える」ようにする最もシンプルな方式。鮮度のズレが業務上許容できる読み取りデータに最適です。実装が単純で、無効化漏れの事故が起きにくいのが利点。欠点は「更新したのに最大 N 秒は古い値が出る」こと。

(b) 明示的な無効化（イベントベース）——データを更新したタイミングで、対応するキャッシュを自分で消す方式。@cache.memoize した関数は、その関数とキー引数を指定して無効化できます。cache.clear() は全キャッシュをクリアします（アプリコンテキスト内で実行）。

def update_product(product_id: int, payload: dict) -> None:
    db.session.execute(...)  # 実データを更新
    db.session.commit()
    # 更新したものに対応するキャッシュだけを失効させる
    cache.delete_memoized(get_product_summary, product_id)

# 全キャッシュをまとめてクリア（運用バッチ・デプロイ後など）
with app.app_context():
    cache.clear()

💡 無効化戦略の選び方：迷ったら TTL を既定にしてください。明示的無効化は「更新箇所を一つでも書き忘れると古いデータを配り続ける」という、見つけにくいバグを生みます。「更新が頻繁でなく、数十秒の鮮度ズレが許容できる」なら短めの TTL（例：30〜60 秒）だけで運用が回ります。「更新が即座に反映されなければならず、かつ更新箇所が明確に特定できる」場合に限り、明示的無効化を足す——これが事故りにくい順序です。TTL と明示的無効化は併用もできます（短めの TTL を保険にしつつ、更新時に明示削除）。

2.6 キャッシュスタンピード：失効の瞬間に殺到する

TTL キャッシュには見落としがちな落とし穴があります。人気のキーが失効したまさにその瞬間、それを待っていた大量のリクエストが一斉に「キャッシュミス」となり、全員が同じ重いクエリを叩いて DB に殺到する——これが キャッシュスタンピード（thundering herd / dog-piling） です。読み取りを軽くするはずのキャッシュが、失効の瞬間に最悪の負荷スパイクを生みます。

対策の方向性は 3 つ。

• TTL をずらす（ジッター）：エントリごとに TTL に乱数の揺らぎを足し、失効タイミングを分散させる。同時失効を防ぐ最も手軽な手。
• 事前再生成（cache warming）：失効前にバックグラウンドジョブ（Celery）で再計算しておき、ユーザーリクエストがミスに当たらないようにする。
• ロック（single-flight）：失効時に最初の 1 リクエストだけが再計算し、他はそれを待つ。実装は複雑になるため、本当に重いキーに限定する。

⚠️ スタンピードは「成功するほど顕在化する」。トラフィックが小さいうちは表面化せず、アクセスが増えてキャッシュヒット率が上がるほど、失効時のスパイクが鋭くなります。**最重要・最高コストのキーほど、失効が同時に来ない設計（ジッターや事前再生成）**を初めから入れておくのが安全です。

2.7 HTTP キャッシュ：アプリの「外」でもう一段省く

Flask-Caching が**サーバー側（アプリ内）**のキャッシュなのに対し、HTTP キャッシュはクライアント・CDN・プロキシに「再取得しなくていい」と伝える、補完的なレイヤーです。両方を併用すると、そもそもアプリにリクエストが届く回数自体を減らせます。

• Cache-Control：public, max-age=300 のように、クライアント/CDN に「300 秒は再取得不要」と伝える。静的に近いレスポンスに有効。
• ETag / Last-Modified と条件付きリクエスト：レスポンスに ETag（内容のハッシュ等）を付けておくと、クライアントは次回 If-None-Match で問い合わせ、内容が変わっていなければサーバーは 304 Not Modified（ボディなし）を返せる。帯域とシリアライズコストを節約できます。

from flask import request, make_response


@app.route("/api/catalog")
def catalog():
    data = get_catalog()                       # 重い集計（Flask-Cachingでさらに保護してもよい）
    resp = make_response(jsonify(data))
    resp.headers["Cache-Control"] = "public, max-age=300"
    resp.set_etag(compute_etag(data))          # 内容に基づくETag
    return resp.make_conditional(request)      # If-None-Match一致なら 304 を返す

make_conditional(request) が If-None-Match / If-Modified-Since を見て、一致すれば自動で 304 に変換してくれます。CDN（CloudFront 等）を前段に置くなら、Cache-Control をオリジン（Flask）から正しく出すことで、CDN がエッジでキャッシュし、アプリの負荷をさらに下げられます。

3. Flask-Limiter：レート制限で守る・公平にする・コストを抑える

3.1 なぜレート制限が必要か

レート制限（Rate Limiting）は「単位時間あたりのリクエスト数に上限を設ける」仕組みで、パフォーマンスとセキュリティの両方に効きます。動機は 4 つ。

1. 濫用・スクレイピング対策：1 クライアントが API を叩き続けてリソースを独占するのを防ぐ。
2. コスト制御：従量課金の下流（外部 API・LLM・DB）を、過剰呼び出しから守る。レート制限は直接コストに効く。
3. 公平性（fairness）：一部の重いクライアントが、他の全ユーザーの体験を劣化させないようにする。
4. ブルートフォース防御：ログインや OTP 検証エンドポイントへの総当たり攻撃を、試行回数で頭打ちにする（認証は Flask 認証ガイド）。

そして、レート制限は DoS/DDoS 防御の一枚目のアプリ層でもあります。アプリの手前に WAF（Web Application Firewall）を置く多層防御の発想と組み合わせるのが本筋です（Flask セキュリティ実装ガイド）。

3.2 セットアップ：default_limits と per-route

Flask-Limiter のセットアップは公式の形をそのまま使います。

from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

app = Flask(__name__)
limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["200 per day", "50 per hour"],
    storage_uri="redis://localhost:6379",
)

ファクトリ構成なら init_app で束ねます（ピラー §3 の作法）。

# extensions.py
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(get_remote_address)

# create_app 内
limiter.init_app(app)

• default_limits：全ルートに既定で適用される上限。"200 per day"、"50 per hour" のように人間可読な文字列で書けます。
• per-route の @limiter.limit：個別ルートに上限を上書き/追加できます。
• @limiter.exempt：特定ルート（ヘルスチェックなど）をレート制限から除外。

@app.route("/api/login", methods=["POST"])
@limiter.limit("5 per minute")            # ブルートフォース対策で厳しめに
def login():
    ...


@app.route("/api/export")
@limiter.limit("1 per day")               # 重い処理は1日1回に絞る
def export():
    ...


@app.route("/health")
@limiter.exempt                           # ヘルスチェックは制限しない
def health():
    return {"status": "ok"}

上限を超えたとき、ビュー関数は呼ばれません。Flask-Limiter が 429 Too Many Requests を送出します。

3.3 最重要の落とし穴：memory:// を本番で使うな

Flask-Limiter の既定/手軽な選択肢に メモリストレージ（memory://） がありますが、公式は本番利用について明確に警告しています。

"should be used with caution in any production setup since: Each application process will have it's own storage [and] The state of the rate limits will not persist beyond the process' life-time."

訳せば、「本番では注意して使うべき。各アプリケーションプロセスが自分専用のストレージを持ち、レート制限の状態はプロセスの寿命を超えて保持されない」。

これが致命的なのは、Gunicorn が複数のワーカープロセスを立てるからです。memory:// だと、各ワーカーが独立したカウンタを持つため、

• ワーカーが 8 個あれば、50 per hour の制限が実質 50 × 8 = 400 per hour まで通ってしまう（ワーカーごとに別カウント）。
• ワーカーが再起動すれば、カウンタはリセットされる。
• ロードバランサが別ワーカーに振れば、また別のカウンタで数え直す。

つまり memory:// ではレート制限が「効いているように見えて、まったく効いていない」。これは Flask-Caching の SimpleCache（§2.2）と完全に同じ構造の罠です。マルチワーカー前提の Gunicorn 運用（本番デプロイガイド）では、必ず共有バックエンドを storage_uri に指定します。公式が挙げる選択肢は Redis（redis://host:port）・Memcached・MongoDB です。

limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["200 per day", "50 per hour"],
    storage_uri="redis://localhost:6379",   # ← 全ワーカー/全コンテナで共有
)

⚠️ 「ローカルで効いたから本番でも効く」は通用しない。開発時は単一プロセス（flask run）なので memory:// でもレート制限が効いて見えます。だが本番の Gunicorn は多ワーカー・多コンテナ。ストレージを共有 Redis にしない限り、制限はワーカー数とコンテナ数の分だけ素通りします。Flask-Caching の SimpleCache と Flask-Limiter の memory:// は、マルチプロセスを忘れた瞬間に壊れる双子の罠だと覚えてください。

3.4 キー関数：誰を単位に数えるか

「1 単位」を何にするかを決めるのがキー関数です。get_remote_address（クライアント IP 単位）が既定的ですが、要件に応じて変えます。

from flask_login import current_user

def user_or_ip():
    # 認証済みならユーザー単位、未認証ならIP単位
    if current_user.is_authenticated:
        return str(current_user.id)
    return get_remote_address()

limiter = Limiter(user_or_ip, app=app, storage_uri="redis://localhost:6379")

💡 B2B SaaS ではテナント/プラン単位が効く。木材流通 SaaS のような B2B では、IP 単位よりテナント（契約企業）単位や API キー単位でレート制限する方が、プランに応じた公平な制御ができます。「無料プランは 60 req/min、有料は 600 req/min」のような商品設計を、キー関数と動的な上限で表現できます。

3.5 プロキシ背後では ProxyFix で本物の IP を取る

get_remote_address でキーを切るとき、致命的な前提があります。Flask が ALB / nginx のようなリバースプロキシの背後にいる場合、request.remote_addr はクライアントの IP ではなくプロキシの IP になります。何も対処しないと、全クライアントが同一の「プロキシ IP」として 1 つの制限枠を共有してしまい、レート制限が無意味になります（一人が枠を使い切ると全員が 429）。

解決は Werkzeug の ProxyFix。公式の形がこれです。

from werkzeug.middleware.proxy_fix import ProxyFix

app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1)   # 前段のプロキシ段数
limiter = Limiter(get_remote_address, app=app)

x_for は X-Forwarded-For をセットするプロキシの段数です。ここを間違えると、今度はクライアントが偽の X-Forwarded-For を注入してレート制限を回避できてしまうセキュリティ問題になります。ProxyFix の段数の数え方（ALB だけなら 1、API Gateway → ALB の 2 段なら実測して 2、など）は 本番デプロイガイド で詳説しています。IP ベースのレート制限・監査ログ・アクセス制御はすべて「本物のクライアント IP が取れている」ことが前提です。

3.6 429 ハンドリングと Retry-After

上限超過時に投げられる 429 を、API として親切に返します。Retry-After ヘッダで「いつ再試行してよいか」をクライアントに伝えるのが定石です。

from flask import jsonify


@app.errorhandler(429)
def ratelimit_handler(e):
    resp = jsonify(error="rate limit exceeded", detail=str(e.description))
    resp.status_code = 429
    # Flask-Limiter は超過時のレスポンスに Retry-After を付与できる（設定/ヘッダ）
    return resp

Flask-Limiter は標準のレート制限ヘッダ（残り回数・リセット時刻・Retry-After）をレスポンスに付与できます。これを有効にしておくと、クライアント側が自律的にバックオフでき、無駄な再試行で 429 を量産する悪循環を断てます。エラーレスポンスを JSON で統一する設計は エラー処理・可観測性ガイド に揃えてあります。

4. データベース：本当のボトルネックはほぼここにいる

計測（§1）を素直にやると、遅さの正体はたいてい DB に行き着きます。ここを 4 点に絞って潰します。詳細な ORM 設計は Flask-SQLAlchemy 実践ガイド に、接続プーリングの深掘りは PostgreSQL コネクションプーリングガイド にあります。

4.1 N+1 問題：検出と eager loading

N+1 問題は、ORM 最大の性能バグです。一覧（1 クエリ）を取ったあと、各行の関連データをループの中で 1 件ずつ引いてしまい、1 + N 本のクエリが飛ぶ現象です。100 件の一覧で、関連を 1 つ引くだけで 101 本——これが「一覧が遅い」の典型的な正体です。

# ❌ N+1：orders を1回引いて、ループ内で各 order.customer を都度引く
orders = db.session.execute(db.select(Order)).scalars().all()
for order in orders:
    print(order.customer.name)   # ← ここで order ごとに SELECT が飛ぶ

検出は §1.2 の SQL ログが最も確実です。1 リクエストで似たクエリが何十本も並んでいたら N+1 を疑います。解決は eager loading（先読み）——関連を最初の 1〜2 クエリでまとめて取ります。

from sqlalchemy.orm import selectinload

# ✅ selectinload：customer を別の1クエリでまとめて先読み（合計2クエリ）
orders = db.session.execute(
    db.select(Order).options(selectinload(Order.customer))
).scalars().all()
for order in orders:
    print(order.customer.name)   # 追加クエリは飛ばない

selectinload（IN 句でまとめ取り）と joinedload（JOIN で 1 クエリ化）の使い分け、コレクション/スカラ関連での向き不向きは、ORM 側の知識です。詳細は Flask-SQLAlchemy 実践ガイド を参照してください。あわせて、WHERE や JOIN で使う列に索引（インデックス）が張られているかも必ず確認します。索引欠落のフルスキャンは、eager loading でも救えません。

4.2 コネクションプール：ワーカー数との掛け算が事故る

DB 接続は「作るのが高い」資源なので、SQLAlchemy はコネクションプールで再利用します。問題は、Gunicorn の各ワーカーが独立プロセスで、プールもワーカーごとに別々だということです。

ここで掛け算の事故が起きます。

最大接続数 ≒ pool_size × ワーカー数 × タスク（コンテナ）数

例：pool_size=5 × ワーカー 8 個 × タスク 4 個 = 160 接続。PostgreSQL の既定 max_connections（しばしば 100 程度）を簡単に超え、新規接続が FATAL: too many connections で弾かれます。これは本番で頻発する事故です。

# SQLAlchemy のプール設定（Flask-SQLAlchemy 経由）
app.config["SQLALCHEMY_ENGINE_OPTIONS"] = {
    "pool_size": 5,         # 常時保持する接続数
    "max_overflow": 5,      # 一時的に超過を許す数
    "pool_pre_ping": True,  # 接続を使う前に生存確認（切れたコネクションを掴まない）
    "pool_recycle": 1800,   # 30分でリサイクル（DB/プロキシのアイドル切断対策）
}

• pool_pre_ping=True：プールから取り出した接続を使う前に軽い ping を打ち、DB/プロキシ側で切れていた接続を掴んでエラーになるのを防ぎます。ほぼ必須の設定です。
• pool_recycle：一定秒数で接続を作り直し、DB やプロキシのアイドルタイムアウトで黙って切られた接続を避けます。

⚠️ プールサイズはワーカー数とセットで設計する。pool_size だけ見て決めると、ワーカー数・コンテナ数を掛けた瞬間に max_connections を突破します。デプロイ前に必ず pool_size × max_overflow を含む最大 × ワーカー数 × タスク数 を計算し、DB の max_connections 以内に収まるか確認してください。この検算は 本番デプロイガイド のチェックリストにも入れてあります。

4.3 PgBouncer：多ワーカー/サーバーレスの接続爆発を吸収する

ワーカー数 × コンテナ数が増えると、§4.2 の掛け算はどうしても膨らみます。さらに Lambda のようなサーバーレスでは、同時実行数だけ接続が瞬間的に開き、PostgreSQL を一撃で枯渇させます。この「接続爆発」を吸収するのが PgBouncer のような外部コネクションプーラです。

PgBouncer をアプリと PostgreSQL の間に挟むと、

• アプリ（多数のワーカー/Lambda）は PgBouncer に大量の接続を張ってよい。
• PgBouncer が 少数の実接続に多重化して PostgreSQL に橋渡しする。

これにより、max_connections を食い潰さずに高い並行度を捌けます。PgBouncer のプールモード（session / transaction / statement）の選び方、サーバーレスでの注意点（プリペアドステートメントとの相性など）は、PostgreSQL コネクションプーリングガイド に集約しています。多ワーカー × 多コンテナ、またはサーバーレスを使うなら、PgBouncer は事実上の必須コンポーネントです。

4.4 ページネーション：全件ロードをやめる

「一覧 API が、テーブル全件をメモリに載せてから返している」——これは静かに肥大化する典型的な性能爆弾です。データが 100 件のうちは平気でも、10 万件になった瞬間にメモリと DB が悲鳴を上げます。最初からページネーションで必要な分だけ取ります。Flask-SQLAlchemy には db.paginate があります。

@app.route("/api/orders")
def list_orders():
    page = request.args.get("page", 1, type=int)
    # 全件ロードせず、ページ単位で取得（LIMIT/OFFSET を自動付与）
    pagination = db.paginate(
        db.select(Order).order_by(Order.created_at.desc()),
        page=page, per_page=50, error_out=False,
    )
    return jsonify(
        items=[o.to_dict() for o in pagination.items],
        page=pagination.page, pages=pagination.pages, total=pagination.total,
    )

大規模テーブルでは OFFSET が深いページで遅くなるため、カーソル（キーセット）ページネーションへの発展も視野に入れますが、まずは「全件ロードを撲滅し、db.paginate で上限を切る」だけで大半の問題は消えます。詳細は Flask-SQLAlchemy 実践ガイド を参照してください。

5. レスポンス層：転送量とシリアライズを削る

DB とキャッシュを固めたあとの「最後の一押し」が、レスポンスそのものの最適化です。効果は DB ほど大きくありませんが、低コストで効きます。

5.1 圧縮（gzip / brotli）

テキスト（JSON / HTML）レスポンスは圧縮で転送量を大きく削れます。圧縮はリバースプロキシ（nginx）や CDN（CloudFront）/ALB で行うのが第一選択です。アプリの CPU を使わずに済みます。前段で圧縮できない構成（プロキシなしや特殊要件）でのみ、Flask-Compress でアプリ側 gzip/brotli を有効にします。

# 前段プロキシで圧縮できない場合のフォールバック
from flask_compress import Compress

Compress(app)   # Accept-Encoding に応じて gzip/brotli で自動圧縮

💡 圧縮はインフラ層に寄せる。Flask-Compress はアプリの CPU を消費します。ALB/CloudFront/nginx が前段にいるなら、圧縮はそちらに任せ、Flask は本来の処理に CPU を使う方が、同じワーカー数でより多くのリクエストを捌けます。「アプリでできること」と「インフラでやるべきこと」を切り分けるのが、コスト効率の良い設計です。

5.2 JSON シリアライズのコスト

大きなレスポンスでは、JSON シリアライズ自体が無視できない CPU コストになります。対策は素直です。

• 不要なフィールドを返さない：API のスキーマを「実際に使う列」に絞る。シリアライザ（marshmallow など）の only / exclude で削る（marshmallow + Flask-SQLAlchemy ガイド）。
• N+1 を消す（§4.1）：シリアライズ中に関連を引くと、シリアライズと N+1 が複合して遅くなる。eager loading で先に取り切る。
• ページネーション（§4.4）：そもそも 1 レスポンスのサイズを上限で切る。

5.3 ストリーミングと async の「限定的な」効果

巨大なレスポンス（大きな CSV エクスポート等）は、全体をメモリに作ってから返すとメモリを食い、TTFB（最初のバイトまでの時間）も悪化します。Flask はジェネレータを返すことでストリーミングレスポンスを作れ、行を生成しながら流せます。

from flask import Response


@app.route("/api/export.csv")
def export_csv():
    def generate():
        yield "id,name\n"
        for row in iter_rows():          # DB から少しずつ取りながら流す
            yield f"{row.id},{row.name}\n"
    return Response(generate(), mimetype="text/csv")

そして async def ビューですが、§1.1 で触れた通りスループット向上策ではありません。各リクエストは依然 1 ワーカーを占有し、同時処理数は増えません。async が効くのは「1 ビュー内で複数の独立した外部 IO を並行化して、そのビューのレイテンシを縮める」場合に限られます。アプリ全体を非同期前提にしたい/大量同時接続を捌きたいなら、ASGI（Quart など）や gevent ワーカーという別の道を検討します。この線引きは 本番デプロイガイド §7 と ピラー で詳説しています。

6. コスト効率：最適化はそのまま請求額に効く

ここまでの最適化は、レイテンシだけでなくインフラコストに直接効きます。これはサイト全体で一貫して訴えているテーマです——「速い・安い・安全」を一人 × 生成 AI で両立させる。その「安い」を支えるのが、この章の発想です。

最適化 → コスト削減の因果は明快です。

核心は 「1 リクエストの処理時間が縮む ＝ ワーカーの占有時間が縮む ＝ 同じコンテナ数でより多くのリクエストを捌ける ＝ より少ない/小さいコンテナで足りる」 という連鎖です。Flask は同期で 1 ワーカー = 1 リクエスト（§1.1）なので、処理時間の短縮がそのまま必要コンテナ数の削減になり、ECS/Fargate の課金（vCPU・メモリ × 稼働時間）を直接押し下げます。

💡 筆者の実感：木材流通 SaaS を ECS(Fargate) で運用するうえで、コストを左右したのは「いかにキャッシュで DB を叩かず、N+1 を消して 1 リクエストを軽くし、ワーカーを過不足なく並べるか」でした。読み取りが支配的な業務 SaaS では、マスタデータをキャッシュし、一覧を eager loading + ページネーションで軽くするだけで、必要なタスク数（＝月額）が目に見えて変わります。最適化は「速さ」の話であると同時に、まぎれもなく「お金」の話です。スケールは水平（タスク数）で、しかし1 リクエストを軽くしてから——この順序がコスト効率を決めます。

7. Flask パフォーマンス／コスト最適化チェックリスト

最後に、本記事の流れを「計測 → DB → キャッシュ → レート制限 → ワーカー → 圧縮」の順で実務チェックリストにまとめます。上から順に効果が大きいので、上から潰してください。

まとめ：速さは「計測の順序」で決まり、それはそのままコストになる

Flask のパフォーマンス最適化は、派手な技ではなく地味な順序の規律です。本記事の要点を一言ずつにまとめます。

1. 計測してから動く。所要時間ログと SQL ログで p95/p99 とボトルネックを特定する。推測で micro 最適化しない。Flask は同期で 1 ワーカー = 1 リクエスト——速さの本質はワーカー設計と DB 待ち時間にある。
2. **キャッシュ（Flask-Caching / Redis）**で read-heavy・高コストな読み取りをメモリ速度にする。@cache.cached（ビュー）と @cache.memoize（関数を引数ごと）を使い分け、SimpleCache を本番で使わない。本当に難しいのは無効化——TTL を既定に、必要なら明示削除を足し、スタンピードを意識する。
3. **レート制限（Flask-Limiter）**で濫用・コスト・公平性・ブルートフォースを制御する。memory:// はマルチワーカーで素通りする双子の罠——必ず共有 Redis を storage_uri に。プロキシ背後では ProxyFix で本物の IP を取る。
4. データベースが真のボトルネック。N+1 を eager loading で潰し、索引を張り、pool_size × ワーカー × タスク を max_connections 内に収め、多ワーカー/サーバーレスでは PgBouncer を挟み、全件ロードを db.paginate に置き換える。
5. レスポンス層は最後の一押し。圧縮はインフラ層に寄せ、不要フィールドを削り、巨大レスポンスはストリーミング。async はスループット策ではない（1 ワーカー占有の制約）。
6. これらはそのままコスト。1 リクエストが軽くなれば同じワーカーでより多く捌け、より少ない/小さいコンテナで足りる。最適化は「速さ」の話であると同時に「お金」の話である。

筆者が経済産業大臣賞 B2B SaaS のバックエンドを ECS(Fargate) 上で、速く・安く運用できたのは、この「計測 → DB → キャッシュ → レート制限 → ワーカー → 圧縮」という順序を守ったからです。全体像は Flask 本番運用ガイド（ピラー） に、デプロイ側の設計は 本番デプロイガイド に、ORB と接続の深掘りは Flask-SQLAlchemy 実践ガイド と PostgreSQL コネクションプーリングガイド にあります。速さは才能ではなく、計測の順序を守る規律です。