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

.
├── 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

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

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

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

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

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()

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")

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)

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'))

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

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)]

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

$ 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"

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

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

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

@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}

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

# 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))
    )

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

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

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": "[email protected]", "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": "[email protected]", "name": "A"}

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

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

    client.post("/users", json={"email": "[email protected]"})
    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": "[email protected]"})
    assert res.status_code == 409

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

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

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)