FastAPIのサービスをAIに書かせ、よくある間違いをレビューで見つける

## Python・FastAPI・ライブラリの最低バージョンを決める すべての依存パッケージを以下のバージョン以上に固定すること。ここで示す指針はすべて、このバージョンを前提にしている。古いバージョンのままだと、後述する非同期API、`Annotated` による依存性注入、v2シリアライズといったイディオムが気づかないうちに壊れる。 | 依存パッケージ | 最低バージョン | この下限が必要な理由 | |---|---|---| | Python | 3.11 | `StrEnum`、および `X \| Y` のユニオン演算子 | | FastAPI | 0.115 | `Annotated[T, Depends(...)]` が標準的な書き方になる | | Pydantic | 2.7 | v1のAPI（`json_encoders`、`.dict()`）が削除されている | | pydantic-settings | 2.4 | Pydantic v2以降は別パッケージになった | | SQLAlchemy | 2.0 | 非同期API（`AsyncSession`、`async_sessionmaker`） | | Alembic | 1.13 | 非同期対応のマイグレーション | | httpx | 0.27 | インプロセステスト用の `ASGITransport` | | PyJWT | 2.9 | メンテナンスが続いている。`python-jose` の代替 | | ruff | 0.6 | lint・フォーマット・import整理をこれ1つでこなす | ## ファイルの種類ではなく、ドメインごとにフォルダを分ける ディレクトリ構成はファイル種別ではなく、ドメイン（境界づけられたコンテキスト）単位で分ける。各ドメインに専用のパッケージを持たせ、共通の関心事は薄いグローバル層にとどめる。 - 各ドメインパッケージ：`router.py`（エンドポイント）、`schemas.py`（Pydanticモデル）、`models.py`（ORMモデル）、`service.py`（ビジネスロジック）、`dependencies.py`、`config.py`（ドメイン固有の設定）、`constants.py`、`exceptions.py`、`utils.py`。 - グローバル最上位層：`config.py`（全体設定）、`models.py`（共有基底クラス）、`exceptions.py`、`database.py`（非同期エンジンとセッションファクトリ）、`main.py`（アプリ本体とlifespan）。 - ドメインをまたぐimport：明示的にモジュールをimportしてエイリアスを付ける（例：あるドメインの `service` を `auth_service` としてimportする）。ワイルドカードimportや、内部の深い階層への直接参照は絶対に使わない。どちらもドメイン同士を密結合させ、リファクタリングを妨げる。 ## アプリが固まらないよう、async def と def のルートを使い分ける ルートの書き方は、そのルートが行う処理内容に合わせて選ぶ。選択を誤ると、イベントループが固まるか、スレッドを無駄に浪費することになる。 | ルートの処理内容 | 使うべき書き方 | |---|---| | awaitableかつノンブロッキングなI/O | `async def` | | ブロッキングI/Oで、非同期クライアントが存在しない | `def`（同期、スレッドプールで実行） | | 両方が混在する | `async def` ＋ ブロッキング処理部分は `run_in_threadpool` | | 約50ms以上かかるCPUバウンドな処理 | ワーカープロセスへオフロード（Celery / Arq / RQ） | `time.sleep`、`requests`、同期DBドライバ、素の `open()` といったブロッキング呼び出しを `async def` の中に絶対に入れないこと。そのワーカー上のすべてのリクエストが固まってしまう。同期の `def` ルートならスレッドプールのワーカーを1つ占有するだけで済むので、避けられないブロッキング処理の置き場として安全。`async def` の中では、対応するawaitableな処理（`asyncio.sleep`、`aiofiles`、非同期ドライバ）を使うか、同期ライブラリを `run_in_threadpool` でラップする。注意点：デフォルトのスレッドプールは約40スレッドしかないため、そこが埋まるとすべての同期ルートが遅くなる。またスレッドはコルーチンよりコストが高いので、ブロッキングする理由がないのに同期ルートを選んではいけない。 ## Pydanticで入力を確かめ、設定をドメインごとに分ける - 自前のチェックより組み込みのバリデータ・型を優先する：`Field(min_length=, max_length=, pattern=, ge=)`、`EmailStr`、`AnyUrl`、選択肢が閉じている場合は `StrEnum`。 - 制約と矛盾するデフォルト値を組み合わせない。`Field(ge=18, default=None)` はバグである。必須にする（`Field(ge=18)`）か、任意にする（`int | None = Field(default=None, ge=18)`）かをはっきり決める。 - 最新のシリアライズ方式を使う。`json_encoders` はPydantic v2で非推奨になっているので、フィールドごとのルールには `@field_serializer` を、型への注釈には `PlainSerializer` を使う。 - 設定はドメインごとに分割する。`pydantic-settings` は独立したパッケージであり、全ドメインが1つのグローバル設定オブジェクトを読む形にせず、各ドメインに専用の `env_prefix` を持つ `BaseSettings` サブクラスを1つずつ用意する。 ## 依存関数をAnnotatedで書き、受け取るデータのチェックもさせる - 依存性注入には `Annotated[T, Depends(fn)]` を使う。FastAPI 0.95以降の標準的な書き方である。デフォルト引数形式（`param: T = Depends(fn)`）は避ける。これはレガシーな書き方で、デフォルト値絡みの落とし穴がある。 - 依存関数には注入だけでなくバリデーションもさせる。idでリソースを読み込み、存在しなければドメイン固有の404を返し、検証済みのオブジェクトを返すようにすれば、ルート側はすでにチェック済みのものを受け取れる。 - 依存関数同士を連鎖させてロジックを再利用する。1つの依存関数が他の依存関数に依存してもよい。例えば、所有者チェックの依存関数が、id読み込みの依存関数とトークン解析の依存関数の両方に依存する、といった形。 - ルール：依存関数はリクエスト単位でキャッシュされるため、同じ `Depends(x)` を1つのリクエスト内で何度使っても実行は1回だけになる。依存関数は `async def` を優先する。同期関数はスレッドプールで実行されるため、小さなCPUだけのチェックには無駄なオーバーヘッドになる。エンドポイント間で依存関数を使い回すには、共有するパス変数に同じ名前を付ける。 ## JWTの認証と、非同期のデータベース接続を書く - JWT：メンテナンスが止まっている `python-jose` ではなく `PyJWT`（`import jwt`）を使う。デコード処理は `InvalidTokenError` を捕まえるガードの内側で行い、生の例外を外部にそのまま見せることなく、ドメイン固有の認証エラーを送出する。 - セッション：SQLAlchemy 2.0の非同期API（`create_async_engine`、`async_sessionmaker`、`AsyncSession`）を優先し、`get_db` という非同期ジェネレータの依存関数を通じてルートにセッションを渡す。`encode/databases` はメンテナンスモードに入っているため、新規プロジェクトでは選ばない。 - 命名：`lower_case_snake` を使う。テーブル名は単数形（`post`、`user`、`post_like`）。関連するテーブルは共通のプレフィックスでまとめる（`payment_account`、`payment_bill`）。日時には `_at`、日付には `_date` のサフィックスを付ける。外部キーのカラム名は、どこに登場しても同じ名前を使い回す。 - `MetaData` にプロジェクト全体で1つのインデックス・制約命名規則を設定し、生成される名前が予測可能なままになるようにする。 - SQLファースト、Pydanticはその次：結合・集計・JSON整形はSQL側で行い、Pydanticへの変換はレスポンスのバリデーション用にとどめ、データ変形には使わない。 ## バックグラウンドタスクと、Celeryなどのタスクキューを使い分ける タスクのコストと信頼性の要件に合わせてツールを選ぶ。 | インプロセスのバックグラウンドタスクを使う場面 | 本格的なキュー（Celery / Arq / RQ）を使う場面 | |---|---| | タスクが1秒未満で終わる | 数秒〜数分かかる | | 気づかず失われても問題ない | リトライ、デッドレター、可視性が必要 | | インプロセスで完結する（メール送信、1行のログ記録など） | CPU負荷が高い、または専用プールが必要 | | スケジューリングが不要 | cron、ETA、レート制限が必要 | インプロセスのバックグラウンドタスクは、レスポンス送信後に同じワーカープロセス内で実行される。そのワーカーが落ちればタスクは失われ、リトライも発生しない。そのため、オンコールのエンジニアを呼び出すような重要な処理には絶対に使ってはいけない。 ## httpxと本物のデータベースでテストする - 最初から非同期HTTPクライアントを使う：`ASGITransport(app=app)` を介した `httpx.AsyncClient` でアプリをインプロセスで動かす。`async_asgi_testclient` は使わない。メンテナンスが止まっている。 - 内部をモンキーパッチするのではなく、フレームワーク組み込みの `app.dependency_overrides[dep] = fake` で依存関数を差し替える。各テストの後にオーバーライドをクリアすること。 - 統合テストでは実際のデータベース（一時的なスキーマや使い捨てのコンテナ）を使い、認証と外部サービスだけをオーバーライドする。データベースをモックすると、モックと本番の乖離が隠れてしまい、それが後で本番で表面化する。 ## マイグレーションを戻せる形にし、APIごとに説明を書く - マイグレーション：すべてのマイグレーションを静的かつリバーシブルに保ち、非同期テンプレートから生成する。ファイル名は日付プレフィックスとスラグを組み合わせた説明的なタイムスタンプ付きにして、履歴を順番通りに読めるようにする。 - APIドキュメント：現在の環境が許可リストに含まれない場合はOpenAPIのURLを `None` に設定し、対象外の環境ではインタラクティブドキュメントを非表示にする。各エンドポイントはレスポンスモデル、ステータスコード、summary、description、tags、そしてエラーステータスごとのレスポンス項目まで、漏れなくドキュメント化する。 - Lint：フォーマッタとlintツールを実行し（例えば `ruff check --fix` の後に `ruff format`）、pre-commitフックやCIに組み込む。1つの最新ツールが、これまで別々だったフォーマッタ・import整理・デッドコード検出ツールを置き換える。 ## FastAPIの差分で、AIのよくある間違いを見つけて直す 差分をレビューする際は、以下のようなエージェントのミスを指摘すること。各行は実際に起こる失敗パターンとその修正方法を示す。 | アンチパターン | なぜ問題なのか | 修正方法 | |---|---|---| | `async def` の中で `requests.get(...)` | イベントループをブロックする | `httpx.AsyncClient`、または `await run_in_threadpool(...)` | | `async def` 内での `time.sleep` / `open()` / 同期DBドライバ | ループをブロックする | 対応する非同期版（`asyncio.sleep`、`aiofiles`、非同期ドライバ） | | `from jose import jwt` | メンテナンスされていないライブラリ | `import jwt`（PyJWT） | | `async_asgi_testclient` | メンテナンスされていない | `httpx.AsyncClient` ＋ `ASGITransport` | | `ConfigDict(json_encoders=...)` | Pydantic v2では非推奨 | `@field_serializer` または `PlainSerializer` | | `Field(ge=18, default=None)` | 制約とデフォルト値が矛盾している | 必須か任意かどちらかに決める | | デフォルト引数形式の `Depends(...)` | レガシーなデフォルト値の落とし穴 | `Annotated[T, Depends(...)]` | | ルート本体を素の `Exception` でキャッチする | バグを隠し、500エラーを気づかれない200に変えてしまう | 具体的な例外クラスをキャッチし、明示的なHTTPエラーを送出する | | 呼び出し対応が必要な処理にバックグラウンドタスクを使う | リトライがなく、ワーカーと一緒に消える | Celery / Arq / RQ | | `async def` 内での同期ORMセッション | ループをブロックし、プールがデッドロックすることもある | `AsyncSession` | | モデルを返しつつ同じモデルを `response_model` にも設定する | モデルが2回構築される | dictや行を返すか、`response_model` を外す | | ドメインをまたぐ深い階層へのimport | 密結合を生む | ドメインのモジュールをimportしてエイリアスを付ける | | アプリ全体で `BaseSettings` を1つだけ使う | 全ドメインがすべての変数を読んでしまう | ドメインごとに `BaseSettings` を1つ用意する | | 統合テストでデータベースをモックする | モックと本番の乖離が生じる | 実際のDBを使い、認証・外部サービスだけを依存性オーバーライドする |