0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@cozy_666_techin株式会社Flexibility

SQLAlchemy入門 — FastAPIと組み合わせて使う

0
Posted at

はじめに

FastAPIの記事でDBを扱う場面が何度か出てきたが、ORM部分をずっと「省略」していた。

PythonのORMは大きく分けてDjango ORMとSQLAlchemyの2つがある。FastAPIと組み合わせて使うのは圧倒的にSQLAlchemyが多い。Django ORMはDjangoのエコシステムに強く依存しているのでFastAPIには使いにくい。

SQLAlchemyにはCoreとORMの2層があって、最初に「どっちを使えばいいの?」と迷ったので、そこから整理した。

インストール

pip install sqlalchemy
pip install psycopg2-binary  # PostgreSQLドライバ
# または
pip install asyncpg          # 非同期PostgreSQLドライバ

CoreとORMの2層構造

SQLAlchemyは2層に分かれている。

SQLAlchemy ORM   ← 今回メインで使う
      ↓
SQLAlchemy Core  ← SQL式言語・低レベルAPI
      ↓
DBAPI            ← psycopg2, asyncpgなど
      ↓
Database         ← PostgreSQL, MySQL, SQLiteなど

Core — SQL式言語。SQLに近い書き方でクエリを組み立てる。柔軟だが冗長。
ORM — Coreの上に乗ったオブジェクトマッパー。Django ORMやEloquentに近い感覚で使える。

新しく書くコードは基本的にORMを使う。Coreを直接使うのは複雑なクエリや性能が必要な場面。

モデルの定義（SQLAlchemy 2.0以降）

SQLAlchemy 2.0で書き方が大きく変わった。今はDeclarativeBaseと型アノテーションを使う書き方が推奨。

# models.py
from datetime    import datetime
from typing      import Optional
from sqlalchemy  import String, Text, ForeignKey, func
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
)

class Base(DeclarativeBase):
    pass


class Author(Base):
    __tablename__ = "authors"

    id:         Mapped[int]           = mapped_column(primary_key=True)
    name:       Mapped[str]           = mapped_column(String(50))
    email:      Mapped[str]           = mapped_column(String(100), unique=True)
    created_at: Mapped[datetime]      = mapped_column(default=func.now())

    # リレーション
    posts: Mapped[list["Post"]] = relationship(back_populates="author")


class Post(Base):
    __tablename__ = "posts"

    id:        Mapped[int]           = mapped_column(primary_key=True)
    title:     Mapped[str]           = mapped_column(String(200))
    body:      Mapped[str]           = mapped_column(Text)
    author_id: Mapped[int]           = mapped_column(ForeignKey("authors.id"))
    created_at: Mapped[datetime]     = mapped_column(default=func.now())

    # NullableなフィールドはOptionalで表現
    category_id: Mapped[Optional[int]] = mapped_column(
        ForeignKey("categories.id"), nullable=True
    )

    # リレーション
    author:   Mapped["Author"]           = relationship(back_populates="posts")
    category: Mapped[Optional["Category"]] = relationship()
    tags:     Mapped[list["Tag"]]        = relationship(
        secondary="post_tags",  # 中間テーブル
        back_populates="posts",
    )

Mapped[str]という型アノテーションを使うことで、エディタの補完が効くようになった。Django ORMのmodels.CharField()に近い感覚だが、Pythonの型ヒントと統合されている分モダンな書き方。

エンジンとセッション

# database.py
from sqlalchemy     import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://user:password@localhost/mydb"

# エンジン作成
engine = create_engine(
    DATABASE_URL,
    echo=True,      # SQLをログに出力（開発時）
    pool_size=5,    # コネクションプールのサイズ
)

# セッションファクトリ
SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
)

# テーブル作成
Base.metadata.create_all(bind=engine)

エンジンがDB接続の設定、セッションが実際のDB操作の単位。LaravelのDB Facadeに相当するのがエンジン、Eloquentのモデル操作がセッション経由で行われるイメージ。

FastAPIとの統合

FastAPIのDependsでセッションを注入するのが定番パターン。

# database.py
from sqlalchemy     import create_engine
from sqlalchemy.orm import sessionmaker, Session
from typing         import Generator

engine       = create_engine("postgresql://user:pass@localhost/mydb")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# main.py
from fastapi  import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from database import get_db
from models   import Author, Post

app = FastAPI()

# レスポンス用Pydanticモデル
class AuthorResponse(BaseModel):
    id:    int
    name:  str
    email: str

    model_config = {"from_attributes": True}

# リクエスト用Pydanticモデル
class AuthorCreate(BaseModel):
    name:  str
    email: str

@app.get("/authors", response_model=list[AuthorResponse])
def list_authors(db: Session = Depends(get_db)):
    return db.query(Author).all()

@app.get("/authors/{author_id}", response_model=AuthorResponse)
def get_author(author_id: int, db: Session = Depends(get_db)):
    author = db.query(Author).filter(Author.id == author_id).first()
    if author is None:
        raise HTTPException(status_code=404, detail="見つかりません")
    return author

@app.post("/authors", response_model=AuthorResponse, status_code=201)
def create_author(author: AuthorCreate, db: Session = Depends(get_db)):
    db_author = Author(**author.model_dump())
    db.add(db_author)
    db.commit()
    db.refresh(db_author)  # DBで生成されたidなどを取得
    return db_author

db.refresh(db_author)を忘れるとidやcreated_atなどDB側で生成される値が取れない。最初にここで詰まった。

CRUDの基本操作

from sqlalchemy.orm import Session
from models         import Author, Post

def crud_examples(db: Session):

    # ① 作成
    author = Author(name="田中", email="tanaka@example.com")
    db.add(author)
    db.commit()
    db.refresh(author)

    # ② 全件取得
    authors = db.query(Author).all()

    # ③ 条件で検索
    author = db.query(Author).filter(Author.email == "tanaka@example.com").first()

    # ④ 複数条件
    posts = db.query(Post).filter(
        Post.author_id == 1,
        Post.title.contains("Python"),
    ).all()

    # ⑤ ソート
    posts = db.query(Post).order_by(Post.created_at.desc()).all()

    # ⑥ ページネーション
    posts = db.query(Post).offset(20).limit(10).all()

    # ⑦ 更新
    db.query(Author).filter(Author.id == 1).update({"name": "田中太郎"})
    db.commit()

    # または取得してから更新
    author = db.query(Author).get(1)
    author.name = "田中太郎"
    db.commit()

    # ⑧ 削除
    db.query(Author).filter(Author.id == 1).delete()
    db.commit()

Django ORMのPost.objects.filter()に相当するのがdb.query(Post).filter()。セッション経由で操作するのがSQLAlchemyの特徴。

JOIN操作

from sqlalchemy.orm import joinedload, selectinload

# ① 遅延ロード（デフォルト、N+1が発生する）
posts = db.query(Post).all()
for post in posts:
    print(post.author.name)  # 各postごとにSELECTが発行される

# ② joinedload（JOIN、ForeignKey用）
posts = db.query(Post).options(
    joinedload(Post.author),
    joinedload(Post.category),
).all()

# ③ selectinload（IN句を使った別クエリ、ManyToMany用）
posts = db.query(Post).options(
    selectinload(Post.tags),
).all()

# ④ 明示的なJOIN
from sqlalchemy import join

posts = (
    db.query(Post, Author)
    .join(Author, Post.author_id == Author.id)
    .filter(Author.name == "田中")
    .all()
)

Django ORMのselect_related()がjoinedload()、prefetch_related()がselectinload()に相当する。N+1問題の対処方法は同じ発想。

集計クエリ

from sqlalchemy import func, desc

# 件数
count = db.query(func.count(Post.id)).scalar()

# GROUP BY + 集計
from sqlalchemy import func

results = (
    db.query(
        Author.name,
        func.count(Post.id).label("post_count"),
        func.max(Post.created_at).label("latest_post"),
    )
    .join(Post, Author.id == Post.author_id)
    .group_by(Author.id, Author.name)
    .order_by(desc("post_count"))
    .all()
)

for name, count, latest in results:
    print(f"{name}: {count}件 (最新: {latest})")

func.count()やfunc.max()のようにSQLの集計関数をPythonの関数として呼ぶ。Django ORMのCount()やMax()に相当する。

Alembic — マイグレーション管理

SQLAlchemy単体にはマイグレーション機能がないので、Alembicを使う。

pip install alembic

# 初期化
alembic init alembic

myproject/
├── alembic/
│   ├── env.py          # 設定ファイル
│   ├── script.py.mako  # マイグレーションテンプレート
│   └── versions/       # マイグレーションファイル
├── alembic.ini
├── models.py
└── main.py

# alembic/env.py の設定
from models import Base  # Baseをimport

target_metadata = Base.metadata  # ← 追加

# alembic.ini
sqlalchemy.url = postgresql://user:pass@localhost/mydb

# マイグレーションファイルを自動生成（Django の makemigrations に相当）
alembic revision --autogenerate -m "create authors table"

# マイグレーション実行（Django の migrate に相当）
alembic upgrade head

# 一つ前に戻す
alembic downgrade -1

# 現在の状態を確認
alembic current

Djangoのmakemigrationsと同じように、モデルの変更を検出して差分のマイグレーションファイルを自動生成してくれる。

# 自動生成されるマイグレーションファイル
def upgrade() -> None:
    op.create_table("authors",
        sa.Column("id",         sa.Integer(),     nullable=False),
        sa.Column("name",       sa.String(50),    nullable=False),
        sa.Column("email",      sa.String(100),   nullable=False),
        sa.Column("created_at", sa.DateTime(),    nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("email"),
    )

def downgrade() -> None:
    op.drop_table("authors")

非同期版SQLAlchemy

FastAPIで非同期エンドポイントを使う場合は非同期版を使う。

pip install asyncpg

# async_database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"

async_engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(
    async_engine,
    expire_on_commit=False,
)

async def get_async_db():
    async with AsyncSessionLocal() as session:
        yield session

# 非同期エンドポイント
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

@app.get("/authors")
async def list_authors(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(Author))
    return result.scalars().all()

@app.post("/authors", status_code=201)
async def create_author(author: AuthorCreate, db: AsyncSession = Depends(get_async_db)):
    db_author = Author(**author.model_dump())
    db.add(db_author)
    await db.commit()
    await db.refresh(db_author)
    return db_author

非同期版はdb.query()が使えずselect()を使う書き方になる。同期版と書き方が少し違うので最初は戸惑った。

CRUDの分離 — よくある設計パターン

FastAPIのプロジェクトではDB操作をCRUD関数として分離するパターンが多い。

myapp/
├── main.py
├── models.py
├── schemas.py    # Pydanticモデル
├── crud.py       # DB操作
└── database.py

# crud.py
from sqlalchemy.orm import Session
from models         import Author
from schemas        import AuthorCreate

def get_author(db: Session, author_id: int) -> Author | None:
    return db.query(Author).filter(Author.id == author_id).first()

def get_authors(db: Session, skip: int = 0, limit: int = 100) -> list[Author]:
    return db.query(Author).offset(skip).limit(limit).all()

def create_author(db: Session, author: AuthorCreate) -> Author:
    db_author = Author(**author.model_dump())
    db.add(db_author)
    db.commit()
    db.refresh(db_author)
    return db_author

def delete_author(db: Session, author_id: int) -> bool:
    author = get_author(db, author_id)
    if author is None:
        return False
    db.delete(author)
    db.commit()
    return True

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
import crud, schemas

app = FastAPI()

@app.get("/authors/{author_id}", response_model=schemas.AuthorResponse)
def read_author(author_id: int, db: Session = Depends(get_db)):
    author = crud.get_author(db, author_id)
    if author is None:
        raise HTTPException(status_code=404, detail="見つかりません")
    return author

ルーターはHTTPの関心事だけ、crud.pyはDB操作だけと責務を分離する。テストもしやすくなる。

Django ORMとSQLAlchemyの比較

項目 Django ORM SQLAlchemy
セットアップ Djangoに組み込み 独立・設定が必要
マイグレーション makemigrations（自動検出） Alembic（別途インストール）
クエリAPI Post.objects.filter() db.query(Post).filter()
セッション管理 自動（requestに紐づく） 手動（get_dbで注入）
非同期 限定的 ネイティブ対応
FastAPIとの相性 ✕ ◎
柔軟性 低め 高い
学習コスト 低い 中くらい

まとめ

  • SQLAlchemyはCoreとORMの2層。普段はORMを使う
  • FastAPIとの組み合わせはget_db()でセッションをDIするのが定番
  • db.refresh()を忘れるとDBで生成された値が取れない
  • N+1はForeignKeyにjoinedload()、ManyToManyにselectinload()で対処
  • マイグレーションはAlembicを使う
  • 非同期はAsyncSessionとselect()を使う書き方に変わる

Django ORMより設定が多いが、FastAPIと組み合わせる場合はSQLAlchemyが自然な選択。Alembicを入れてからDjangoのmakemigrationsと同じ感覚でマイグレーションを管理できるようになった。

0
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?