はじめに
Python の ORM といえば SQLAlchemy。
でも実際は「覚える範囲、広すぎ問題」があって、最初は絶対に迷う。
この記事では、最低限＋実務レベルで絶対に使う部分だけ、未来の自分が読み返しても迷わない形で整理する。
SQLAlchemy は何ができる？
SQLAlchemy は「ORM＋SQL ツールキット」の二刀流フレームワーク。
- ORM（Python クラスで DB を扱う）
- Core（生 SQL を Python から安全に書く）
FastAPI や Flask と組み合わせる場面では ほぼ ORM のみ で十分。
実務では ORM と Core をハイブリッドで使う場面もあるが、まずは ORM を理解すればOK。
基本構造：ORM の 4 コンポーネント（最重要）
SQLAlchemy ORM の最小構成は以下の 4 つ。
- Base（ORM の親クラス）
- Model（テーブル）
- Engine（DB 接続）
- Session（DB 操作用オブジェクト）
これを順番に見ていく。
Base（ORM の土台）
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
すべてのモデルはこの Base を継承する。
Django で言う models.Model 的な存在。
Model（テーブル定義）
新しい SQLAlchemy の書き方「Mapped + mapped_column」を使うときれいになる。
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String, Integer
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(String(200), unique=True)
ポイント：
- Python の型ヒント＝そのまま DB 型推論に使える
-
Mapped[]で SQLAlchemy に「これは ORM カラムですよ」と教える - 実務ではほぼこの書き方一択
Engine（DB との接続管理）
from sqlalchemy import create_engine
engine = create_engine(
"sqlite:///./app.db",
echo=True, # 実行された SQL をログに出す
)
実務では postgresql+psycopg:// をよく使う。
Session（DB の操作をする実体）
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(
bind=engine,
autocommit=False,
autoflush=False,
)
使うときは：
db = SessionLocal()
user = db.query(User).first()
print(user)
Session は「短命」で使う（FastAPI は1リクエスト1セッション）。
CRUD（実務の 90% はこれ）
Create
user = User(name="Test", email="test@example.com")
db.add(user)
db.commit()
db.refresh(user)
refresh() は INSERT 後に生成された id などを取得。
Read
user = db.query(User).filter(User.id == 1).first()
複数：
users = db.query(User).all()
Update
user = db.query(User).filter(User.id == 1).first()
user.name = "NewName"
db.commit()
Delete
db.delete(user)
db.commit()
リレーション（1対多・多対多）
これも実務必須。
User ←→ Post（1対多）
from sqlalchemy.orm import relationship
from sqlalchemy import ForeignKey
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
posts: Mapped[list["Post"]] = relationship(back_populates="user")
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
user: Mapped[User] = relationship(back_populates="posts")
SQLAlchemy のリレーションは「両方向」で書くと扱いやすい。
async SQLAlchemy（FastAPI の新標準）
FastAPI + SQLAlchemy は async が今の主流。
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
engine = create_async_engine("sqlite+aiosqlite:///./test.db")
async_session = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
使用：
async with async_session() as session:
result = await session.execute(select(User))
users = result.scalars().all()
FastAPI との組み合わせ
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
エンドポイント：
@app.get("/users")
def list_users(db: Session = Depends(get_db)):
return db.query(User).all()
Clean Architecture での配置（実務レベル）
SQLAlchemy を Clean Architecture で使うなら：
/domain
models/
repositories/
/data
sqlalchemy/
models/
repositories/
session.py
ポイント：
- domain は SQLAlchemy を知らない（import しない）
- repositories が SQLAlchemy 実装を隠蔽する
- data 層に Model / Session を閉じ込める
例：UserRepository
class UserRepositoryImpl(UserRepository):
def __init__(self, session):
self.session = session
def find_by_id(self, user_id: int):
return (
self.session
.query(UserModel)
.filter(UserModel.id == user_id)
.first()
)
実務の “正しい SQLAlchemy の使い方” はまさにこれ。
Migration（Alembic）
SQLAlchemy を使うなら Alembic も自動的にセット。
初期化：
alembic init migrations
model → migration：
alembic revision --autogenerate -m "create user"
alembic upgrade head
実務では必須級。
はじめに
- Model 定義
- CRUD
- Relation
- Session の扱い
- async
- Migration（Alembic）
- Clean Architecture 組み込み
この順番が絶対に効率いい。