---
title: "FastAPIのベストプラクティス｜ディレクトリ構成・SQLAlchemy 2.1の設計・テストを実測で解説"
url: "https://www.issoh.co.jp/tech/details/4347/"
published: 2024-11-28
updated: 2026-10-30
categories: ["Python"]
publisher: "株式会社一創"
---

# FastAPIのベストプラクティス｜ディレクトリ構成・SQLAlchemy 2.1の設計・テストを実測で解説

FastAPIでデータベースを扱うアプリを書き始めると、「ファイルをどう分けるか」「セッションをどこで開いてどこでcommitするか」「テストでDBをどう差し替えるか」で手が止まります。この記事は、FastAPIとSQLAlchemyを組み合わせた実務向けの構成を、ディレクトリ構成からマイグレーション・テストまで1つの型として示します。掲載したコードは2026年10月15日に、FastAPI 0.143.1・SQLAlchemy 2.1.4・Pydantic 2.14.1・Python 3.14.6・PostgreSQL 18.6の組み合わせで実行し、pytestとAlembicまで通したものです。

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

- **ディレクトリは層で分ける**：models・schemas・repositories・services・apiに分け、ドメインが増えて1つのディレクトリに別機能のファイルが混ざり始めたらドメイン単位へ移す。
- **commitはサービス層で呼ぶ**：yieldの後にcommitを書くと、commitが失敗してもクライアントには200が返る（実測）。
- **同期のSQLAlchemyならエンドポイントは`def`**：`async def`の中で同期セッションを呼ぶと、0.5秒のクエリ10本の同時実行に5.13秒かかった（`def`は0.58秒）。
- **SQLAlchemy 2.1の変更を前提にする**：Python 3.11以上が必要で、`postgresql://`の既定ドライバがpsycopg 3になり、非同期用のgreenletは`sqlalchemy[asyncio]`で入れる。
- **制約名の規則を最初に入れる**：`naming_convention`を設定したBaseを使い、Alembicの自動生成で制約名を固定する。
- **テストは`dependency_overrides`**：`get_db`だけを差し替え、SQLiteのインメモリDBは`StaticPool`で1接続に固定する。

以下、各項目をコードと実行結果つきで説明します。FastAPIの入門や起動方法は[FastAPIとは｜最短で動かす手順とasync/defの使い分け](/tech/details/3859/)で扱っているため、この記事はデータベースを含む構成設計に絞ります。

## 2025〜2026年の変更で古くなった書き方

2024年以前の記事には、今の版では警告が出る書き方や、挙動が変わった機能を使うコードが含まれています。2025年9月以降の変更のうち、FastAPIとSQLAlchemyの構成に影響するものを並べました。

| 変更                             | 版（日付）                        | 構成への影響                       |
| ------------------------------ | ---------------------------- | ---------------------------- |
| yield後の処理をレスポンス送信後へ戻す          | FastAPI 0.118.0（2025-09-29）  | yield後のcommit失敗が応答に出ない       |
| `Depends(scope="function")`を追加 | FastAPI 0.121.0（2025-11-03）  | 後始末を応答前に終えられる                |
| `pydantic.v1`のサポートを削除          | FastAPI 0.128.0（2025-12-27）  | `orm_mode`などv1の書き方は不可        |
| pyproject.tomlのentrypointを文書化  | FastAPI 0.135.2（2026-03-01）  | CLIにパスを渡さずに済む                |
| テストクライアントがhttpx2に対応            | Starlette 1.2.0（2026-05-28）  | 1.8.0はhttpxのままだと警告           |
| Python 3.11以上が必須               | SQLAlchemy 2.1.0（2026-09-24） | 3.10の環境は2.0系に固定              |
| greenletを既定で入れない               | SQLAlchemy 2.1.0             | 非同期は`sqlalchemy[asyncio]`が必要 |
| `postgresql://`の既定がpsycopg 3   | SQLAlchemy 2.1.0             | psycopg2だけの環境は接続不可           |

FastAPI側の変更は[FastAPIのリリースノート](https://fastapi.tiangolo.com/release-notes/)、SQLAlchemy側は[SQLAlchemy 2.1の移行ガイド](https://docs.sqlalchemy.org/en/21/changelog/migration%5F21.html)、httpx2は[Starletteのリリースノート](https://starlette.dev/release-notes/)で確認できます。最後の行は手元で再現しており、psycopg2-binaryだけを入れたvenvで`create_engine("postgresql://...")`を呼ぶと`ModuleNotFoundError: No module named 'psycopg'`で止まりました。新規の開発はSQLAlchemy 2.1とPython 3.11以上で始めます。Python 3.10を外せない既存環境は2.0系（2026年10月15日時点の最新は9月15日公開の2.0.54）を使い、2.1へ上げる前にドライバとgreenletの有無を確認します。

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

FastAPIのディレクトリ構成に公式の決まりはありません。公式チュートリアルの[Bigger Applications（複数ファイルのアプリ）](https://fastapi.tiangolo.com/tutorial/bigger-applications/)は`APIRouter`でファイルを分ける方法までで、DB層の置き場所は扱っていません。実務で使われる型は、ファイルの種類で分ける構成と、ドメイン（機能）単位で分ける構成の2つです。

### ファイル種別で分ける構成（小〜中規模の標準）

この記事のコードはすべて次の構成で動かしています。

```
.
├── pyproject.toml
├── alembic.ini
├── alembic/
│   ├── env.py
│   └── versions/
├── app/
│   ├── main.py            # FastAPIの生成・ルーター登録・lifespan
│   ├── core/
│   │   └── config.py      # pydantic-settingsの設定クラス
│   ├── db/
│   │   ├── base.py        # DeclarativeBaseと制約名の規則
│   │   └── session.py     # エンジン・セッション・get_db
│   ├── models/            # SQLAlchemyのテーブル定義
│   │   ├── __init__.py    # 全モデルをここでimport
│   │   ├── user.py
│   │   └── post.py
│   ├── schemas/
│   │   └── user.py        # Pydanticの入出力スキーマ
│   ├── repositories/
│   │   └── user.py        # DBの読み書き
│   ├── services/
│   │   └── user.py        # 業務ルールとcommit
│   └── api/
│       ├── deps.py        # SessionDepなどの共通依存
│       └── routers/
│           └── users.py
└── tests/
    ├── conftest.py
    └── test_users.py
```

依存の向きは api → services → repositories → models の一方向にそろえ、modelsからapiをimportしないようにします。向きを決めておくと、ファイルが増えたときの循環importを防げます。`models/__init__.py`で全モデルをimportしておくのは、Alembicの自動生成とテストのテーブル作成が、モデルの読み込み漏れで差分を見落とさないようにするためです。

### ドメイン単位で分ける構成（ドメインが多いモノリス）

```
src/
├── auth/
│   ├── router.py
│   ├── schemas.py
│   ├── models.py
│   ├── dependencies.py
│   ├── service.py
│   └── exceptions.py
├── posts/
│   ├── router.py
│   ├── schemas.py
│   ├── models.py
│   ├── service.py
│   └── ...
├── config.py      # 全体の設定
├── database.py    # DB接続
└── main.py
```

GitHubの[fastapi-best-practices](https://github.com/zhanymkanov/fastapi-best-practices)（zhanymkanov）は、ファイル種別の構成について「マイクロサービスや小さなプロジェクトではうまくいくが、多くのドメインを持つモノリスでは伸びなかった」と書き、Netflixの[Dispatch](https://github.com/Netflix/dispatch)に倣ったドメイン単位の構成を勧めています。別ドメインの部品を使うときは`from src.auth import constants as auth_constants`のように、モジュール名を明示してimportします。

| 判断軸             | ファイル種別                  | ドメイン単位         |
| --------------- | ----------------------- | -------------- |
| ドメインの数          | 数個まで                    | 多い             |
| 1機能の追加で触るディレクトリ | 複数（models・schemas・api等） | 原則1つ           |
| 共通処理の置き場        | 各層の中                    | src直下のモジュール    |
| 向く体制            | 1チーム・マイクロサービス           | ドメインごとに担当が分かれる |

最初はファイル種別で始めます。ドメイン単位は、1つのディレクトリに別機能のファイルが混ざって探しにくくなった時点、あるいは担当チームがドメインで分かれた時点で移す構成で、エンドポイントが十数本の段階で採用すると、ディレクトリの数だけが増えます。

### pyproject.tomlのentrypointによるアプリ位置の固定

```
[project]
name = "app"
version = "0.1.0"
requires-python = ">=3.11"

[tool.fastapi]
entrypoint = "app.main:app"
```

[FastAPI CLIのドキュメント](https://fastapi.tiangolo.com/fastapi-cli/)は、`[tool.fastapi]`の`entrypoint`にアプリの場所を書く方法を案内しています。`fastapi dev app/main.py`のようにパスを毎回渡す必要がなくなり、VS Code拡張やFastAPI Cloudもアプリを見つけられます。上の構成で引数なしの`fastapi run`を実行すると、起動ログは次のとおりでした（行頭の絵文字は省いています）。

```
$ fastapi run
Starting FastAPI in production mode
Using import string: app.main:app
Server started at http://0.0.0.0:8000
```

`fastapi`コマンドはfastapi-cliが提供するため、`pip install "fastapi[standard]"`で入れます。この記事のアプリとテストを再現するには、`pip install "fastapi[standard]" sqlalchemy pydantic-settings alembic "psycopg[binary]" pytest httpx2`で必要なパッケージがそろいます。

## 設定値の一元管理とpydantic-settings

```
from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    database_url: str = "sqlite:///./app.db"
    sql_echo: bool = False

@lru_cache
def get_settings() -> Settings:
    return Settings()
```

`BaseSettings`を継承したクラスは、引数で渡されなかった項目を環境変数（と`.env`）から読みます。FastAPIの[設定のドキュメント](https://fastapi.tiangolo.com/advanced/settings/)のとおり`get_settings()`に`@lru_cache`を付けると、Settingsは初回呼び出しで1回だけ作られ、リクエストのたびに`.env`を読み直しません。`DATABASE_URL`を環境変数で渡せば、開発はSQLite、本番はPostgreSQLとコードを変えずに切り替えられ、後述のAlembicも同じ`get_settings()`から接続先を読みます。pydantic-settings 2.16.0もPython 3.11以上が必要です。

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

### DeclarativeBaseとMapped\[\]による2.0以降の書き方

```
from datetime import datetime

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

from app.db.base import Base

class User(Base):
    __tablename__ = "users"

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

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

```
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db.base import Base

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    author: Mapped["User"] = relationship(back_populates="posts")
```

2.0以降は`Mapped[]`の型注釈から列の型とNULL可否が決まり、`Mapped[str | None]`ならNULL許可、`Mapped[str]`ならNOT NULLです。この書き方は2.1でも変わりません。`relationship`は`back_populates`で双方向を明示します。SQLAlchemyとPydanticを1クラスで兼ねる[SQLModel](/tech/details/10322/)もあり、FastAPI公式の[SQLデータベースのチュートリアル](https://fastapi.tiangolo.com/tutorial/sql-databases/)はSQLModelで書かれています。SQLModelでもテーブル用と入出力用のモデルは分けられます。この記事はSQLAlchemyのモデルとPydanticのスキーマを別ファイルに分け、テーブルの列とレスポンスの項目を別々に変更できる形にしています。

### 制約名の命名規則（naming\_convention）の初期設定

```
from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase

# 制約名の規則を決めておくと、Alembicの差分で名前が安定する
NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)
```

Alembicのマイグレーションは制約を名前で参照します（[Alembicの命名規則のドキュメント](https://alembic.sqlalchemy.org/en/latest/naming.html)）。名前を付けないとDB側の自動命名になり、後で制約を消す・変えるときに実際の名前を調べる手間が生じます。このBaseを使って`alembic revision --autogenerate`を実行すると、生成されたスクリプトの制約名は規則どおりになりました。

```
sa.PrimaryKeyConstraint('id', name=op.f('pk_users'))
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
sa.ForeignKeyConstraint(['author_id'], ['users.id'], name=op.f('fk_posts_author_id_users')),
sa.PrimaryKeyConstraint('id', name=op.f('pk_posts'))
```

規則が使われるのは名前を付けていない制約とインデックスで、名前を付けた制約はその名前のまま作られます（`my_uq`と名付けたUNIQUE制約は`my_uq`のままでした）。例外がこの例の`ck`で、`%(constraint_name)s`を使うため、CHECK制約には`name=`が必須です。`CheckConstraint("x > 0", name="positive")`をtestテーブルに付けると`ck_test_positive`になり、名前を付けないと`InvalidRequestError`（Naming convention including %(constraint\_name)s token requires that constraint is explicitly named.）で止まりました。運用中のDBに後から規則を入れても既存の制約名は変わらないため、そろえたい場合は改名のマイグレーションを別に書きます。最初のマイグレーションより前に入れておくのが一番手間がかかりません。

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

### エンジンとセッションの生成（SQLAlchemy 2.1の既定ドライバ）

```
from collections.abc import Iterator

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

from app.core.config import get_settings

settings = get_settings()
engine = create_engine(settings.database_url, echo=settings.sql_echo, pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

def get_db() -> Iterator[Session]:
    with SessionLocal() as session:
        yield session
```

SQLAlchemy 2.1では、ドライバ名を省いた`postgresql://`の既定DBAPIがpsycopg2からpsycopg（psycopg 3）に変わりました。検証環境で`create_engine`の`dialect.driver`を確かめると`psycopg`でした。PostgreSQLなら`pip install "psycopg[binary]"`を入れ、URLを`postgresql+psycopg://`と明示しておくと、2.0と2.1のどちらでも同じドライバで動きます。両ドライバの違いは[psycopg2とpsycopg3の違い](/tech/details/8225/)にまとめています。MySQLは`mysql+pymysql://`、SQL Serverは`mssql+pyodbc://`のようにURLとドライバを替えます。

`pool_pre_ping=True`はプールから接続を取り出すときに生存確認をして、DB側で切れた接続を再利用するエラーを防ぎます。処理やトランザクションの途中で切れた接続までは救えません（[SQLAlchemyの接続プールのドキュメント](https://docs.sqlalchemy.org/en/21/core/pooling.html)）。`autoflush=False`は、2.1でautoflushがORM文かCore文かを問わず全実行で走るようになったこともあり、flushの時機をリポジトリ側で決めるために切っています。`expire_on_commit=False`はcommit後に属性を読み直しに行かない設定で、commitの後にレスポンスへ変換するFastAPIの流れでは余分なSELECTを防ぎます。

### get\_dbの依存性注入とcommitの置き場所

```
from typing import Annotated

from fastapi import Depends
from sqlalchemy.orm import Session

from app.db.session import get_db

SessionDep = Annotated[Session, Depends(get_db)]
```

エンドポイントでは`session: SessionDep`と書くだけでセッションを受け取れます。FastAPIは同じリクエスト内で同じ依存を1回だけ呼んで結果を使い回すので（[サブ依存のドキュメント](https://fastapi.tiangolo.com/tutorial/dependencies/sub-dependencies/)）、1リクエストの中で複数のリポジトリが同じセッションを共有します。

迷いやすいのはcommitの置き場所です。yieldの後に`session.commit()`を書く例をよく見かけます。

```
# 避ける書き方：yieldの後でcommitする
def get_db_commit_on_exit():
    with SessionLocal() as session:
        yield session
        session.commit()  # FastAPI 0.118.0以降はレスポンス送信後に実行される
```

FastAPI 0.118.0以降、yieldの後の処理はレスポンスを送ってから実行されます。commit時に検査される一意制約（`DEFERRABLE INITIALLY DEFERRED`）を付けたテーブルへ同じ値を2回送ると、2回目はcommitが`IntegrityError`で失敗したのに、クライアントには200が返りました。DBに残った行は1件です。

```
$ curl -sS -w ' %{http_code}
' -X POST localhost:8000/items/1   # 1回目
{"saved":1} 200
$ curl -sS -w ' %{http_code}
' -X POST localhost:8000/items/1   # 2回目（commitで一意制約違反）
{"saved":1} 200
# サーバーログ
sqlalchemy.exc.IntegrityError: (psycopg.errors.UniqueViolation) duplicate key value violates unique constraint "t_uq"
```

標準の対策は、commitをサービス層（小規模ならエンドポイント）で明示的に呼ぶことです。commitの例外がエンドポイントの中で起きるので、409や500として正しく返せます。後始末をどうしても依存関数の側で済ませたい場合は、FastAPI 0.121.0で入った`scope`を使います。

```
# yieldの後のcommitをレスポンス送信前に終わらせたい場合
SessionCommitDep = Annotated[Session, Depends(get_db_commit_on_exit, scope="function")]
```

`scope="function"`にすると、yieldの後の処理がレスポンス送信前に実行されます（[Advanced Dependenciesのドキュメント](https://fastapi.tiangolo.com/advanced/advanced-dependencies/)）。上と同じcommit付きの依存関数をこの指定で使うと、同じ値の2回目は500（Internal Server Error）で返り、失敗がクライアントに届きました。`scope`だけ変えても、commitを書いていない`get_db`のままでは何も保存されない点に注意します。後始末に1秒かかる依存で測ると、既定（`scope="request"`）のレスポンスは0.02〜0.09秒、`scope="function"`は1.17秒で、後始末を待ってから返っていました。ストリーミングのレスポンスでセッションを使い続けるなら既定のままにします。

## リポジトリパターンとサービス層の分け方

```
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models import User

class UserRepository:
    def __init__(self, session: Session) -> None:
        self.session = session

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

    def get_by_email(self, email: str) -> User | None:
        return self.session.scalar(select(User).where(User.email == email))

    def add(self, user: User) -> User:
        self.session.add(user)
        self.session.flush()
        return user
```

```
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.models import User
from app.repositories.user import UserRepository
from app.schemas.user import UserCreate

class EmailAlreadyExists(Exception):
    pass

def register_user(session: Session, data: UserCreate) -> User:
    repo = UserRepository(session)
    if repo.get_by_email(data.email):
        raise EmailAlreadyExists(data.email)
    try:
        user = repo.add(User(email=data.email, name=data.name))
        session.commit()  # トランザクションの境界はサービス層で決める
    except IntegrityError as exc:
        # 事前確認とINSERTの間に同じメールが登録された（同時登録）場合
        session.rollback()
        if repo.get_by_email(data.email) is not None:
            raise EmailAlreadyExists(data.email) from exc
        raise
    return user
```

リポジトリは1テーブルの読み書きに徹し、`flush()`までにとどめます。commitの判断をサービス層が持つので、複数のリポジトリにまたがる処理を1トランザクションにまとめられます。`register_user`はメールの重複を業務エラー（`EmailAlreadyExists`）として送出し、HTTPの409への変換はルーターに任せています。事前の重複確認だけでは、確認とINSERTの間に同じメールが登録される同時登録を防げないため、`IntegrityError`をrollbackしてから同じ業務エラーに変換しています。この処理を外した版では、後述の3本目のテストが`sqlite3.IntegrityError: UNIQUE constraint failed: users.email`で落ちました。サービス層が`HTTPException`を知らない形にしておくと、同じ処理をバッチやCLIからも呼べます。

小規模では採用しない判断もありえます。エンドポイントが数本で、同じクエリが2か所以上に出てこないうちは、リポジトリを作らずエンドポイントからセッションを直接使って構いません。同じ`select`が複数のルーターに現れた時点、あるいはテストでDBアクセスを差し替えたくなった時点で切り出します。FastAPIを使ったAPIの設計・開発を外部に依頼したい場合は、issohの[API開発・システム連携](/service/system/api/)でも支援しています。

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

```
from pydantic import BaseModel, ConfigDict, Field

class UserCreate(BaseModel):
    email: str = Field(max_length=255)
    name: str | None = Field(default=None, max_length=100)

class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    name: str | None
```

```
from fastapi import APIRouter, HTTPException, status

from app.api.deps import SessionDep
from app.repositories.user import UserRepository
from app.schemas.user import UserCreate, UserRead
from app.services.user import EmailAlreadyExists, register_user

router = APIRouter(prefix="/users", tags=["users"])

@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED)
def create_user(data: UserCreate, session: SessionDep):
    try:
        return register_user(session, data)
    except EmailAlreadyExists:
        raise HTTPException(status.HTTP_409_CONFLICT, "email already registered")

@router.get("/{user_id}", response_model=UserRead)
def read_user(user_id: int, session: SessionDep):
    user = UserRepository(session).get(user_id)
    if user is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "user not found")
    return user
```

`response_model`や戻り値の型で出力項目を絞らずにSQLAlchemyのオブジェクトを返すと、パスワードハッシュのような外に出したくない属性まで応答に含まれるおそれがあります。入力用（`UserCreate`）と出力用（`UserRead`）を分け、出力側に`model_config = ConfigDict(from_attributes=True)`を付けます。v1の`orm_mode = True`がこの名前に変わったもので、FastAPI 0.128.0では`pydantic.v1`のサポート自体が外れました。`response_model=UserRead`を指定すると、`UserRead`に定義した列だけがJSONになり、同じ型が[Swagger UI（/docs）](/tech/details/2246/)のスキーマにも出ます。ConfigDictのほかの項目は[Pydantic v2のConfigDict設定一覧](/tech/details/7583/)で整理しています。

## 同期と非同期の選び方（async defの落とし穴）

### async defの中で同期セッションを呼んだときの待ち時間

```
@app.get("/async")
async def in_async(session: SessionDep):
    session.execute(text("select pg_sleep(0.5)"))  # 同期セッションをasync defで呼ぶ
    return {"ok": True}

@app.get("/sync")
def in_sync(session: SessionDep):
    session.execute(text("select pg_sleep(0.5)"))
    return {"ok": True}
```

同じ`select pg_sleep(0.5)`を、httpxのAsyncClientから10本同時に送った結果です。エンジンは`pool_size=20`で作り、`uvicorn block:app`（ワーカー1つ）で起動しました。計測用のクライアントは次のとおりで、各経路に1回アクセスしてから`asyncio.gather`で10本を同時に送り、全応答までの経過時間を`time.perf_counter`で測っています。秒数はこの条件（Intel Mac・ローカルのPostgreSQL 18.6）での測定例です。

```
import asyncio, sys, time, httpx
async def main(path, n=10):
    async with httpx.AsyncClient(base_url="http://127.0.0.1:8000", timeout=60) as c:
        await c.get(path)  # ウォームアップ
        t = time.perf_counter()
        await asyncio.gather(*[c.get(path) for _ in range(n)])
        print(path, n, "requests:", round(time.perf_counter() - t, 2), "s")
asyncio.run(main(sys.argv[1]))
```

```
/async 10 requests: 5.13 s
/sync 10 requests: 0.58 s
/async 10 requests: 5.06 s
/sync 10 requests: 0.56 s
```

`async def`版は5.13秒、`def`版は0.58秒でした。`async def`の関数はイベントループの上で直接動くため、同期ドライバの待ち時間がループを止め、リクエストが1本ずつ処理されます。`def`の関数はFastAPIがスレッドプールで動かすので並行に進みます。同期のSQLAlchemyを使うなら、エンドポイントもyieldの依存関数も`def`で書きます。

### AsyncSessionに切り替える条件と必要な設定

```
# pip install "sqlalchemy[asyncio]" "psycopg[binary]"
from collections.abc import AsyncIterator

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import selectinload

from app.models import User

engine = create_async_engine("postgresql+psycopg://user:pass@localhost/app")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

async def get_user(session: AsyncSession, user_id: int) -> User | None:
    # 関連は取得時に読み込む（遅延ロードはMissingGreenletになる）
    return await session.scalar(
        select(User).where(User.id == user_id).options(selectinload(User.posts))
    )
```

非同期で揃えるなら、エンジンからエンドポイントまで全部を非同期にします。SQLAlchemy 2.1ではgreenletが既定で入らなくなり、入れずに`create_async_engine`を使うと`ImportError: The SQLAlchemy asyncio module requires that the Python 'greenlet' library is installed.`で止まりました。もう1つの落とし穴は遅延ロードで、AsyncSessionで取得したユーザーの`user.posts`をそのまま読むと`MissingGreenlet`になったため、`selectinload(User.posts)`で先に読み込みます。`expire_on_commit=False`は、SQLAlchemyの[asyncio拡張のドキュメント](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)が非同期では通常Falseにすべきとしている設定です。

非同期への切り替えは、スレッドの空き待ちが応答時間を延ばしていると測定で確かめられたときに検討します。`def`のエンドポイントを動かすAnyIOのスレッド数の上限は、検証環境（AnyIO 4.15.1）で40でした。[Starletteのスレッドプールの説明](https://starlette.dev/threadpool/)のとおり、この40は同期の依存関数などとも共有され、`total_tokens`で変更もできます。DB接続プールの待ち時間と合わせて測り、スレッド待ちが目立たず外部APIの呼び出しも少ないなら、同期のSQLAlchemyと`def`のほうがコードが単純で、遅延ロードの例外にも悩まされません。

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

```
pip install alembic
alembic init alembic
# ここで次のコード例のとおりalembic/env.pyを編集してから、以下を実行する
alembic revision --autogenerate -m "create users and posts"
alembic upgrade head
alembic check          # モデルとDBに差分が残っていないかを確認
```

```
# alembic/env.py（変更する箇所だけ）
from app.core.config import get_settings
from app.db.base import Base
from app import models  # noqa: F401  全モデルを登録する

config = context.config
config.set_main_option(
    "sqlalchemy.url", get_settings().database_url.replace("%", "%%")
)
target_metadata = Base.metadata
```

本番のテーブルは`Base.metadata.create_all()`で作らず、Alembicの履歴で管理します。`env.py`で`target_metadata`に`Base.metadata`を渡し、接続先を設定クラスから読むと、アプリとマイグレーションで接続先がずれません。

自動生成で気をつけたいのは、metadataに無いテーブルを削除対象として出す点です。検証中、DBに残っていた別のテーブル`t`に対して次のログが出て、生成スクリプトに`drop_table`が入りました。

```
INFO  [alembic.autogenerate.compare.tables] Detected added table 'posts'
INFO  [alembic.autogenerate.compare.tables] Detected removed table 't'
```

別のツールが作るテーブルが同じDBにあるなら、`env.py`の`include_object`で除外するか、生成スクリプトを適用前に必ず読みます。適用後は`alembic check`でモデルとDBに差分が残っていないかを確かめられ、今回は`No new upgrade operations detected.`と出ました。Alembicの使い方と自動生成の落とし穴は[Alembic（アレンビック）とは](/tech/details/3417/)で詳しく扱っています。

## テストのベストプラクティス：dependency\_overridesとSQLite

```
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.db.base import Base
from app.db.session import get_db
from app.main import app
from app import models  # noqa: F401  テーブル定義を登録する

@pytest.fixture
def client():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    TestingSession = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)

    def override_get_db():
        with TestingSession() as session:
            yield session

    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()
    engine.dispose()
```

```
def test_create_and_read_user(client):
    res = client.post("/users", json={"email": "a@example.com", "name": "A"})
    assert res.status_code == 201
    user_id = res.json()["id"]

    res = client.get(f"/users/{user_id}")
    assert res.status_code == 200
    assert res.json() == {"id": user_id, "email": "a@example.com", "name": "A"}

def test_duplicate_email_returns_409(client):
    client.post("/users", json={"email": "a@example.com"})
    res = client.post("/users", json={"email": "a@example.com"})
    assert res.status_code == 409

def test_concurrent_duplicate_returns_409(client, monkeypatch):
    from app.repositories.user import UserRepository

    client.post("/users", json={"email": "a@example.com"})
    original = UserRepository.get_by_email
    calls = {"n": 0}

    def miss_first_check(self, email):
        # 事前確認をすり抜けた重複INSERTを模擬する（実際の並行実行ではない）
        calls["n"] += 1
        return None if calls["n"] == 1 else original(self, email)

    monkeypatch.setattr(UserRepository, "get_by_email", miss_first_check)
    res = client.post("/users", json={"email": "a@example.com"})
    assert res.status_code == 409
```

```
$ pytest -q
...                                                                      [100%]
3 passed in 0.20s
```

- **差し替えるのはget\_dbだけ**：`app.dependency_overrides[get_db]`でセッションだけをテスト用に替え、ルーティングとスキーマの検証は本物のまま通します。
- **インメモリDBはStaticPoolで1接続に固定**：`sqlite://`は接続ごとに別のDBになるため、`poolclass=StaticPool`で同じ接続を使い回します。`check_same_thread=False`は、TestClientがエンドポイントを別スレッドで動かすので必要です。
- **差し替えは後片付けする**：テストの最後に`app.dependency_overrides.clear()`を呼び、他のテストに差し替えを残しません。

検証中に2つの罠を踏みました。1つ目は、conftest.pyで`import app.models`と書いた版で、名前`app`がFastAPIのインスタンスからパッケージ`app`に上書きされ、`AttributeError: module 'app' has no attribute 'dependency_overrides'`で全テストが落ちました。`from app import models`と書けば起きません。2つ目はStarlette 1.8.0のTestClientで、httpxのままだと`` Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead. ``という警告が出て、`pip install httpx2`で消えました。SQLiteで再現できない差（JSON型・部分インデックス・遅延評価の制約など）を確かめるテストは、本番と同じ版のPostgreSQLに専用のテストDBを用意し、別のfixtureでその接続URLからエンジンを作って`get_db`を差し替えます。掲載のfixtureはSQLite固定なので、`DATABASE_URL`を変えてもテスト先は変わりません。PostgreSQL用のfixtureでは`StaticPool`と`check_same_thread`を外し、テストごとにテーブルの作成と削除を行います。

## DBアクセスで避けるべきアンチパターン

ここまでの章で扱った「async defの中の同期セッション」「yieldの後のcommit」のほかに、本番で問題になりやすいものを挙げます。

### N+1問題（20件の取得で21クエリ）

```
users = session.scalars(select(User)).all()
total = sum(len(u.posts) for u in users)          # 1人ごとにSELECTが飛ぶ

users = session.scalars(
    select(User).options(selectinload(User.posts))  # 投稿をまとめて取得
).all()
```

```
lazy users= 20 posts= 60 queries= 21
selectinload users= 20 posts= 60 queries= 2
```

ユーザー20人・各3件の投稿で数えると、遅延ロードのままループで`user.posts`を読むと21クエリ、`selectinload`を付けると2クエリでした。一覧APIで関連を返すときは、取得時に`selectinload`か`joinedload`を指定します。

### セッションのグローバル共有と起動時のcreate\_all

モジュールの先頭でSessionを1つ作って使い回すと、並行するリクエストの変更が同じトランザクションに混ざります。セッションは必ず`get_db`でリクエストごとに作ります。本番の起動処理（lifespan）で`create_all()`を呼ぶ構成も避けます。`create_all()`は存在しないテーブルを作るだけで、既存テーブルへの列の追加や型の変更は反映しないため、Alembicの履歴とDBの実体がずれていきます。lifespanでは、この記事の`main.py`のように終了時の`engine.dispose()`だけを書いておけば足ります。

```
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.api.routers import users
from app.db.session import engine

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    engine.dispose()  # 終了時に接続プールを閉じる

app = FastAPI(lifespan=lifespan)
app.include_router(users.router)
```

## よくある質問

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

公式に決まった構成はありません。公式チュートリアルはAPIRouterでファイルを分ける方法までを示しています。DBを使うアプリなら、models・schemas・repositories・services・apiに分けるファイル種別の構成で始め、ドメインが増えて1つのディレクトリに別機能のファイルが混ざり始めたらドメイン単位の構成へ移すのが扱いやすい順序です。

### SQLAlchemyは2.0と2.1のどちらで書くべきですか？

Python 3.11以上が使えるなら2.1で始めます。モデルの書き方（`DeclarativeBase`・`Mapped[]`・`mapped_column()`）は2.0と同じです。2.0から上げるときは、既定ドライバ（psycopg 3）・greenlet・autoflushの3つの変更を確認し、Python 3.10の環境は2.0系に留めます。

### Pydanticのorm\_modeが使えないのはなぜですか？

Pydantic v2で`from_attributes`に名前が変わったためです。`model_config = ConfigDict(from_attributes=True)`と書くと、SQLAlchemyのオブジェクトから属性を読んでスキーマを作れます。FastAPIは0.128.0で`pydantic.v1`のサポートを外したため、v1の書き方を残したコードは最新版のFastAPIでは動きません。

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

必須ではありません。エンドポイントが数本で、同じクエリが2か所以上に出てこないうちは、エンドポイントからセッションを直接使っても回ります。切り出す目安は本文のとおりですが、commitをサービス層（またはエンドポイント）で呼ぶ形だけは、規模にかかわらず最初からそろえておきます。

### PostgreSQL・MySQL・SQL Serverで同じコードが動きますか？

SQLAlchemyのORMはDBの違いを吸収するので、接続URLとドライバを替えれば大部分は共通です。PostgreSQLは`postgresql+psycopg://`、MySQLは`mysql+pymysql://`、SQL Serverは`mssql+pyodbc://`が代表的な指定で、SQLAlchemy 2.1ではSQL Server向けにmssql-pythonドライバへの対応も入りました。JSON型や全文検索などDB固有の機能を使う箇所は差が出るため、テストは本番と同じDBでも回します。

## 関連記事

- [FastAPIとは｜最短で動かす手順とasync/defの使い分け【2026年版】](/tech/details/3859/)
- [Alembic（アレンビック）とは？SQLAlchemyのマイグレーションの使い方とautogenerateの落とし穴【1.20対応】](/tech/details/3417/)
- [SQLModelのモデル定義とテーブル作成｜unique制約・CRUD・リレーション・FastAPI連携【0.0.42】](/tech/details/10322/)
- [FastAPIのSwagger UI（/docs）の使い方｜表示・設定・本番で隠す手順【2026年版】](/tech/details/2246/)
- [Uvicornとは？FastAPIを動かすPython製ASGIサーバーの基礎から本番デプロイまで](/tech/details/4843/)

---

出典: [FastAPIのベストプラクティス｜ディレクトリ構成・SQLAlchemy 2.1の設計・テストを実測で解説](<https://www.issoh.co.jp/tech/details/4347/>)（株式会社一創）