FastAPIでSQLModel:PydanticとSQLAlchemyのボイラープレートなしで型安全なAPIを構築する

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

クイックスタート:5分で作るSQLModel API

SQLModelを本番環境で6ヶ月間使ってきた率直な感想はこうだ:FastAPIプロジェクトにおいて最も実用的なライブラリ選択の一つだということ。派手さがあるからではなく、API開発で最も面倒な部分——PydanticスキーマとSQLAlchemyモデルを同期させながら並行管理する作業——を完全に排除してくれるからだ。

必要なものをすべて1つのコマンドでインストールする:

pip install sqlmodel fastapi uvicorn

完全に動作するユーザーAPIの例:

from fastapi import FastAPI, HTTPException
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import Optional

# 単一モデル:ORMテーブル + Pydanticバリデーションを1クラスで定義
class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str

engine = create_engine("sqlite:///./app.db")
SQLModel.metadata.create_all(engine)

app = FastAPI()

@app.post("/users/", response_model=User)
def create_user(user: User):
    with Session(engine) as session:
        session.add(user)
        session.commit()
        session.refresh(user)
        return user

@app.get("/users/", response_model=list[User])
def list_users():
    with Session(engine) as session:
        return session.exec(select(User)).all()

uvicorn main:app --reloadで起動して/docsを開く。追加設定なしで完全に型付けされたSwagger UIが表示される。table=Trueフラグがこれを可能にしている。このフラグはSQLAlchemyのマッパーレジストリにクラスを登録しながら、Pydanticのバリデーション動作をすべて維持する。

詳細解説:SQLModelがPydanticとSQLAlchemyを橋渡しする仕組み

SQLModelは両方のライブラリの上に直接構築されている。table=Trueでクラスを定義すると、メタクラスがSQLAlchemyのORMテーブルとして登録する。このフラグなしでは純粋なPydanticモデルになる——クラス構文は同じだが、実行時の動作が異なる。

この違いが重要なのは、APIの境界とデータベーススキーマでほぼ常に別のモデルが必要になるからだ。以下が全プロジェクトで使うようになったパターンだ:

class UserBase(SQLModel):
    name: str
    email: str

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    hashed_password: str  # APIレスポンスには絶対に含めない

class UserCreate(UserBase):
    password: str  # 生パスワード、入力時のみ受け付ける

class UserRead(UserBase):
    id: int  # レスポンスでは必ずnon-nullが保証される

エンドポイントはコンテキストに応じた適切なモデルを使う:

@app.post("/users/", response_model=UserRead)
def create_user(user_data: UserCreate):
    hashed = hash_password(user_data.password)
    db_user = User(
        name=user_data.name,
        email=user_data.email,
        hashed_password=hashed
    )
    with Session(engine) as session:
        session.add(db_user)
        session.commit()
        session.refresh(db_user)
        return db_user

3つのクラス、1つの基底定義——フィールドの重複なし。UserCreateが入力バリデーションを担い、UserReadがシリアライズされたレスポンスの形を決め、Userがサーバーから絶対に漏れてはいけない機密フィールドを含む実際のデータベーステーブルにマッピングされる。苦い経験から学んだことだ:セキュリティレビューでhashed_passwordがAPIレスポンスに漏れていることが発覚した。テーブルモデルをそのままレスポンス型として使っていたからだ。

応用:リレーションシップ、クエリ、マイグレーション

テーブルのリレーションシップ定義

リレーションシップはSQLAlchemyの標準Relationship APIを使用し、SQLModel独自のインポートで利用できる。以下はUserにリンクしたPostモデルの例だ:

from sqlmodel import Relationship
from typing import List

class Post(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    content: str
    author_id: int = Field(foreign_key="user.id")
    author: Optional["User"] = Relationship(back_populates="posts")

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str
    posts: List[Post] = Relationship(back_populates="author")

JOINを使ったクエリは生のSQLAlchemyとまったく同じように書ける——新しい構文を覚える必要がない:

@app.get("/users/{user_id}/posts")
def get_user_posts(user_id: int):
    with Session(engine) as session:
        user = session.get(User, user_id)
        if not user:
            raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
        return user.posts

Alembicによるスキーママイグレーション

SQLModelはAlembicにきれいに統合できる。重要なステップはAlembicにSQLModel.metadataを指定することだ:

pip install alembic
alembic init migrations

migrations/env.pyの中で、メタデータが確実に読み込まれるようすべてのテーブルモデルをインポートしてから割り当てる:

from sqlmodel import SQLModel
from app.models import User, Post  # すべてのテーブルモデルを必ずインポートする

target_metadata = SQLModel.metadata

通常通りマイグレーションを生成して適用する:

alembic revision --autogenerate -m "postsテーブルを追加"
alembic upgrade head

非同期エンジンのセットアップ

高スループットAPIには、SQLModelが非同期セッションラッパーを提供している:

from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

# 本番環境のPostgreSQLにはasyncpgを使用する
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

@app.get("/users/")
async def list_users():
    async with AsyncSession(engine) as session:
        result = await session.exec(select(User))
        return result.all()

SQLiteにはaiosqlite、PostgreSQLにはasyncpgを非同期ドライバとしてインストールする。

実践的なヒント:本番環境6ヶ月使用が教えてくれたこと

表面的なAPIは1日で学べるほど小さい。しかし荒削りな部分はチュートリアルの例を超えて進んで初めて現れる——そしてその一部は本当に気づきにくい。

テーブルモデルをレスポンス型として直接公開しない

どこでもUserを使うのは簡便さの点で魅力的だ。しかし我慢してほしい。テーブルモデルには絶対にシリアライズしたくないフィールドが含まれることが多い——パスワードハッシュ、内部監査フラグ、論理削除のタイムスタンプなど。常に専用のUserReadクラスを通してレスポンスをルーティングすること。

遅延ロードされたリレーションシップでのDetachedInstanceErrorに注意

SQLAlchemyの遅延ロードは、アクセスするまでリレーションシップの取得を静かに先送りにする。セッションが閉じた後にリレーションシップにアクセスするとDetachedInstanceErrorが発生する。解決策はセッションスコープ内でeagerロードすることだ:

from sqlalchemy.orm import selectinload

statement = select(User).options(selectinload(User.posts))
user = session.exec(statement).first()

型アノテーションがデータベースのNullabilityを決定する

SQLModelはPythonの型ヒントから直接カラム制約を推測する。str型のフィールドはNOT NULLになる。Optional[str]またはstr | None型のフィールドはnullableになる。これを見落とすと、Alembicの自動生成が実際のスキーマから静かに乖離していく——たいてい最悪のタイミングで発覚する。

高速で独立したテストにはインメモリSQLiteを使う

import pytest
from sqlmodel import create_engine, Session, SQLModel
from fastapi.testclient import TestClient
from app.main import app, get_session

@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine("sqlite:///:memory:")
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

@pytest.fixture(name="client")
def client_fixture(session: Session):
    def override_get_session():
        return session
    app.dependency_overrides[get_session] = override_get_session
    return TestClient(app)

各テストはクリーンで独立したスキーマを取得し、テアダウンは不要だ。

Alembic自動生成には死角がある

自動生成はすべてを検出するわけではない——カスタム制約、サーバーサイドのデフォルト値、一部のインデックス型は手書きのマイグレーションスクリプトが必要だ。本番データベースに対してalembic upgrade headを実行する前に、必ず生成されたファイルを確認すること。

FastAPIの作者であるSebastián RamírezがSQLModelも構築した——だからこそ2つがこれほど自然にフィットする理由がよくわかる。6ヶ月経った今、私のチームは生のSQLAlchemyに手を伸ばす必要が一度もなかった。新しいFastAPIプロジェクトでは、今やこれが実験ではなく出発点になっている。

Share: