2024.11.28 Python

FastAPIのベストプラクティス｜SQLAlchemyでのディレクトリ構成・モデル設計・テスト

FastAPIでデータベースを扱うアプリを書き始めると、「モデル・スキーマ・DB接続をどこに置くか」「セッションはどう共有するか」「テストでDBをどう差し替えるか」で必ず手が止まります。この記事は、FastAPIとSQLAlchemyを組み合わせた実務アプリの設計を、推奨ディレクトリ構成から本番運用まで一つの型として示します。前提はSQLAlchemy 2.0とPydantic v2で、2024年以前の記事に多い旧スタイル（declarative_base()・orm_mode）ではなく、現行の書き方に統一しています。

まとめ：FastAPI＋SQLAlchemyベストプラクティスの要点

  • 層で分ける：models（SQLAlchemy）／schemas（Pydantic）／crud・repository（DB操作）／api（ルーティング）を明確に分離する。
  • SQLAlchemyは2.0スタイル：DeclarativeBase＋Mapped[]＋mapped_column()で型を持たせる。
  • セッションは依存性注入：get_dbをyieldで定義し、Annotated[Session, Depends(get_db)]で受け取る。リクエスト単位で開いて確実に閉じる。
  • Pydanticはモデルと分離：レスポンス用スキーマにmodel_config = ConfigDict(from_attributes=True)を付け、ORMオブジェクトをそのまま返さない。
  • マイグレーションはAlembic：スキーマ変更はコードではなくマイグレーションで管理する。
  • テストは依存性オーバーライド：app.dependency_overridesでDBをテスト用に差し替える。

以下、各ポイントを実装例とともに掘り下げます。FastAPI自体の入門やasync／defの使い分けはFastAPIとは｜最短で動かす手順とasync/defの使い分けで扱っているため、本記事はデータベースを含む構成設計に絞ります。

FastAPIとSQLAlchemyを層で分ける設計方針

FastAPIは1ファイルでも動きますが、DBが絡むと責務が混ざり、変更やテストが難しくなります。ベストプラクティスの核は「役割ごとにレイヤーを分ける」ことです。具体的には次の4層を分離します。

  • モデル層（models）：SQLAlchemyのテーブル定義。DBの構造そのもの。
  • スキーマ層（schemas）：Pydanticで定義する入出力の型。APIの契約。
  • データアクセス層（crud／repository）：DBへの読み書きロジック。
  • API層（api／routers）：エンドポイントとルーティング。

この分離で、DB実装を変えてもAPIの型は変わらず、テストではデータアクセス層だけを差し替えられます。バージョンはSQLAlchemyが2.0（2023年リリース、型注釈ベースの新スタイルが標準）、Pydanticがv2（orm_modeはfrom_attributesへ改称）を前提にします。導入時はpip install fastapi sqlalchemy "pydantic[email]" alembicで揃い（EmailStrを使うためemail-validatorを含む[email]指定にしています）、DBドライバは接続先に応じて別途入れます。

ディレクトリ構成のベストプラクティス

「fastapi ディレクトリ構成」で最初に迷うのがフォルダの切り方です。中〜大規模では、責務ごとにディレクトリを分ける次の構成が扱いやすく、コミュニティでも定番です。

app
├── main.py          # FastAPIインスタンスとlifespan、routerの登録
├── core
│   └── config.py    # 設定値（環境変数の読み込み）
├── db
│   ├── base.py      # DeclarativeBase（全modelの基底）
│   └── session.py   # engineとsessionmaker、get_db
├── models           # SQLAlchemyモデル（テーブル定義）
├── schemas          # Pydanticスキーマ（入出力の型）
├── repository       # データアクセス（CRUD）
└── api
    └── routers      # エンドポイント

小規模なら無理に深くしないのも重要です。ファイルが10個程度までならmodels.py・schemas.py・crud.pyのフラット構成で十分で、機能が増えてからmodels/user.pyのように機能単位へ割ると、過剰なディレクトリで迷子になりません。判断基準は「1ファイルが200〜300行を超え、複数の関心事が同居し始めたら分割」です。ディレクトリ数を増やすこと自体は設計の良さを意味しません。

SQLAlchemyモデル定義のベストプラクティス

DeclarativeBaseとMapped[]による2.0スタイル

SQLAlchemy 2.0では、基底クラスをDeclarativeBaseの継承で定義し、カラムはMapped[型]＋mapped_column()で書きます。旧来のColumn(Integer)だけの書き方と違い、型注釈がそのままモデルの型情報になり、エディタ補完と型チェックが効きます。

from datetime import datetime
from sqlalchemy import String, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    posts: Mapped[list["Post"]] = relationship(back_populates="author")

リレーションとNULL可否の型表現

2.0では、Mapped[str]はNOT NULL、Mapped[str | None]はNULL許可というように、Python型の有無がそのままNULL制約に対応します。曖昧さが減るので、旧スタイルのnullable=Trueを手で書き回す必要が減ります。リレーションはrelationship(back_populates=...)で双方向を明示し、外部キーはmapped_column(ForeignKey("users.id"))で張ります。SQLAlchemyとPydanticを1クラスで兼ねたい場合はSQLModel（SQLAlchemy＋Pydantic統合ORM）という選択肢もありますが、細かい制御が要る本番用途では両者を分けたほうが破綻しにくくなります。

データベース接続とセッション管理

エンジンとセッションの生成

接続はエンジンを1つ作り、sessionmakerでセッションの生成器を用意します。接続先はURLを変えるだけで、PostgreSQL・MySQL・SQL Serverいずれにも対応します（例：PostgreSQLはpostgresql+psycopg://user:pass@localhost/dbname）。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("postgresql+psycopg://user:pass@localhost/app", pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False)

pool_pre_ping=Trueは、切断済みコネクションを使う前に生存確認する設定で、本番で「MySQL server has gone away」等の断線エラーを防ぎます。autoflush=Falseは既定のTrueをあえて切っており、クエリ発行のたびに未確定の変更が自動でDBへ流れるのを止めて、commitのタイミングを呼び出し側で明示的に制御する狙いです。

get_dbによる依存性注入

セッションはグローバルに共有せず、リクエストごとに開いて確実に閉じるのが鉄則です。yieldを使った依存関数で定義し、エンドポイントではAnnotatedで受け取ります。FastAPIは同一リクエスト内で同じ依存の結果をキャッシュするため、複数のリポジトリが同じセッションを共有できます。

from typing import Annotated
from fastapi import Depends
from sqlalchemy.orm import Session


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


DbSession = Annotated[Session, Depends(get_db)]


@app.get("/users/{user_id}")
def read_user(user_id: int, db: DbSession):
    ...

高負荷でDBがボトルネックになる場合は、create_async_engineとAsyncSessionによる非同期構成に切り替えます。ただし後述のとおり、同期エンジンのままasync defにするのは逆効果です。

リポジトリパターンとCRUD層の分離

共通リポジトリクラスの実装

DB操作をエンドポイントに直書きすると、同じクエリが各所に散らばります。共通のリポジトリクラスにCRUDを集約すると、テーブルごとに同じ操作を再利用でき、モックへの差し替えも容易になります。

class UserRepository:
    def __init__(self, db: Session):
        self.db = db

    def get(self, user_id: int) -> User | None:
        return self.db.get(User, user_id)

    def create(self, email: str) -> User:
        user = User(email=email)
        self.db.add(user)
        self.db.commit()
        self.db.refresh(user)
        return user

サービス層との責務分担

ビジネスロジック（バリデーション、複数テーブルにまたがる処理）はリポジトリではなくサービス層に置きます。リポジトリは「1テーブルの読み書き」に徹し、commitのタイミングだけは呼び出し側（サービス）が制御する設計にすると、複数操作を1トランザクションにまとめられます。小規模ならサービス層を省き、リポジトリとエンドポイントの2層で始めても構いません。

Pydanticスキーマとレスポンスの分離

SQLAlchemyモデルをそのままレスポンスに返すと、パスワードハッシュなど外に出したくないカラムまで露出します。入出力はPydanticスキーマで定義し、用途ごとに分けます。Pydantic v2ではORMオブジェクトを受け取るためにmodel_config = ConfigDict(from_attributes=True)を付けます（v1のorm_mode = Trueからの改称）。

from pydantic import BaseModel, ConfigDict, EmailStr


class UserCreate(BaseModel):
    email: EmailStr


class UserRead(BaseModel):
    id: int
    email: EmailStr
    model_config = ConfigDict(from_attributes=True)


@app.post("/users", response_model=UserRead)
def create_user(payload: UserCreate, db: DbSession):
    return UserRepository(db).create(payload.email)

response_model=UserReadを指定すると、FastAPIが返却前にフィルタし、UserReadに定義した列だけをJSON化します。この型はそのままSwagger UI（/docs）の自動生成ドキュメントにも反映され、APIの契約書として機能します。作成用（Create）・更新用（Update）・返却用（Read）を分けるのが定番です。

Alembicによるマイグレーション運用

テーブル定義を変えたら、Base.metadata.create_all()で作り直すのではなく、Alembicでマイグレーションを管理します。差分を自動生成し、適用・巻き戻しを履歴として残せます。

alembic init migrations
alembic revision --autogenerate -m "add users table"
alembic upgrade head

自動生成を効かせるには、migrations/env.pyのtarget_metadataに自作のBase.metadataを渡し、全モデルをimportしておく必要があります。importが漏れたモデルは差分検出されないため、モデルを1か所（例：db/base.py）でまとめて読み込むのが定石です。自動生成は万能ではなく、カラムの型変更やインデックス名の変更は取りこぼすことがあるので、生成されたスクリプトは適用前に必ず目視で確認します。

テストのベストプラクティス

DBを使うテストで本番DBに触れないよう、app.dependency_overridesでget_dbをテスト用に差し替えます。SQLiteのインメモリDB（sqlite:///:memory:）を使えば、テストごとにまっさらなDBを高速に用意できます。

from fastapi.testclient import TestClient


def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()


Base.metadata.create_all(bind=test_engine)  # テスト用DBにテーブルを作成
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)


def test_create_user():
    res = client.post("/users", json={"email": "[email protected]"})
    assert res.status_code == 200
    assert res.json()["email"] == "[email protected]"

ポイントは、依存性オーバーライドでDBだけを差し替え、アプリ本体はそのまま検証することです。これにより、ルーティングやスキーマ検証を含めたエンドツーエンドの挙動を、本番DBを汚さずに確認できます。SQLite特有のSQL差異が問題になる場合は、テスト専用のPostgreSQLコンテナを立てて同一エンジンで検証します。

避けるべきアンチパターン

ベストプラクティスの裏返しとして、本番で問題が起きる前に潰しておきたい典型的な失敗を挙げます。いずれも「動くけれど本番で壊れる」類のものです。

  • 同期SQLAlchemyをasync defで使う：同期エンジンのままエンドポイントをasync defにすると、DBアクセス中にイベントループを塞ぎ、同時実行性がむしろ落ちます。同期エンジンならdefで書く（FastAPIが別スレッドで処理する）か、非同期化するならAsyncSessionまで一貫させます。defとasync defの使い分けはFastAPIとはの記事で詳しく整理しています。
  • N+1問題：リレーションをループ内で都度参照すると、件数分のクエリが飛びます。selectinload()やjoinedload()で事前読み込みし、1〜2クエリにまとめます。
  • セッションのグローバル共有：1つのSessionを使い回すと、並行リクエストで状態が混ざり、トランザクションが破綻します。必ずリクエスト単位で生成・破棄します。
  • モデルを直接返す：response_modelを指定せずSQLAlchemyモデルを返すと、非公開カラムの露出や遅延ロード起因のエラーを招きます。返却は必ずPydanticスキーマ経由にします。

本番でのプロセス起動やワーカー構成はUvicornとGunicornの違い・本番デプロイの記事にまとめています。

よくある質問

FastAPIのディレクトリ構成に決まりはありますか？

公式に強制される構成はありません。ただしmodels・schemas・crud（repository）・apiを分ける層構造が実質的な定番です。小規模はフラット、機能が増えたら機能単位へ分割するのが扱いやすい方針です。

SQLAlchemyは1.4系と2.0系どちらで書くべきですか？

新規開発は2.0スタイル（DeclarativeBase・Mapped[]・mapped_column()）を推奨します。型注釈が効き、非同期にも対応しやすく、公式ドキュメントも2.0系が基準になっています。

Pydanticのorm_modeが使えないのはなぜですか？

Pydantic v2でfrom_attributesへ改称されたためです。model_config = ConfigDict(from_attributes=True)と書くことで、SQLAlchemyオブジェクトから属性アクセスで値を読み込めます。

リポジトリパターンは小規模でも必要ですか？

必須ではありません。エンドポイントが数個なら直書きでも回ります。同じクエリが複数箇所に現れ始めた、あるいはテストでDBをモックしたくなった時点で導入すると効果が出ます。

DBはPostgreSQL・MySQL・SQL Serverのどれでも同じコードで動きますか？

SQLAlchemyのORM層はDBを抽象化するため、接続URLとドライバを変えれば大部分は共通です。ただしJSON型や全文検索などDB固有機能を使う箇所は差異が出るため、本番と同じDBでテストするのが安全です。

関連記事

資料請求

RELATED POSTS 関連記事

一覧へ戻る