L13 ch01 — SQLAlchemy で DB 連携

L13 ch01 — SQLAlchemy で DB 連携

メモリ list から本物の DB へ

ゴール

• SQLAlchemy で SQLite に接続
• モデルを Python クラスで定義
• DI (Depends) でセッション管理

なぜ SQLAlchemy?

• Python の デファクト ORM
• 複数 DB 対応 (SQLite / MySQL / PostgreSQL)
• 型安全 + IDE 補完

インストール

uv add sqlalchemy

接続と モデル定義

from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

DATABASE_URL = "sqlite:///./tasks.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class Base(DeclarativeBase):
    pass

class TaskModel(Base):
    __tablename__ = "tasks"
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String, nullable=False)
    done = Column(Boolean, default=False)

Base.metadata.create_all(bind=engine)

get_db (DI セッション管理)

from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

yield + finally で リクエスト終了時に確実に close。

CRUD エンドポイント

@app.get("/tasks")
def list_tasks(db: Session = Depends(get_db)):
    return db.query(TaskModel).all()

@app.post("/tasks", status_code=201)
def create(task: TaskIn, db: Session = Depends(get_db)):
    db_task = TaskModel(title=task.title, done=task.done)
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task

トランザクション

@app.patch("/tasks/{id}")
def update(id: int, db: Session = Depends(get_db)):
    try:
        # 複数の db.add / db.delete
        db.commit()
    except Exception:
        db.rollback()
        raise

drill

drill/ — fill-session, add-delete