総合実習 — TODO API を完成させる

todo-api/
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── .env.example
├── .dockerignore
├── alembic.ini
├── alembic/
│   ├── env.py
│   └── versions/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── config.py
│   │   ├── security.py
│   │   └── logging.py
│   ├── api/
│   │   ├── deps.py
│   │   ├── auth.py
│   │   ├── todos.py
│   │   └── health.py
│   ├── models/
│   │   ├── base.py
│   │   ├── user.py
│   │   └── todo.py
│   ├── schemas/
│   │   ├── common.py
│   │   ├── user.py
│   │   └── todo.py
│   ├── services/
│   │   ├── user.py
│   │   └── todo.py
│   ├── db/
│   │   └── session.py
│   └── tasks/
│       └── notifier.py
└── tests/
    ├── conftest.py
    ├── test_auth.py
    └── test_todos.py

uv init todo-api --python 3.14
cd todo-api
uv add "fastapi[standard]" sqlalchemy "sqlalchemy[asyncio]" aiosqlite asyncpg
uv add argon2-cffi pyjwt pydantic-settings
uv add --dev pytest pytest-asyncio pytest-cov httpx pytest-httpx
uv add --dev pyright ruff alembic

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    app_name: str = "Todo API"
    database_url: str = "sqlite+aiosqlite:///./todo.db"
    jwt_secret: str = "dev-only-change-me"
    jwt_expire_minutes: int = 60
    notification_check_interval: int = 60   # バックグラウンド作業の間隔 (秒)

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

settings = Settings()

from fastapi import FastAPI
from app.core.config import settings
from app.api import auth, todos, health

app = FastAPI(title=settings.app_name, version="0.1.0")
app.include_router(auth.router)
app.include_router(todos.router)
app.include_router(health.router)

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base

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", cascade="all, delete-orphan")

from datetime import datetime, date
from sqlalchemy import String, Boolean, DateTime, Date, ForeignKey, Integer, JSON, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)

    title: Mapped[str] = mapped_column(String(200))
    description: Mapped[str] = mapped_column(default="")
    done: Mapped[bool] = mapped_column(Boolean, default=False, index=True)
    priority: Mapped[int] = mapped_column(Integer, default=3)
    due_date: Mapped[date | None] = mapped_column(Date, nullable=True, index=True)
    tags: Mapped[list[str]] = mapped_column(JSON, default=list)

    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()
    )

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

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)
AsyncSessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

from datetime import datetime, timedelta, timezone
import jwt
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
from app.core.config import settings

_hasher = PasswordHasher()
ALGORITHM = "HS256"

def hash_password(plain: str) -> str:
    return _hasher.hash(plain)

def verify_password(plain: str, hashed: str) -> bool:
    try:
        _hasher.verify(hashed, plain)
        return True
    except VerifyMismatchError:
        return False

def create_access_token(subject: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": subject,
        "iat": now,
        "exp": now + timedelta(minutes=settings.jwt_expire_minutes),
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=ALGORITHM)

def decode_token(token: str) -> dict:
    return jwt.decode(token, settings.jwt_secret, algorithms=[ALGORITHM])

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.core.security import hash_password, verify_password

async def create(db: AsyncSession, email: str, password: str) -> User:
    user = User(email=email, hashed_password=hash_password(password))
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_by_email(db: AsyncSession, email: str) -> User | None:
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()

async def authenticate(db: AsyncSession, email: str, password: str) -> User | None:
    user = await get_by_email(db, email)
    if user is None or not verify_password(password, user.hashed_password):
        return None
    return user

from typing import Annotated, AsyncIterator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt import InvalidTokenError
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import AsyncSessionLocal
from app.core.security import decode_token
from app.models.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

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

DBSession = Annotated[AsyncSession, Depends(get_session)]

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: DBSession,
) -> User:
    exc = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        "有効でないトークン",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_token(token)
        user_id = int(payload["sub"])
    except (InvalidTokenError, KeyError, ValueError):
        raise exc

    user = await db.get(User, user_id)
    if user is None:
        raise exc
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

# ページネーション依存性
from dataclasses import dataclass

@dataclass
class Pagination:
    skip: int = 0
    limit: int = 50

PaginationDep = Annotated[Pagination, Depends()]

from pydantic import BaseModel, EmailStr, Field, ConfigDict, SecretStr

class UserCreate(BaseModel):
    email: EmailStr
    password: SecretStr = Field(min_length=8, max_length=128)

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    email: EmailStr

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

from typing import Annotated
from fastapi import APIRouter, HTTPException, status, Depends
from fastapi.security import OAuth2PasswordRequestForm
from app.api.deps import DBSession, CurrentUser
from app.schemas.user import UserCreate, UserOut, Token
from app.services import user as user_service
from app.core.security import create_access_token

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

@router.post("/register", response_model=UserOut, status_code=201)
async def register(payload: UserCreate, db: DBSession) -> UserOut:
    existing = await user_service.get_by_email(db, payload.email)
    if existing:
        raise HTTPException(409, "既に登録済みのメール")
    return await user_service.create(db, payload.email, payload.password.get_secret_value())

@router.post("/login", response_model=Token)
async def login(
    form: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: DBSession,
) -> Token:
    user = await user_service.authenticate(db, form.username, form.password)
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Incorrect credentials")
    token = create_access_token(subject=str(user.id))
    return Token(access_token=token)

# /me ルートも一緒に
me_router = APIRouter(tags=["me"])

@me_router.get("/me", response_model=UserOut)
async def me(current: CurrentUser) -> UserOut:
    return current

from datetime import datetime, date
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Annotated

Priority = Annotated[int, Field(ge=1, le=5)]

class TodoBase(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True)

    title: str = Field(min_length=1, max_length=200)
    description: str = ""
    priority: Priority = 3
    due_date: date | None = None
    tags: list[str] = Field(default_factory=list)

    @field_validator("tags")
    @classmethod
    def normalize_tags(cls, v: list[str]) -> list[str]:
        # 重複除去 + 小文字化 + 空タグ除去
        return sorted({t.strip().lower() for t in v if t.strip()})

class TodoCreate(TodoBase):
    pass

class TodoUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    description: str | None = None
    done: bool | None = None
    priority: Priority | None = None
    due_date: date | None = None
    tags: list[str] | None = None

class TodoOut(TodoBase):
    model_config = ConfigDict(from_attributes=True)

    id: int
    done: bool
    created_at: datetime
    updated_at: datetime

from pydantic import BaseModel

class Page[T](BaseModel):
    items: list[T]
    total: int
    skip: int
    limit: int

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.todo import Todo
from app.schemas.todo import TodoCreate, TodoUpdate

async def create(db: AsyncSession, data: TodoCreate, owner_id: int) -> Todo:
    todo = Todo(**data.model_dump(), owner_id=owner_id)
    db.add(todo)
    await db.commit()
    await db.refresh(todo)
    return todo

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

async def list_filtered(
    db: AsyncSession,
    owner_id: int,
    *,
    skip: int = 0,
    limit: int = 50,
    done: bool | None = None,
    tag: str | None = None,
) -> tuple[list[Todo], int]:
    stmt = select(Todo).where(Todo.owner_id == owner_id)
    if done is not None:
        stmt = stmt.where(Todo.done == done)
    if tag is not None:
        # JSON カラムで tag が含まれる行 (DB 依存的 — SQLite/Postgres 両方サポート)
        stmt = stmt.where(Todo.tags.contains([tag]))

    total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar_one()
    items = list((await db.execute(
        stmt.order_by(Todo.created_at.desc()).offset(skip).limit(limit)
    )).scalars())
    return items, total

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

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

from fastapi import APIRouter, HTTPException, Query
from typing import Annotated
from app.api.deps import DBSession, CurrentUser, PaginationDep
from app.schemas.todo import TodoCreate, TodoUpdate, TodoOut
from app.schemas.common import Page
from app.services import todo as todo_service

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

@router.post("", response_model=TodoOut, status_code=201)
async def create_todo(payload: TodoCreate, user: CurrentUser, db: DBSession) -> TodoOut:
    return await todo_service.create(db, payload, owner_id=user.id)

@router.get("", response_model=Page[TodoOut])
async def list_todos(
    user: CurrentUser,
    db: DBSession,
    pagination: PaginationDep,
    done: Annotated[bool | None, Query()] = None,
    tag: Annotated[str | None, Query()] = None,
) -> Page[TodoOut]:
    items, total = await todo_service.list_filtered(
        db, owner_id=user.id,
        skip=pagination.skip, limit=pagination.limit,
        done=done, tag=tag,
    )
    return Page(items=items, total=total, skip=pagination.skip, limit=pagination.limit)

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

@router.patch("/{todo_id}", response_model=TodoOut)
async def update_todo(
    todo_id: int,
    payload: TodoUpdate,
    user: CurrentUser,
    db: DBSession,
) -> TodoOut:
    todo = await todo_service.get(db, todo_id, owner_id=user.id)
    if todo is None:
        raise HTTPException(404, "Todo not found")
    return await todo_service.update(db, todo, payload)

@router.delete("/{todo_id}", status_code=204)
async def delete_todo(todo_id: int, user: CurrentUser, db: DBSession) -> None:
    todo = await todo_service.get(db, todo_id, owner_id=user.id)
    if todo is None:
        raise HTTPException(404, "Todo not found")
    await todo_service.delete(db, todo)

from fastapi import APIRouter
from app.api.deps import DBSession
from sqlalchemy import text

router = APIRouter(tags=["ops"])

@router.get("/health")
async def health() -> dict[str, str]:
    return {"status": "healthy"}

@router.get("/health/db")
async def health_db(db: DBSession) -> dict[str, str]:
    try:
        await db.execute(text("SELECT 1"))
        return {"status": "ok", "db": "connected"}
    except Exception as e:
        return {"status": "error", "db": str(e)}

import asyncio
from datetime import date, timedelta
from sqlalchemy import select
from app.db.session import AsyncSessionLocal
from app.models.todo import Todo
from app.core.config import settings

async def check_due_todos() -> list[int]:
    """締め切りが今日または明日の未完了 TODO の id 一覧。"""
    tomorrow = date.today() + timedelta(days=1)
    async with AsyncSessionLocal() as db:
        result = await db.execute(
            select(Todo.id).where(
                Todo.done == False,
                Todo.due_date.between(date.today(), tomorrow),
            )
        )
        return list(result.scalars())

async def notifier_loop():
    """周期的に締め切り間近の TODO を確認して通知シミュレーション。"""
    while True:
        ids = await check_due_todos()
        for todo_id in ids:
            print(f"[notifier] 締め切り間近 TODO #{todo_id}")    # 実際はメール/Slack など
        await asyncio.sleep(settings.notification_check_interval)

from contextlib import asynccontextmanager
import asyncio
from fastapi import FastAPI
from app.tasks.notifier import notifier_loop
from app.api import auth, todos, health

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 起動時にバックグラウンド作業開始
    task = asyncio.create_task(notifier_loop())
    app.state.notifier_task = task
    yield
    # 終了時に整理
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

app = FastAPI(title="Todo API", lifespan=lifespan)
app.include_router(auth.router)
app.include_router(todos.router)
app.include_router(health.router)

import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from app.main import app
from app.api.deps import get_session, get_current_user
from app.models.base import Base
from app.models.user import User
from app.core.security import hash_password

TEST_URL = "sqlite+aiosqlite:///:memory:"

@pytest_asyncio.fixture
async def db_engine():
    engine = create_async_engine(TEST_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest_asyncio.fixture
async def db_session(db_engine):
    Session = async_sessionmaker(db_engine, expire_on_commit=False)
    async with Session() as session:
        yield session

@pytest_asyncio.fixture
async def test_user(db_session) -> User:
    user = User(email="test@example.com", hashed_password=hash_password("password123"))
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user

@pytest_asyncio.fixture
async def authed_client(db_session, test_user):
    async def override_session():
        yield db_session

    async def override_user():
        return test_user

    app.dependency_overrides[get_session] = override_session
    app.dependency_overrides[get_current_user] = override_user

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as c:
        yield c

    app.dependency_overrides.clear()

import pytest

pytestmark = pytest.mark.asyncio

async def test_create_and_get_todo(authed_client):
    create = await authed_client.post("/todos", json={"title": "テスト", "priority": 4})
    assert create.status_code == 201
    todo_id = create.json()["id"]

    get = await authed_client.get(f"/todos/{todo_id}")
    assert get.json()["title"] == "テスト"
    assert get.json()["priority"] == 4

async def test_filter_by_done(authed_client):
    await authed_client.post("/todos", json={"title": "open"})
    closed = await authed_client.post("/todos", json={"title": "closed"})
    await authed_client.patch(f"/todos/{closed.json()['id']}", json={"done": True})

    open_only = await authed_client.get("/todos?done=false")
    assert open_only.json()["total"] == 1
    assert open_only.json()["items"][0]["title"] == "open"

async def test_tag_normalization(authed_client):
    resp = await authed_client.post("/todos", json={
        "title": "with tags",
        "tags": ["Python", "  fastapi  ", "python", ""],
    })
    # 重複除去 + 小文字 + 空タグ除去
    assert sorted(resp.json()["tags"]) == ["fastapi", "python"]

async def test_404_on_other_user_todo(authed_client, db_session):
    # 別ユーザー + そのユーザーの TODO を直接挿入
    from app.models.user import User
    from app.models.todo import Todo
    other = User(email="other@x.com", hashed_password="x")
    db_session.add(other)
    await db_session.commit()
    foreign = Todo(title="not yours", owner_id=other.id)
    db_session.add(foreign)
    await db_session.commit()

    resp = await authed_client.get(f"/todos/{foreign.id}")
    assert resp.status_code == 404    # 別ユーザーの TODO は見えない

FROM python:3.14-slim AS builder
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev
COPY app/ ./app/
COPY alembic/ ./alembic/
COPY alembic.ini ./

FROM python:3.14-slim AS runtime
RUN useradd --create-home --uid 1000 appuser
WORKDIR /app
COPY --from=builder /app /app
ENV PATH="/app/.venv/bin:$PATH"
USER appuser
EXPOSE 8000

# マイグレーション適用後にサーバー開始
CMD ["sh", "-c", "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000"]

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql+asyncpg://todo:todo@db:5432/todo
      JWT_SECRET: dev-secret-only
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: todo
      POSTGRES_PASSWORD: todo
      POSTGRES_DB: todo
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U todo"]
      interval: 5s
      retries: 5
    volumes:
      - pg-data:/var/lib/postgresql/data

volumes:
  pg-data:

docker compose up -d
curl http://localhost:8000/health    # {"status":"healthy"}