팀 프로젝트 폴더 구조 — 단일 파일에서 도메인형까지 (FastAPI 입문 7편)

from pydantic import BaseModel, EmailStr, Field


class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(min_length=1, max_length=50)


class User(BaseModel):
    id: int
    email: EmailStr
    name: str

from app.users.schemas import User, UserCreate


class UserRepository:
    def __init__(self) -> None:
        self._rows: dict[int, User] = {}
        self._next_id = 1

    def add(self, data: UserCreate) -> User:
        user = User(id=self._next_id, **data.model_dump())
        self._rows[user.id] = user
        self._next_id += 1
        return user

    def get(self, user_id: int) -> User | None:
        return self._rows.get(user_id)

    def find_by_email(self, email: str) -> User | None:
        return next((u for u in self._rows.values() if u.email == email), None)


user_repo = UserRepository()

from fastapi import HTTPException, status

from app.users.repository import UserRepository
from app.users.schemas import User, UserCreate


def create_user(repo: UserRepository, data: UserCreate) -> User:
    if repo.find_by_email(data.email):
        raise HTTPException(status.HTTP_409_CONFLICT, detail="이미 가입된 이메일입니다")
    return repo.add(data)


def get_user_or_404(repo: UserRepository, user_id: int) -> User:
    user = repo.get(user_id)
    if user is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"{user_id}번 사용자가 없습니다")
    return user

from typing import Annotated

from fastapi import Depends

from app.users.repository import UserRepository, user_repo


def get_user_repo() -> UserRepository:
    return user_repo


UserRepoDep = Annotated[UserRepository, Depends(get_user_repo)]

from fastapi import APIRouter, status

from app.users import service
from app.users.dependencies import UserRepoDep
from app.users.schemas import User, UserCreate

router = APIRouter(prefix="/users", tags=["사용자"])


@router.post("", response_model=User, status_code=status.HTTP_201_CREATED, summary="사용자 만들기")
def create_user(payload: UserCreate, repo: UserRepoDep):
    return service.create_user(repo, payload)


@router.get("/{user_id}", response_model=User, summary="사용자 하나 보기")
def read_user(user_id: int, repo: UserRepoDep):
    return service.get_user_or_404(repo, user_id)

from typing import Annotated

from fastapi import Depends, HTTPException, status

from app.tasks.repository import TaskRepository, task_repo
from app.tasks.schemas import Task


def get_task_repo() -> TaskRepository:
    return task_repo


TaskRepoDep = Annotated[TaskRepository, Depends(get_task_repo)]


def get_task_or_404(task_id: int, repo: TaskRepoDep) -> Task:
    task = repo.get(task_id)
    if task is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"{task_id}번 할 일이 없습니다")
    return task


TaskDep = Annotated[Task, Depends(get_task_or_404)]

from typing import Annotated

from fastapi import Depends, Query


class Pagination:
    def __init__(
        self,
        skip: Annotated[int, Query(ge=0)] = 0,
        limit: Annotated[int, Query(ge=1, le=100)] = 20,
    ) -> None:
        self.skip = skip
        self.limit = limit


PageDep = Annotated[Pagination, Depends()]

from datetime import datetime

from pydantic import BaseModel, Field, field_validator


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100, examples=["보고서 초안 쓰기"])
    priority: int = Field(default=3, ge=1, le=5)
    owner_id: int


class TaskPatch(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=100)
    priority: int | None = Field(default=None, ge=1, le=5)
    done: bool | None = None

    @field_validator("title", "priority", "done")
    @classmethod
    def reject_null(cls, value):
        if value is None:
            raise ValueError("이 필드는 null로 지울 수 없습니다")
        return value


class Task(BaseModel):
    id: int
    title: str
    priority: int
    done: bool
    owner_id: int
    created_at: datetime

from datetime import UTC, datetime

from app.tasks.schemas import Task, TaskCreate


class TaskRepository:
    """지금은 메모리 딕셔너리. DB로 바꿀 때 이 파일만 고친다."""

    def __init__(self) -> None:
        self._rows: dict[int, Task] = {}
        self._next_id = 1

    def add(self, data: TaskCreate) -> Task:
        task = Task(id=self._next_id, done=False, created_at=datetime.now(UTC), **data.model_dump())
        self._rows[task.id] = task
        self._next_id += 1
        return task

    def list(self, done: bool | None = None) -> list[Task]:
        rows = list(self._rows.values())
        return rows if done is None else [t for t in rows if t.done == done]

    def get(self, task_id: int) -> Task | None:
        return self._rows.get(task_id)

    def save(self, task: Task) -> Task:
        self._rows[task.id] = task
        return task

    def remove(self, task_id: int) -> None:
        self._rows.pop(task_id, None)


task_repo = TaskRepository()

from app.tasks.repository import TaskRepository
from app.tasks.schemas import Task, TaskCreate, TaskPatch
from app.users import service as user_service
from app.users.repository import UserRepository


def create_task(tasks: TaskRepository, users: UserRepository, data: TaskCreate) -> Task:
    user_service.get_user_or_404(users, data.owner_id)  # 다른 도메인은 service를 통해서만
    return tasks.add(data)


def update_task(tasks: TaskRepository, task: Task, data: TaskPatch) -> Task:
    changes = data.model_dump(exclude_unset=True)
    return tasks.save(task.model_copy(update=changes))

from fastapi import APIRouter, status

from app.common.pagination import PageDep
from app.tasks import service
from app.tasks.dependencies import TaskDep, TaskRepoDep
from app.tasks.schemas import Task, TaskCreate, TaskPatch
from app.users.dependencies import UserRepoDep

router = APIRouter(prefix="/tasks", tags=["할 일"])


@router.post("", response_model=Task, status_code=status.HTTP_201_CREATED, summary="할 일 만들기")
def create_task(payload: TaskCreate, tasks: TaskRepoDep, users: UserRepoDep):
    return service.create_task(tasks, users, payload)


@router.get("", response_model=list[Task], summary="할 일 목록")
def list_tasks(tasks: TaskRepoDep, page: PageDep, done: bool | None = None):
    return tasks.list(done)[page.skip : page.skip + page.limit]


@router.get("/{task_id}", response_model=Task, summary="할 일 하나 보기")
def read_task(task: TaskDep):
    return task


@router.patch("/{task_id}", response_model=Task, summary="할 일 일부 수정")
def update_task(task: TaskDep, payload: TaskPatch, tasks: TaskRepoDep):
    return service.update_task(tasks, task, payload)


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT, summary="할 일 삭제")
def delete_task(task: TaskDep, tasks: TaskRepoDep):
    tasks.remove(task.id)

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")

    app_env: str = "local"
    cors_origins: list[str] = ["http://localhost:3000"]

    @property
    def is_prod(self) -> bool:
        return self.app_env == "production"


settings = Settings()

from fastapi import APIRouter, FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.core.config import settings
from app.tasks.router import router as tasks_router
from app.users.router import router as users_router

app = FastAPI(
    title="할 일 API",
    version="1.0.0",
    docs_url=None if settings.is_prod else "/docs",
    redoc_url=None if settings.is_prod else "/redoc",
    openapi_url=None if settings.is_prod else "/openapi.json",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_methods=["*"],
    allow_headers=["*"],
)

api_v1 = APIRouter(prefix="/api/v1")
api_v1.include_router(users_router)
api_v1.include_router(tasks_router)
app.include_router(api_v1)


@app.get("/health", tags=["운영"], summary="서버 상태 확인")
def health():
    return {"status": "ok"}

uv run fastapi dev

import pytest
from fastapi.testclient import TestClient

from app.main import app
from app.tasks.dependencies import get_task_repo
from app.tasks.repository import TaskRepository
from app.users.dependencies import get_user_repo
from app.users.repository import UserRepository


@pytest.fixture
def client():
    tasks, users = TaskRepository(), UserRepository()
    app.dependency_overrides[get_task_repo] = lambda: tasks
    app.dependency_overrides[get_user_repo] = lambda: users
    yield TestClient(app)
    app.dependency_overrides.clear()


@pytest.fixture
def user_id(client):
    return client.post("/api/v1/users", json={"email": "kim@example.com", "name": "김철수"}).json()["id"]

app.dependency_overrides[get_task_repo] = lambda: tasks

def test_create_user(client):
    res = client.post("/api/v1/users", json={"email": "lee@example.com", "name": "이영희"})
    assert res.status_code == 201


def test_duplicate_email_is_409(client, user_id):
    res = client.post("/api/v1/users", json={"email": "kim@example.com", "name": "또철수"})
    assert res.status_code == 409


def test_invalid_email_is_422(client):
    res = client.post("/api/v1/users", json={"email": "not-an-email", "name": "누구"})
    assert res.status_code == 422

def test_create_and_read(client, user_id):
    res = client.post("/api/v1/tasks", json={"title": "보고서 쓰기", "owner_id": user_id})
    assert res.status_code == 201
    task_id = res.json()["id"]
    assert client.get(f"/api/v1/tasks/{task_id}").json()["title"] == "보고서 쓰기"


def test_unknown_owner_is_404(client):
    res = client.post("/api/v1/tasks", json={"title": "주인 없는 일", "owner_id": 999})
    assert res.status_code == 404


def test_patch_keeps_other_fields(client, user_id):
    task_id = client.post("/api/v1/tasks", json={"title": "회의록", "priority": 2, "owner_id": user_id}).json()["id"]
    res = client.patch(f"/api/v1/tasks/{task_id}", json={"done": True})
    assert res.json()["done"] is True and res.json()["priority"] == 2


def test_patch_null_title_is_422(client, user_id):
    task_id = client.post("/api/v1/tasks", json={"title": "제목", "owner_id": user_id}).json()["id"]
    assert client.patch(f"/api/v1/tasks/{task_id}", json={"title": None}).status_code == 422


def test_delete_then_404(client, user_id):
    task_id = client.post("/api/v1/tasks", json={"title": "지울 일", "owner_id": user_id}).json()["id"]
    assert client.delete(f"/api/v1/tasks/{task_id}").status_code == 204
    assert client.get(f"/api/v1/tasks/{task_id}").status_code == 404

[project]
name = "todo-api"
version = "0.1.0"
requires-python = ">=3.13"
dependencies = [
    "fastapi[standard]>=0.141.1",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
    "ruff>=0.16.9",
]

[tool.ruff]
line-length = 120

[tool.pytest.ini_options]
testpaths = ["tests"]