종합 실습 — 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"}