FastAPIのサンプルを実用構成へ育てる
FastAPIは、短いコードでAPIを作れるPythonのWebフレームワークです。
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def index():
return {"message": "Hello"}
この手軽さは魅力ですが、DBを使う実用APIになると考えることが急に増えます。
DB接続はどこで作り、いつ閉じるのか
入力用と出力用のモデルを分けるべきか
存在しないIDは何を返すのか
PATCHで未指定とnullをどう区別するか
テーブル変更をどう管理するか
テスト用DBへどう差し替えるか
CORSで何を許可すべきか
この記事では、タスク管理APIを作りながら、これらを一つずつ解決します。
完成するAPIは次の通りです。
| Method | Path | 処理 |
|---|---|---|
POST |
/api/tasks |
タスク作成 |
GET |
/api/tasks |
一覧・絞り込み・ページング |
GET |
/api/tasks/{id} |
1件取得 |
PATCH |
/api/tasks/{id} |
一部更新 |
DELETE |
/api/tasks/{id} |
削除 |
GET |
/health |
ヘルスチェック |
DBは最初にSQLiteを使います。SQLAlchemyをDBアクセスの境界に置くため、後からPostgreSQLへ移行しやすい構成です。
使用する技術と役割
FastAPI
HTTP、ルーティング、依存性注入、OpenAPI
Pydantic
リクエスト検証、レスポンス変換
SQLAlchemy 2.0
DB接続、ORM、トランザクション
Alembic
テーブル変更履歴の管理
pytest + TestClient
APIテスト
FastAPIは特定のDBライブラリを強制しません。この記事では、仕組みを明確に理解するためSQLAlchemy 2.0を直接使います。
プロジェクトを準備する
Python 3.11以上を想定します。フォルダを作り、仮想環境を用意します。
mkdir task-api
cd task-api
python -m venv .venv
.\.venv\Scripts\Activate.ps1
macOSまたはLinuxでは有効化コマンドが異なります。
source .venv/bin/activate
必要なパッケージをインストールします。
python -m pip install fastapi "uvicorn[standard]" sqlalchemy alembic pydantic-settings pytest httpx
インストール後、依存関係を保存します。
python -m pip freeze > requirements.txt
本番プロジェクトでは、意図しない更新を避けるためバージョンを固定し、定期的に更新します。
ファイル構成
次の構成にします。
task-api/
├─ app/
│ ├─ __init__.py
│ ├─ config.py
│ ├─ db.py
│ ├─ models.py
│ ├─ schemas.py
│ ├─ main.py
│ └─ routers/
│ ├─ __init__.py
│ └─ tasks.py
├─ tests/
│ ├─ __init__.py
│ └─ test_tasks.py
├─ alembic/
├─ alembic.ini
├─ .env
└─ requirements.txt
小規模な段階から、HTTP、DB、設定、スキーマを分けます。すべてをmain.pyへ書くと、テスト用DBへの差し替えや機能追加が難しくなるためです。
環境変数を設定クラスへ集める
app/config.pyを作ります。
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
app_name: str = "Task API"
database_url: str = "sqlite:///./task.db"
allowed_origins: str = "http://localhost:5173"
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
@property
def origins(self) -> list[str]:
return [
origin.strip()
for origin in self.allowed_origins.split(",")
if origin.strip()
]
@lru_cache
def get_settings() -> Settings:
return Settings()
.envを作ります。
APP_NAME=Task API
DATABASE_URL=sqlite:///./task.db
ALLOWED_ORIGINS=http://localhost:5173,http://127.0.0.1:5173
設定値をコードの各所で直接os.getenv()するより、一つの型付きクラスに集めた方が、何が必要か分かりやすくなります。
本番のパスワードやAPIキーを.envへ入れる場合は、.gitignoreへ追加します。
.venv/
__pycache__/
.pytest_cache/
.env
*.db
DB接続とSessionを作る
app/db.pyを作ります。
from collections.abc import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from app.config import get_settings
settings = get_settings()
connect_args = (
{"check_same_thread": False}
if settings.database_url.startswith("sqlite")
else {}
)
engine = create_engine(
settings.database_url,
connect_args=connect_args,
pool_pre_ping=True,
)
SessionLocal = sessionmaker(
bind=engine,
class_=Session,
autoflush=False,
expire_on_commit=False,
)
class Base(DeclarativeBase):
pass
def get_db() -> Generator[Session, None, None]:
with SessionLocal() as session:
yield session
engineはDBとの接続方法とコネクションプールを管理します。SessionLocalはSessionを作るファクトリです。
重要なのは、グローバルなSessionを1個作って使い回さないことです。SQLAlchemyのSessionは状態を持ち、1つのDBトランザクションを表します。並行するリクエスト間で共有するものではありません。
FastAPIのyield依存関係を使うと、リクエストごとにSessionを作り、処理後に確実に閉じられます。
リクエスト開始
↓
Sessionを作る
↓
エンドポイントへ注入
↓
レスポンス処理
↓
Sessionを閉じる
expire_on_commit=Falseは、commit後にレスポンスへ変換するとき、属性を読むための再問い合わせを避けやすくします。
ORMモデルを定義する
app/models.pyを作ります。
from datetime import datetime, timezone
from enum import Enum
from sqlalchemy import DateTime, Enum as SqlEnum, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.db import Base
def utc_now() -> datetime:
return datetime.now(timezone.utc)
class TaskStatus(str, Enum):
todo = "todo"
doing = "doing"
done = "done"
class Task(Base):
__tablename__ = "tasks"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200), index=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
status: Mapped[TaskStatus] = mapped_column(
SqlEnum(TaskStatus, native_enum=False),
default=TaskStatus.todo,
index=True,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=utc_now,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=utc_now,
onupdate=utc_now,
)
SQLAlchemy 2.0ではMappedとmapped_column()を使う宣言形式が標準です。型情報がORMマッピングにもエディター補完にも利用されます。
TaskStatusを文字列Enumにしておくと、DBとAPIで許可する状態をtodo、doing、doneへ限定できます。
APIスキーマをORMモデルと分ける
app/schemas.pyを作ります。
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.models import TaskStatus
class TaskCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str = Field(min_length=1, max_length=200)
description: str | None = Field(default=None, max_length=5000)
status: TaskStatus = TaskStatus.todo
@field_validator("title")
@classmethod
def title_must_not_be_blank(cls, value: str) -> str:
value = value.strip()
if not value:
raise ValueError("title must not be blank")
return value
class TaskUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
title: str | None = Field(default=None, min_length=1, max_length=200)
description: str | None = Field(default=None, max_length=5000)
status: TaskStatus | None = None
@field_validator("title")
@classmethod
def title_must_not_be_blank(cls, value: str | None) -> str | None:
if value is None:
raise ValueError("title must not be null")
value = value.strip()
if not value:
raise ValueError("title must not be blank")
return value
@field_validator("status")
@classmethod
def status_must_not_be_null(
cls, value: TaskStatus | None
) -> TaskStatus | None:
if value is None:
raise ValueError("status must not be null")
return value
class TaskRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
title: str
description: str | None
status: TaskStatus
created_at: datetime
updated_at: datetime
class TaskList(BaseModel):
items: list[TaskRead]
total: int
offset: int
limit: int
ORMモデルとAPIスキーマには役割の違いがあります。
Task
DBのテーブル構造
TaskCreate
作成時にクライアントが送れる項目
TaskUpdate
更新時に送れる項目
TaskRead
APIが返す項目
もしORMモデルをそのまま入力に使うと、将来owner_idやis_adminのような内部項目を追加したとき、意図せず外部から書き換えられる危険があります。
extra="forbid"により、APIが知らない入力項目は無視せず422エラーにします。クライアントのタイプミスを早く発見できます。
CRUDルーターを実装する
app/routers/tasks.pyを作ります。
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.db import get_db
from app.models import Task, TaskStatus
from app.schemas import TaskCreate, TaskList, TaskRead, TaskUpdate
router = APIRouter(prefix="/tasks", tags=["tasks"])
DbSession = Annotated[Session, Depends(get_db)]
def find_task_or_404(task_id: int, db: Session) -> Task:
task = db.get(Task, task_id)
if task is None:
raise HTTPException(status_code=404, detail="Task not found")
return task
@router.post("", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
def create_task(payload: TaskCreate, db: DbSession) -> Task:
task = Task(**payload.model_dump())
db.add(task)
db.commit()
db.refresh(task)
return task
@router.get("", response_model=TaskList)
def list_tasks(
db: DbSession,
task_status: TaskStatus | None = Query(default=None, alias="status"),
q: str | None = Query(default=None, min_length=1, max_length=100),
offset: int = Query(default=0, ge=0),
limit: int = Query(default=20, ge=1, le=100),
) -> TaskList:
conditions = []
if task_status is not None:
conditions.append(Task.status == task_status)
if q is not None:
conditions.append(Task.title.contains(q.strip()))
total_statement = select(func.count()).select_from(Task).where(*conditions)
total = db.scalar(total_statement) or 0
statement = (
select(Task)
.where(*conditions)
.order_by(Task.created_at.desc(), Task.id.desc())
.offset(offset)
.limit(limit)
)
items = list(db.scalars(statement).all())
return TaskList(items=items, total=total, offset=offset, limit=limit)
@router.get("/{task_id}", response_model=TaskRead)
def get_task(task_id: int, db: DbSession) -> Task:
return find_task_or_404(task_id, db)
@router.patch("/{task_id}", response_model=TaskRead)
def update_task(task_id: int, payload: TaskUpdate, db: DbSession) -> Task:
task = find_task_or_404(task_id, db)
changes = payload.model_dump(exclude_unset=True)
for field, value in changes.items():
setattr(task, field, value)
db.commit()
db.refresh(task)
return task
@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, db: DbSession) -> Response:
task = find_task_or_404(task_id, db)
db.delete(task)
db.commit()
return Response(status_code=status.HTTP_204_NO_CONTENT)
ここではSQLAlchemy 2.0形式のselect()とSession.scalars()を使っています。古い記事に多いsession.query(Task)はレガシーAPIなので、新規コードでは2.0形式に揃えると混乱が減ります。
PATCHでは未指定とnullを区別する
部分更新で重要なのがexclude_unset=Trueです。
changes = payload.model_dump(exclude_unset=True)
例えば現在のdescriptionが"説明"の場合を考えます。
{}
これはdescriptionを変更しません。
{
"description": null
}
これはdescriptionを明示的に空へ戻します。
単純にpayload.model_dump()すると、送られていない項目もデフォルトのNoneになり、意図せず既存値を消す可能性があります。
この更新スキーマでは、titleとstatusは「未指定なら変更しない、明示的nullなら422」としています。descriptionだけは明示的nullで空へ戻せます。APIの仕様としてnullをどう扱うか、項目ごとに決めることが重要です。
アプリケーションを組み立てる
app/main.pyを作ります。
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import get_settings
from app.routers import tasks
def create_app() -> FastAPI:
settings = get_settings()
app = FastAPI(
title=settings.app_name,
version="1.0.0",
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.origins,
allow_credentials=True,
allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"],
allow_headers=["Authorization", "Content-Type"],
)
@app.get("/health", tags=["system"])
def health() -> dict[str, str]:
return {"status": "ok"}
app.include_router(tasks.router, prefix="/api")
return app
app = create_app()
create_app()へ分けておくと、設定の異なるアプリをテストで作りやすくなります。
この時点では、起動時にBase.metadata.create_all()を実行していません。テーブル作成と変更はAlembicに担当させます。
CORSの*で解決しない
CORSのoriginは、プロトコル、ホスト、ポートの組み合わせです。
http://localhost:5173
http://127.0.0.1:5173
https://example.com
これらはすべて別のoriginです。
開発中につながらないからと、次のように全許可したくなることがあります。
allow_origins=["*"]
しかしCookieやAuthorizationヘッダーなど認証情報を扱う場合、ワイルドカードでは期待通り動かない構成があります。許可するフロントエンドを明示した方が安全で、トラブルも追いやすくなります。
CORSはブラウザが適用する制約です。curlやサーバー間通信が成功しても、ブラウザだけ失敗する場合は、開発者ツールのNetworkタブでOPTIONSリクエストとレスポンスヘッダーを確認します。
Alembicを初期化する
プロジェクトルートで実行します。
alembic init alembic
alembic/env.pyで、アプリのBaseとモデルを読み込みます。重要な部分は次です。
from app.config import get_settings
from app.db import Base
from app import models
config.set_main_option("sqlalchemy.url", get_settings().database_url)
target_metadata = Base.metadata
from app import modelsは未使用に見えますが必要です。モデルモジュールをimportしなければ、TaskがBase.metadataへ登録されず、Alembicがテーブルを検出できません。
最初のマイグレーション候補を生成します。
alembic revision --autogenerate -m "create tasks table"
生成されたalembic/versions/*.pyを必ず開き、upgrade()とdowngrade()を確認します。
def upgrade() -> None:
op.create_table(
"tasks",
# columns...
)
def downgrade() -> None:
op.drop_table("tasks")
問題がなければDBへ反映します。
alembic upgrade head
現在の状態を確認します。
alembic current
alembic history
--autogenerateは完成したマイグレーションを保証する機能ではなく、モデルとDBを比較して候補を作る機能です。カラム名変更が「古いカラムを削除して新しいカラムを作る」と判定されることもあります。そのまま適用するとデータを失うため、レビューが必須です。
カラムを追加する流れ
例えば期限due_atを追加するとします。
最初にORMモデルへ追加します。
due_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
APIから読み書きするなら、TaskCreate、TaskUpdate、TaskReadにも追加します。
差分を生成します。
alembic revision --autogenerate -m "add due_at to tasks"
生成内容をレビューし、適用します。
alembic upgrade head
Gitへコミットするのはモデル変更だけではありません。対応するマイグレーションファイルも必ず含めます。
APIを起動する
開発サーバーを起動します。
uvicorn app.main:app --reload
次を開きます。
Swagger UI: http://127.0.0.1:8000/docs
ReDoc: http://127.0.0.1:8000/redoc
OpenAPI: http://127.0.0.1:8000/openapi.json
FastAPIは型定義とresponse_modelからOpenAPIを生成します。Swagger UI上でリクエストを送り、422エラーの内容も確認できます。
curlで動作確認する
タスクを作成します。
curl -X POST http://127.0.0.1:8000/api/tasks \
-H "Content-Type: application/json" \
-d '{"title":"APIを実装する","description":"CRUDを完成させる"}'
一覧を取得します。
curl "http://127.0.0.1:8000/api/tasks?status=todo&offset=0&limit=20"
部分更新します。
curl -X PATCH http://127.0.0.1:8000/api/tasks/1 \
-H "Content-Type: application/json" \
-d '{"status":"done"}'
削除します。
curl -X DELETE http://127.0.0.1:8000/api/tasks/1
Windows PowerShellではcurlの解釈が環境で異なることがあります。その場合はcurl.exeを明示するか、Invoke-RestMethodを使います。
$body = @{
title = "APIを実装する"
description = "CRUDを完成させる"
} | ConvertTo-Json
Invoke-RestMethod `
-Uri "http://127.0.0.1:8000/api/tasks" `
-Method Post `
-ContentType "application/json" `
-Body $body
HTTPステータスを正しく使う
このAPIでは次を使っています。
| Status | 意味 | 使用箇所 |
|---|---|---|
200 OK |
取得・更新成功 | GET、PATCH |
201 Created |
リソース作成成功 | POST |
204 No Content |
本文なしで削除成功 | DELETE |
404 Not Found |
指定IDが存在しない | 詳細・更新・削除 |
422 Unprocessable Content |
入力検証エラー | FastAPI/Pydantic |
500 Internal Server Error |
想定外のサーバーエラー | 未処理例外 |
存在しないIDに空の200を返す、作成に常に200を返す、といった曖昧なAPIはクライアント側の実装を難しくします。
commitとrollbackを理解する
db.add()しただけでは、変更は確定していません。
db.add(task)
Sessionが変更を追跡
db.flush()
SQLをDBへ送るがトランザクションは未確定
db.commit()
トランザクションを確定
db.rollback()
トランザクションを取り消す
複数処理を1つの操作として成功・失敗させたい場合は、途中でcommitしません。
def complete_task_and_write_log(db: Session, task: Task) -> None:
try:
task.status = TaskStatus.done
db.add(AuditLog(message=f"completed task {task.id}"))
db.commit()
except Exception:
db.rollback()
raise
タスク更新だけ成功し、監査ログだけ失敗する状態を避けます。
SQLAlchemyではflushに失敗したSessionを続けて使う場合、明示的なrollbackが必要です。IntegrityErrorなどを捕捉するときは忘れないようにします。
DB制約違反をAPIエラーへ変換する
例えばtitleを一意にする場合、アプリ側で事前確認するだけでは不十分です。
リクエストA: 同じtitleがないことを確認
リクエストB: 同じtitleがないことを確認
リクエストA: INSERT成功
リクエストB: INSERT成功
並行実行では両方が確認を通る可能性があります。最終的な整合性はDBのUNIQUE制約で保証します。
title: Mapped[str] = mapped_column(String(200), unique=True, index=True)
API側ではIntegrityErrorを捕捉します。
from sqlalchemy.exc import IntegrityError
try:
db.add(task)
db.commit()
except IntegrityError:
db.rollback()
raise HTTPException(status_code=409, detail="Task title already exists")
ただし、すべてのIntegrityErrorがtitle重複とは限りません。本番では制約名や原因を判定し、誤ったエラーメッセージを返さない設計が必要です。
テスト用DBへ依存関係を差し替える
tests/test_tasks.pyを作ります。
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import StaticPool
from app.db import Base, get_db
from app.main import app
test_engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestSessionLocal = sessionmaker(
bind=test_engine,
class_=Session,
autoflush=False,
expire_on_commit=False,
)
def override_get_db():
with TestSessionLocal() as session:
yield session
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
@pytest.fixture(autouse=True)
def reset_database():
Base.metadata.drop_all(bind=test_engine)
Base.metadata.create_all(bind=test_engine)
yield
Base.metadata.drop_all(bind=test_engine)
def test_create_and_get_task():
create_response = client.post(
"/api/tasks",
json={
"title": "テストを書く",
"description": "正常系を確認する",
},
)
assert create_response.status_code == 201
created = create_response.json()
assert created["id"] > 0
assert created["title"] == "テストを書く"
assert created["status"] == "todo"
get_response = client.get(f"/api/tasks/{created['id']}")
assert get_response.status_code == 200
assert get_response.json()["id"] == created["id"]
def test_rejects_blank_title():
response = client.post("/api/tasks", json={"title": " "})
assert response.status_code == 422
def test_returns_404_for_unknown_task():
response = client.get("/api/tasks/99999")
assert response.status_code == 404
assert response.json() == {"detail": "Task not found"}
def test_patch_keeps_omitted_fields():
created = client.post(
"/api/tasks",
json={"title": "元のタイトル", "description": "残す説明"},
).json()
response = client.patch(
f"/api/tasks/{created['id']}",
json={"status": "done"},
)
assert response.status_code == 200
updated = response.json()
assert updated["title"] == "元のタイトル"
assert updated["description"] == "残す説明"
assert updated["status"] == "done"
def test_delete_task():
created = client.post("/api/tasks", json={"title": "削除対象"}).json()
delete_response = client.delete(f"/api/tasks/{created['id']}")
assert delete_response.status_code == 204
assert delete_response.content == b""
get_response = client.get(f"/api/tasks/{created['id']}")
assert get_response.status_code == 404
FastAPIのdependency_overridesにより、本番コードのget_dbをテスト用へ差し替えています。
本番リクエスト
get_db → task.db
テスト
get_db → メモリ上のSQLite
sqlite://のインメモリDBを複数接続から同じ状態で使うため、ここではStaticPoolを指定しています。
テストを実行します。
pytest -q
FastAPIのTestClientはHTTPXを基盤にしており、通常の同期テスト関数からAPIを呼び出せます。
テストでcreate_all()を使ってよいのか
アプリ本体ではAlembicを使い、単体テストでは高速化のためBase.metadata.create_all()を使っています。
ただしこれだけでは、Alembicのマイグレーションが本当に最初から最後まで適用できるかを検証できません。
実用プロジェクトではテストを2種類に分けます。
API単体テスト
create_allで高速にDBを準備
マイグレーションテスト
空DBへalembic upgrade headを実行
必要なら既存バージョンからの更新も確認
モデルが正しくてもマイグレーションファイルが壊れていることはあります。本番更新前に、バックアップを復元した検証環境でマイグレーションを実行すると安全です。
同期SessionとAsyncSessionの選び方
FastAPIだからDBも必ずasyncにする必要はありません。
同期Sessionには次の利点があります。
コードとトランザクション境界が理解しやすい
対応ドライバーが多い
デバッグしやすい
一般的なCRUDでは十分なことが多い
AsyncSessionが向くのは、非同期DBドライバーを利用し、同時接続数やI/O待ちがボトルネックになり、チームがasyncのトランザクション管理を理解している場合です。
注意点は、1つのAsyncSessionも複数のasyncioタスクへ共有できないことです。SQLAlchemy公式の原則は「Session per thread、AsyncSession per task」です。
最初から複雑にするのではなく、負荷試験と計測結果を見て選びます。
SQLiteからPostgreSQLへ移行するとき
アプリの接続先を環境変数から受ける構成なので、基本的な変更はDATABASE_URLです。
DATABASE_URL=postgresql+psycopg://user:password@localhost:5432/taskdb
ドライバーをインストールします。
python -m pip install "psycopg[binary]"
ただし接続文字列を変えるだけで移行完了ではありません。
SQLiteとPostgreSQLの型差
大文字小文字を含む検索挙動
タイムゾーン
Enumの扱い
制約とインデックス
接続プール数
マイグレーション
既存データ移行
特にSQLiteは開発には便利ですが、同時書き込みや本番運用の性質がPostgreSQLとは異なります。本番がPostgreSQLなら、CIや結合テストでもPostgreSQLを使うテストを用意すると差異を早く発見できます。
ページング設計を改善する
今回のoffsetとlimitは理解しやすく、管理画面などに向きます。
GET /api/tasks?offset=40&limit=20
ただしデータ件数が非常に多い場合、深いoffsetは遅くなることがあります。また、ページ移動中にデータが追加・削除されると重複や取りこぼしが起こりえます。
タイムラインや大量データではカーソル方式を検討します。
GET /api/tasks?after_id=1200&limit=20
カーソル方式では、並び順を安定させる必要があります。created_atだけでは同時刻があり得るため、今回のようにcreated_atとidを組み合わせる考え方が重要です。
N+1問題を避ける
将来TaskにUserとのリレーションを追加し、一覧の各行でtask.ownerへアクセスすると、タスク件数分のSQLが追加されることがあります。
タスク一覧を取得するSQL: 1回
各タスクのownerを取得: 100回
合計: 101回
これがN+1問題です。
SQLAlchemyではselectinload()やjoinedload()を使い、必要な関連をまとめて読み込みます。
from sqlalchemy.orm import selectinload
statement = (
select(Task)
.options(selectinload(Task.owner))
.order_by(Task.id.desc())
)
常に全部を先読みすればよいわけではありません。レスポンスに必要な関連だけを読み、SQLログと計測で確認します。
ログへ何を残すか
最低限、次を追跡できると障害調査がしやすくなります。
時刻
ログレベル
リクエストID
HTTPメソッドとパス
ステータスコード
処理時間
例外情報
パスワード、Authorizationヘッダー、Cookie、個人情報をそのままログへ残してはいけません。
開発中にSQLを確認したい場合はcreate_engine(..., echo=True)も使えますが、本番では出力量と機密情報に注意します。
エラーを捕捉して常に200を返す設計も避けます。
{
"success": false,
"error": "not found"
}
HTTPステータスが200だと、監視、キャッシュ、クライアントが成功と判断してしまいます。HTTPのエラーはHTTPステータスで表現します。
本番前のチェックリスト
設定
秘密情報をGitへ含めていない
開発・テスト・本番のDATABASE_URLが分離されている
DB
Alembicファイルをレビューした
空DBへupgrade headできる
バックアップと復元を試した
必要な列にインデックスがある
API
入力と出力スキーマが分かれている
404、409、422などの挙動が決まっている
一覧にlimit上限がある
PATCHで未指定とnullを区別している
セキュリティ
CORSを必要なoriginへ限定した
認証・認可をエンドポイント単位で確認した
ログへ秘密情報を出していない
HTTPSを使用する
テスト
正常系だけでなく異常系がある
DB依存関係をテスト用へ差し替えている
マイグレーションを検証している
運用
ヘルスチェックがある
リクエストIDと処理時間を記録できる
タイムアウトと接続プールを調整した
認証を追加する場合は、「ログインできること」だけでなく、「別ユーザーのタスクをID変更で読めないこと」を必ずテストします。認証と認可は別の問題です。
よくある失敗
Sessionをグローバルに1個だけ作る
リクエスト間でトランザクション状態が混ざります。Sessionはリクエスト単位で作り、yield依存関係で閉じます。
ルーター内で毎回Sessionを作る
テストで差し替えにくく、1つのユースケース内でトランザクションを共有しにくくなります。依存性注入で外から受け取ります。
起動時に毎回create_all()する
新規テーブルは作れても、既存カラムの変更履歴は管理できません。本体はAlembicを使います。
--autogenerateを確認せず適用する
カラム名変更を削除と追加として生成し、データを失う可能性があります。生成ファイルはコードレビューします。
一覧APIに上限がない
データが増えたとき、一度に全件返してメモリ、DB、ネットワークを圧迫します。limitへ上限を付けます。
ORMオブジェクトを何でもレスポンスにする
内部列や秘密情報を将来追加した際、意図せず外部公開する危険があります。response_modelを明示します。
例外を握りつぶす
rollbackせずSessionを再利用したり、原因不明の200を返したりすると障害が拡大します。想定した例外だけ変換し、それ以外はログを残して再送出します。
まとめ
FastAPIでDB付きAPIを作るとき、重要なのはエンドポイントを短く書くことだけではありません。
HTTP入力
↓ Pydanticで検証
ユースケース
↓ リクエスト単位のSession
SQLAlchemy ORM
↓ トランザクション
DB
そして、テーブル構造はAlembicで履歴として管理します。
ORMモデルを変更
↓
autogenerateで候補作成
↓
人間がレビュー
↓
テストDBで適用
↓
本番へ適用
今回の構成で特に押さえたい点は次です。
Sessionをリクエスト間で共有しない
入力・更新・出力スキーマを分ける
SQLAlchemy 2.0のselect形式を使う
PATCHはexclude_unsetで未指定を守る
DB制約を整合性の最後の砦にする
Alembicの自動生成結果を必ずレビューする
依存関係を差し替えてAPIをテストする
この土台があれば、ユーザー認証、所有者による認可、タグ、監査ログ、PostgreSQL、バックグラウンド処理へ段階的に拡張できます。
最初から巨大なレイヤー構成にする必要はありません。しかし、Sessionの寿命、トランザクション境界、APIスキーマ、マイグレーション、テストの境界だけは、小さい段階から明確にしておくと後の作り直しを大きく減らせます。