DB 連携 — SQLAlchemy 2.x + Alembic

uv add sqlalchemy "sqlalchemy[asyncio]" asyncpg
uv add --dev alembic

uv add aiosqlite

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine, AsyncSession
from app.core.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=False,           # 開発中は True にして SQL ログ
    pool_pre_ping=True,   # 切れた接続を自動回復
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,   # コミット後もオブジェクト使用可能
)

from typing import Annotated, AsyncIterator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import AsyncSessionLocal

async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        yield session

DBSession = Annotated[AsyncSession, Depends(get_session)]

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

from datetime import datetime
from sqlalchemy import String, Boolean, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    description: Mapped[str | None]
    done: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
    )

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(200), unique=True, index=True)
    hashed_password: Mapped[str]

    todos: Mapped[list["Todo"]] = relationship(back_populates="owner")

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    owner: Mapped["User"] = relationship(back_populates="todos")

from sqlalchemy.ext.asyncio import AsyncSession

async def create_todo(db: AsyncSession, data: TodoCreate, owner_id: int) -> Todo:
    todo = Todo(
        title=data.title,
        description=data.description,
        owner_id=owner_id,
    )
    db.add(todo)
    await db.commit()
    await db.refresh(todo)    # DB が埋めた id、created_at などを再ロード
    return todo

from sqlalchemy import select

async def get_todo(db: AsyncSession, todo_id: int) -> Todo | None:
    result = await db.execute(
        select(Todo).where(Todo.id == todo_id)
    )
    return result.scalar_one_or_none()

async def list_todos(db: AsyncSession, skip: int, limit: int) -> list[Todo]:
    result = await db.execute(
        select(Todo).offset(skip).limit(limit).order_by(Todo.created_at.desc())
    )
    return list(result.scalars())

async def update_todo(db: AsyncSession, todo: Todo, data: TodoUpdate) -> Todo:
    update_dict = data.model_dump(exclude_unset=True)
    for field, value in update_dict.items():
        setattr(todo, field, value)
    await db.commit()
    await db.refresh(todo)
    return todo

from sqlalchemy import update

async def mark_all_done(db: AsyncSession, owner_id: int) -> int:
    result = await db.execute(
        update(Todo)
        .where(Todo.owner_id == owner_id)
        .values(done=True)
    )
    await db.commit()
    return result.rowcount

async def delete_todo(db: AsyncSession, todo: Todo) -> None:
    await db.delete(todo)
    await db.commit()

users = (await db.execute(select(User))).scalars()
for user in users:
    todos = user.todos    # ← 毎回追加クエリ

from sqlalchemy.orm import selectinload

result = await db.execute(
    select(User).options(selectinload(User.todos))
)
users = result.scalars()
for user in users:
    print(user.todos)    # 追加クエリなし

from fastapi import APIRouter, HTTPException
from app.api.deps import DBSession
from app.schemas.todo import TodoCreate, TodoOut, TodoUpdate
from app.models.todo import Todo
from sqlalchemy import select

router = APIRouter(prefix="/todos", tags=["todos"])

@router.post("/", response_model=TodoOut, status_code=201)
async def create_todo(payload: TodoCreate, db: DBSession) -> Todo:
    todo = Todo(**payload.model_dump(), owner_id=1)   # owner_id は第26章で認証
    db.add(todo)
    await db.commit()
    await db.refresh(todo)
    return todo

@router.get("/{todo_id}", response_model=TodoOut)
async def read_todo(todo_id: int, db: DBSession) -> Todo:
    todo = await db.get(Todo, todo_id)
    if todo is None:
        raise HTTPException(404, "Todo not found")
    return todo

uv run alembic init -t async alembic

sqlalchemy.url = postgresql+asyncpg://user:pw@localhost/todo

from app.core.config import settings
from app.models.base import Base
from app.models import todo, user   # すべてのモデルを import

config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)

target_metadata = Base.metadata

uv run alembic revision --autogenerate -m "create todos table"

def upgrade() -> None:
    op.create_table(
        "todos",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("title", sa.String(200)),
        ...
    )

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

uv run alembic upgrade head    # 最新まで
uv run alembic downgrade -1    # 一段戻す
uv run alembic current         # 現在のバージョン
uv run alembic history         # すべてのマイグレーション

uv run alembic merge -m "merge a and b" abc123 def456

result = await db.execute(select(Todo))
rows = result.all()    # 各 row が (Todo,) タプル

result = await db.execute(select(Todo))
todos = list(result.scalars())    # Todo オブジェクトのリスト

async def get_session() -> AsyncIterator[AsyncSession]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise