SQLAlchemy = PythonコードでDBを操作するためのORM / SQLツール
Alembic = DBスキーマ変更を管理するマイグレーションツール
SQLAlchemy 2.0 系では Mapped / mapped_column を使った型付きモデル定義が推奨されており、公式のORM Quick Startでもこの形式が使われています。Alembic は SQLAlchemy 向けの軽量マイグレーションツールで、モデル定義からマイグレーション候補を自動生成する --autogenerate も提供されています。(SQLAlchemy)
1. 役割の違い
SQLAlchemy の役割
SQLAlchemy は、PythonからDBを扱うためのライブラリです。
例えば、本来SQLでこう書く処理を、
SELECT * FROM users WHERE email = 'taro@example.com';
Pythonではこう書けます。
stmt = select(User).where(User.email == "taro@example.com")
user = session.scalar(stmt)
SQLAlchemy は主に次のことを担当します。
| 役割 | 内容 |
|---|---|
| モデル定義 | Pythonクラスでテーブル構造を書く |
| ORM | PythonオブジェクトとしてDBレコードを扱う |
| クエリ作成 |
select(), where(), join() など |
| トランザクション | commit / rollback |
| DB接続管理 | engine / session |
Alembic の役割
Alembic は、DBテーブルの変更履歴を管理します。
例えば、最初に users テーブルを作る、その後 name カラムを追加する、さらに status カラムを追加する、という変更をファイルとして残します。
migrations/
versions/
001_create_users.py
002_add_name_to_users.py
003_add_status_to_users.py
これにより、本番DB・ステージングDB・開発DBの構造を同じ状態に保てます。
2. 全体イメージ
app/models/user.py
↓
SQLAlchemyモデルを書く
alembic revision --autogenerate
↓
Alembicが差分を検出
alembic upgrade head
↓
PostgreSQLにテーブル変更を反映
つまり、
モデルを書く
↓
マイグレーションファイルを作る
↓
DBに反映する
という流れです。
3. サンプル構成
FastAPI プロジェクトなら、例えばこういう構成にします。
backend/
app/
main.py
db/
database.py
base.py
models/
user.py
schemas/
user.py
routers/
users.py
alembic/
env.py
versions/
alembic.ini
pyproject.toml
4. インストール
pip install fastapi uvicorn sqlalchemy alembic psycopg[binary] pydantic
PostgreSQL用ドライバは、最近なら psycopg v3 系を使う構成が分かりやすいです。
5. DB接続設定
app/db/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "postgresql+psycopg://app_user:password@localhost:5432/app_db"
engine = create_engine(
DATABASE_URL,
echo=True,
pool_pre_ping=True,
)
SessionLocal = sessionmaker(
bind=engine,
autocommit=False,
autoflush=False,
)
ポイント
| 項目 | 意味 |
|---|---|
create_engine() |
DB接続エンジンを作る |
SessionLocal |
DB操作用セッションを作るファクトリ |
echo=True |
実行SQLをログに出す。開発時に便利 |
pool_pre_ping=True |
切れたDB接続を事前チェックする |
6. SQLAlchemy の Base 定義
app/db/base.py
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
この Base を継承してモデルを作ります。
7. Userモデルの例
app/models/user.py
from datetime import datetime
from sqlalchemy import String, Boolean, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, index=True)
email: Mapped[str] = mapped_column(
String(255),
unique=True,
index=True,
nullable=False,
)
name: Mapped[str | None] = mapped_column(
String(100),
nullable=True,
)
is_active: Mapped[bool] = mapped_column(
Boolean,
nullable=False,
default=True,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
nullable=False,
)
このモデルが表すテーブル
users
id integer primary key
email varchar(255) unique not null
name varchar(100) null
is_active boolean not null
created_at timestamp with timezone
updated_at timestamp with timezone
よく使う指定
| 書き方 | 意味 |
|---|---|
primary_key=True |
主キー |
unique=True |
一意制約 |
index=True |
インデックス作成 |
nullable=False |
NULL禁止 |
default=True |
Python側のデフォルト値 |
server_default=func.now() |
DB側のデフォルト値 |
onupdate=func.now() |
更新時に値を変更 |
8. FastAPIでDBセッションを使う
app/main.py
from fastapi import FastAPI
from app.routers import users
app = FastAPI()
app.include_router(users.router)
app/db/session.py
from collections.abc import Generator
from sqlalchemy.orm import Session
from app.db.database import SessionLocal
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
FastAPI の Depends でこのセッションを注入します。
9. Pydantic スキーマ
app/schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel):
email: EmailStr
name: str | None = None
class UserRead(BaseModel):
id: int
email: EmailStr
name: str | None
is_active: bool
created_at: datetime
model_config = {
"from_attributes": True
}
SQLAlchemyモデルとPydanticの違い
| 種類 | 役割 |
|---|---|
| SQLAlchemyモデル | DBテーブル定義 |
| Pydanticスキーマ | APIの入力・出力定義 |
混同しやすいですが、DB用とAPI用は分けるのが基本です。
10. CRUD API の例
app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserRead
router = APIRouter(
prefix="/users",
tags=["users"],
)
@router.post(
"",
response_model=UserRead,
status_code=status.HTTP_201_CREATED,
)
def create_user(
payload: UserCreate,
db: Session = Depends(get_db),
):
stmt = select(User).where(User.email == payload.email)
existing_user = db.scalar(stmt)
if existing_user:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email already exists",
)
user = User(
email=payload.email,
name=payload.name,
)
db.add(user)
db.commit()
db.refresh(user)
return user
@router.get(
"/{user_id}",
response_model=UserRead,
)
def get_user(
user_id: int,
db: Session = Depends(get_db),
):
user = db.get(User, user_id)
if user is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
return user
@router.get(
"",
response_model=list[UserRead],
)
def list_users(
db: Session = Depends(get_db),
):
stmt = select(User).order_by(User.id.desc())
users = db.scalars(stmt).all()
return users
11. Alembic の初期化
プロジェクト直下で実行します。
alembic init alembic
すると、こういうファイルが作られます。
alembic/
env.py
versions/
alembic.ini
12. Alembic にSQLAlchemyモデルを認識させる
重要なのは alembic/env.py です。
Alembic が User モデルを見つけられるようにします。
alembic/env.py の主な修正例
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
from app.db.base import Base
from app.models.user import User # noqa: F401
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
target_metadata = Base.metadata が重要です。
これにより Alembic が SQLAlchemy のモデル定義を見て、DBとの差分を検出できます。
13. DB接続URLを設定する
alembic.ini
sqlalchemy.url = postgresql+psycopg://app_user:password@localhost:5432/app_db
本番では .env や環境変数から読む形にするのが推奨です。
14. 最初のマイグレーション作成
alembic revision --autogenerate -m "create users table"
すると、以下のようなファイルが作られます。
alembic/versions/xxxxxxxxxxxx_create_users_table.py
中身は例えばこうなります。
"""create users table
Revision ID: 123456789abc
Revises:
Create Date: 2026-06-15 10:00:00
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "123456789abc"
down_revision: Union[str, Sequence[str], None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"users",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("email", sa.String(length=255), nullable=False),
sa.Column("name", sa.String(length=100), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f("ix_users_id"), "users", ["id"], unique=False)
op.create_index(op.f("ix_users_email"), "users", ["email"], unique=True)
def downgrade() -> None:
op.drop_index(op.f("ix_users_email"), table_name="users")
op.drop_index(op.f("ix_users_id"), table_name="users")
op.drop_table("users")
upgrade と downgrade
| 関数 | 役割 |
|---|---|
upgrade() |
DBを新しい状態に進める |
downgrade() |
DBを前の状態に戻す |
15. マイグレーションをDBに適用
alembic upgrade head
これで PostgreSQL に users テーブルが作られます。
確認するなら、
alembic current
履歴を見るなら、
alembic history
戻すなら、
alembic downgrade -1
16. カラム追加の例
例えば、User に role カラムを追加します。
app/models/user.py
role: Mapped[str] = mapped_column(
String(50),
nullable=False,
default="user",
server_default="user",
)
追加後、マイグレーションを作ります。
alembic revision --autogenerate -m "add role to users"
生成される例です。
def upgrade() -> None:
op.add_column(
"users",
sa.Column(
"role",
sa.String(length=50),
server_default="user",
nullable=False,
),
)
def downgrade() -> None:
op.drop_column("users", "role")
DBへ反映します。
alembic upgrade head
17. よくある実装パターン
Repositoryパターンを使う場合
小規模なら router に直接SQLAlchemyを書いてもよいですが、中規模以上なら分離した方が保守しやすいです。
routers/
users.py API入口
services/
users.py 業務ロジック
repositories/
users.py DB操作
models/
user.py DBモデル
schemas/
user.py APIスキーマ
app/repositories/users.py
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.user import User
def find_user_by_email(
db: Session,
email: str,
) -> User | None:
stmt = select(User).where(User.email == email)
return db.scalar(stmt)
def find_user_by_id(
db: Session,
user_id: int,
) -> User | None:
return db.get(User, user_id)
def create_user(
db: Session,
email: str,
name: str | None,
) -> User:
user = User(
email=email,
name=name,
)
db.add(user)
db.commit()
db.refresh(user)
return user
app/services/users.py
from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from app.repositories.users import create_user, find_user_by_email
from app.schemas.user import UserCreate
from app.models.user import User
def register_user(
db: Session,
payload: UserCreate,
) -> User:
existing_user = find_user_by_email(db, payload.email)
if existing_user:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email already exists",
)
return create_user(
db=db,
email=payload.email,
name=payload.name,
)
app/routers/users.py
from fastapi import APIRouter, Depends, status
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.schemas.user import UserCreate, UserRead
from app.services.users import register_user
router = APIRouter(prefix="/users", tags=["users"])
@router.post(
"",
response_model=UserRead,
status_code=status.HTTP_201_CREATED,
)
def create_user(
payload: UserCreate,
db: Session = Depends(get_db),
):
return register_user(db, payload)
このように分けると、責務が明確になります。
18. リレーションの例
User が複数の Post を持つ例です。
app/models/post.py
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, String, Text, func
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)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
title: Mapped[str] = mapped_column(
String(200),
nullable=False,
)
body: Mapped[str] = mapped_column(
Text,
nullable=False,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
user: Mapped["User"] = relationship(
back_populates="posts",
)
app/models/user.py に追加
from sqlalchemy.orm import relationship
# Userクラス内に追加
posts: Mapped[list["Post"]] = relationship(
back_populates="user",
cascade="all, delete-orphan",
)
この場合、Alembic で再度マイグレーションを作ります。
alembic revision --autogenerate -m "create posts table"
alembic upgrade head
19. SQLAlchemy と Alembic の重要な注意点
1. モデル変更だけではDBは変わらない
SQLAlchemyモデルを書き換えても、DBは自動では変わりません。
モデルを変更
↓
alembic revision --autogenerate
↓
生成ファイルを確認
↓
alembic upgrade head
この流れが必要です。
2. autogenerate は必ず確認する
Alembic の --autogenerate は便利ですが、完璧ではありません。
公式ドキュメントでも、自動生成されたマイグレーション候補は人間が確認・必要に応じて修正してから使う流れが説明されています。(Alembic)
特に注意する変更はこれです。
- カラム名変更
- テーブル名変更
- enum変更
- 複雑な制約
- PostgreSQL拡張
- データ移行を伴う変更
例えば name を display_name に変更した場合、Alembic は「nameを削除してdisplay_nameを追加」と判断することがあります。
そのまま適用すると既存データが消える可能性があります。
この場合は手動でこう書くべきです。
def upgrade() -> None:
op.alter_column(
"users",
"name",
new_column_name="display_name",
)
def downgrade() -> None:
op.alter_column(
"users",
"display_name",
new_column_name="name",
)
3. 本番では create_all() を使わない
SQLAlchemyには以下のようにテーブルを作る機能があります。
Base.metadata.create_all(bind=engine)
ただし、これは開発初期やテスト用途向けです。
本番運用では基本的に、
alembic upgrade head
で管理するべきです。
理由は、変更履歴が残り、ステージング・本番・開発環境を同じ状態にできるからです。
20. FastAPIでのおすすめ構成
app/
db/
base.py
database.py
session.py
models/
__init__.py
user.py
post.py
schemas/
user.py
post.py
repositories/
users.py
posts.py
services/
users.py
posts.py
routers/
users.py
posts.py
main.py
alembic/
env.py
versions/
alembic.ini
app/models/__init__.py
Alembic がモデルを読み込めるように、ここで import しておくと便利です。
from app.models.user import User
from app.models.post import Post
__all__ = [
"User",
"Post",
]
alembic/env.py
from app.db.base import Base
import app.models # noqa: F401
target_metadata = Base.metadata
この形にしておくと、モデルが増えても Alembic 側の import を増やし続けなくて済みます。
21. 実務でのベストプラクティス
| 項目 | 推奨 |
|---|---|
| SQLAlchemy | 2.0形式で書く |
| モデル定義 |
Mapped + mapped_column
|
| マイグレーション | Alembic |
| DB反映 | alembic upgrade head |
| 本番 |
create_all() は使わない |
| セッション | FastAPI Dependencyで管理 |
| APIスキーマ | Pydanticと分離 |
| トランザクション | service層で境界を意識 |
| 自動生成 | 必ず中身をレビュー |
| DB URL | 環境変数から読む |
| 命名規則 | index / constraint 名を統一 |
| テスト | テストDBに対してAlembic適用 |
22. まとめ
SQLAlchemy + Alembic は、次のように覚えると分かりやすいです。
SQLAlchemy
PythonでDBを操作する道具
Alembic
DB構造の変更履歴を管理する道具
FastAPIでは、実務上この組み合わせがかなり自然です。
FastAPI
API入口
Pydantic
リクエスト / レスポンスの型定義
SQLAlchemy
DBモデル / クエリ / トランザクション
Alembic
テーブル作成 / カラム追加 / スキーマ変更管理
PostgreSQL
実データ保存
最初はこの3コマンドだけ覚えると十分です。
alembic revision --autogenerate -m "message"
alembic upgrade head
alembic downgrade -1
実務では、SQLAlchemyモデルを変更 → Alembicで差分作成 → 生成ファイルを確認 → DBへ適用、この流れを徹底すると安全に運用できます。