프로젝트 구조와 APIRouter
1. 파일 하나로는 부족해지는 순간
5장에서 만든 블로그의 main.py는 200줄이 넘습니다. 설정, 모델, 스키마, 인증, 엔드포인트가 한 파일에 모여 있습니다.
지금까지는 이 방식이 좋았습니다. 파일을 오갈 필요가 없어 배우기에 편했기 때문입니다. 그런데 여기에 댓글, 좋아요, 태그, 알림 기능을 더하면 어떻게 될까요. 1000줄이 넘어가면 아래와 같은 일이 생깁니다.
- 고칠 부분을 찾는 데 시간이 걸립니다.
- 두 사람이 동시에 작업하면 Git 충돌이 계속 납니다.
- 인증 관련 코드가 어디까지인지 경계가 흐려집니다.
- 테스트할 때 파일 전체를 가져와야 합니다.
이번 절에서는 5장 프로젝트를 여러 파일로 나눠보겠습니다. 기능은 그대로이고 배치만 바뀝니다.
언제 나눠야 하나요
정해진 기준은 없지만, 아래 중 하나에 해당하면 나눌 때가 된 것입니다.
- 파일이 300줄을 넘어갑니다.
- 엔드포인트가 10개를 넘어갑니다.
- 두 명 이상이 같은 파일을 고칩니다.
- "이 함수가 어디 있더라" 하고 검색하기 시작합니다.
반대로 엔드포인트가 세 개인 프로젝트를 여덟 개 파일로 나누는 것은 오히려 읽기 어렵게 만듭니다.
2. 목표 구조
05_blog의 가상환경에서 6-3절의 방법으로 requirements.txt를 저장하세요. 새 06_5_structure 폴더에는 이 파일과 static 폴더를 복사하고, venv는 복사하지 않습니다. 새 폴더에서 가상환경을 생성·활성화한 뒤 pip install -r requirements.txt로 설치합니다. 기존 서버는 먼저 멈추고 기존 가상환경에서는 deactivate로 빠져나오세요. 7장도 이 프로젝트를 이어서 사용합니다.
06_5_structure
┣━ 📄requirements.txt
┣━ 📁static/ # 5장에서 만든 화면 파일들
┗━ 📁app/
┣━ 📄__init__.py
┣━ 📄main.py # 앱을 만들고 라우터를 연결
┣━ 📄config.py # 설정값
┣━ 📄database.py # 엔진, 세션, Base
┣━ 📄models.py # SQLAlchemy 모델
┣━ 📄schemas.py # Pydantic 스키마
┣━ 📄security.py # 비밀번호 해싱, 토큰 생성
┣━ 📄dependencies.py # get_db, get_current_user
┗━ 📁routers/
┣━ 📄__init__.py
┣━ 📄auth.py # 회원가입, 로그인, 내 정보
┗━ 📄blogs.py # 블로그 CRUD
__init__.py는 빈 파일입니다. 이 파일이 있어야 파이썬이 그 폴더를 패키지로 인식합니다.
폴더를 나눌 때 흔히 두 가지 방식을 씁니다.
| 방식 | 나누는 기준 | 예 |
|---|---|---|
| 계층별 | 코드의 역할 | models/, schemas/, routers/ |
| 기능별 | 도메인 | blog/, user/, comment/ |
작은 프로젝트에서는 계층별이 익히기 쉽고, 커지면 기능별이 관리하기 좋습니다. 이 절에서는 계층별로 나눕니다.
3. 파일별 코드
5장 05-6절에서 만든 코드를 그대로 옮깁니다. 내용은 거의 바뀌지 않고 위치만 달라집니다.
3.1 app/config.py
설정값을 한곳에 모읍니다.
SQLALCHEMY_DATABASE_URL = "sqlite:///./blogs.db"
# 실제 서비스에서는 환경 변수로 관리해야 합니다(7장에서 다룹니다).
SECRET_KEY = "9f2c8e1b47a5d3f6089b2e7c4a1d5f83b6e0c9a2d7f4b1e8c3a6d9f2b5e8c1a4"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
3.2 app/database.py
from collections.abc import Generator
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from app.config import SQLALCHEMY_DATABASE_URL
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
)
SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase):
pass
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
3.3 app/models.py
from datetime import date
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class UserModel(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, index=True)
email: Mapped[str] = mapped_column(unique=True, index=True)
hashed_password: Mapped[str]
created_at: Mapped[date]
blogs: Mapped[list["BlogModel"]] = relationship(back_populates="author")
class BlogModel(Base):
__tablename__ = "blogs"
id: Mapped[int] = mapped_column(primary_key=True, index=True)
title: Mapped[str] = mapped_column(index=True)
content: Mapped[str]
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
created_at: Mapped[date]
updated_at: Mapped[date]
author: Mapped["UserModel"] = relationship(back_populates="blogs")
@property
def author_email(self) -> str:
return self.author.email
3.4 app/schemas.py
from datetime import date
from pydantic import BaseModel, ConfigDict, EmailStr, Field
class UserCreate(BaseModel):
email: EmailStr
password: str = Field(min_length=8)
class UserPublic(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
email: EmailStr
created_at: date
class Token(BaseModel):
access_token: str
token_type: str = "bearer"
class BlogCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
content: str = Field(min_length=1)
class BlogUpdate(BlogCreate):
pass
class Blog(BlogCreate):
model_config = ConfigDict(from_attributes=True)
id: int
author_id: int
author_email: EmailStr
created_at: date
updated_at: date
3.5 app/security.py
비밀번호와 토큰을 다루는 코드만 모읍니다.
from datetime import datetime, timedelta, timezone
import jwt
from pwdlib import PasswordHash
from app.config import ACCESS_TOKEN_EXPIRE_MINUTES, ALGORITHM, SECRET_KEY
password_hash = PasswordHash.recommended()
def hash_password(password: str) -> str:
return password_hash.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return password_hash.verify(plain_password, hashed_password)
def create_access_token(user_id: int) -> str:
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
payload = {"sub": str(user_id), "exp": expire}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def decode_access_token(token: str) -> dict | None:
try:
return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except jwt.InvalidTokenError:
return None
decode_access_token이 예외를 던지지 않고 None을 반환하도록 바꾼 점에 주목하세요. 이 파일은 HTTP를 모르는 파일입니다. HTTPException을 던지는 것은 HTTP를 아는 계층인 dependencies.py의 일입니다.
이렇게 나누면 나중에 이 함수를 웹이 아닌 곳, 예를 들어 관리자 스크립트나 배치 작업에서도 그대로 쓸 수 있습니다.
3.6 app/dependencies.py
여러 라우터에서 함께 쓰는 의존성을 모읍니다.
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import UserModel
from app.security import decode_access_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SessionDep = Annotated[Session, Depends(get_db)]
def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: SessionDep,
) -> UserModel:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="자격 증명을 확인할 수 없습니다",
headers={"WWW-Authenticate": "Bearer"},
)
payload = decode_access_token(token)
if payload is None:
raise credentials_exception
user_id = payload.get("sub")
if user_id is None:
raise credentials_exception
user = db.get(UserModel, int(user_id))
if user is None:
raise credentials_exception
return user
CurrentUser = Annotated[UserModel, Depends(get_current_user)]
3.7 app/routers/auth.py
여기서 APIRouter가 처음 나옵니다.
from datetime import date
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from app.dependencies import CurrentUser, SessionDep
from app.models import UserModel
from app.schemas import Token, UserCreate, UserPublic
from app.security import create_access_token, hash_password, verify_password
router = APIRouter(tags=["인증"])
@router.post("/signup", status_code=status.HTTP_201_CREATED)
def signup(user_data: UserCreate, db: SessionDep) -> UserPublic:
"""회원가입을 합니다."""
existing = db.scalar(select(UserModel).where(UserModel.email == user_data.email))
if existing is not None:
raise HTTPException(status_code=400, detail="이미 가입된 이메일입니다")
user = UserModel(
email=user_data.email,
hashed_password=hash_password(user_data.password),
created_at=date.today(),
)
db.add(user)
db.commit()
db.refresh(user)
return user
@router.post("/token")
def login(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
db: SessionDep,
) -> Token:
"""이메일과 비밀번호를 확인하고 액세스 토큰을 발급합니다."""
user = db.scalar(select(UserModel).where(UserModel.email == form_data.username))
if user is None or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="이메일 또는 비밀번호가 올바르지 않습니다",
headers={"WWW-Authenticate": "Bearer"},
)
return Token(access_token=create_access_token(user.id))
@router.get("/me")
def read_me(current_user: CurrentUser) -> UserPublic:
"""지금 로그인한 사용자를 반환합니다."""
return current_user
app = FastAPI()가 router = APIRouter()로 바뀌었고, @app.post가 @router.post로 바뀌었습니다. 그 외에는 5장 코드 그대로입니다.
APIRouter는 엔드포인트를 담아두는 상자라고 생각하시면 됩니다. 나중에 main.py에서 이 상자를 앱에 끼워 넣습니다.
APIRouter(tags=["인증"])처럼 라우터 단위로 태그를 지정하면 각 엔드포인트마다 반복해서 적지 않아도 됩니다.
3.8 app/routers/blogs.py
from datetime import date
from fastapi import APIRouter, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.dependencies import CurrentUser, SessionDep
from app.models import BlogModel, UserModel
from app.schemas import Blog, BlogCreate, BlogUpdate
router = APIRouter(prefix="/blogs", tags=["블로그"])
def get_blog_or_404(db: Session, blog_id: int) -> BlogModel:
blog = db.get(BlogModel, blog_id)
if blog is None:
raise HTTPException(status_code=404, detail="Blog not found")
return blog
def check_owner(blog: BlogModel, user: UserModel) -> None:
if blog.author_id != user.id:
raise HTTPException(
status_code=403, detail="본인이 작성한 글만 수정하거나 삭제할 수 있습니다"
)
@router.get("")
def read_blogs(db: SessionDep) -> list[Blog]:
"""모든 글을 최신순으로 반환합니다."""
return list(db.scalars(select(BlogModel).order_by(BlogModel.id.desc())).all())
@router.get("/{blog_id}")
def read_blog(blog_id: int, db: SessionDep) -> Blog:
"""글 하나를 반환합니다."""
return get_blog_or_404(db, blog_id)
@router.post("", status_code=status.HTTP_201_CREATED)
def create_blog(blog_data: BlogCreate, current_user: CurrentUser, db: SessionDep) -> Blog:
"""새 글을 작성합니다. 로그인이 필요합니다."""
today = date.today()
blog = BlogModel(
title=blog_data.title,
content=blog_data.content,
author_id=current_user.id,
created_at=today,
updated_at=today,
)
db.add(blog)
db.commit()
db.refresh(blog)
return blog
@router.put("/{blog_id}")
def update_blog(
blog_id: int,
blog_data: BlogUpdate,
current_user: CurrentUser,
db: SessionDep,
) -> Blog:
"""글을 수정합니다. 본인이 작성한 글만 가능합니다."""
blog = get_blog_or_404(db, blog_id)
check_owner(blog, current_user)
blog.title = blog_data.title
blog.content = blog_data.content
blog.updated_at = date.today()
db.commit()
db.refresh(blog)
return blog
@router.delete("/{blog_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_blog(blog_id: int, current_user: CurrentUser, db: SessionDep) -> None:
"""글을 삭제합니다. 본인이 작성한 글만 가능합니다."""
blog = get_blog_or_404(db, blog_id)
check_owner(blog, current_user)
db.delete(blog)
db.commit()
APIRouter(prefix="/blogs")가 이 파일의 핵심입니다. 모든 경로 앞에 /blogs가 자동으로 붙으므로, 데코레이터에는 그 뒤만 적으면 됩니다.
| 데코레이터 | 실제 경로 |
|---|---|
@router.get("") | /blogs |
@router.get("/{blog_id}") | /blogs/{blog_id} |
@router.post("") | /blogs |
경로가 바뀌어야 할 때 prefix 한 곳만 고치면 됩니다.
3.9 app/main.py
마지막으로 모든 것을 조립합니다.
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.database import Base, engine
from app.routers import auth, blogs
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(bind=engine)
yield
app = FastAPI(title="위니브 블로그 API", version="1.0.0", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth.router)
app.include_router(blogs.router)
app.frontend("/", directory="static")
main.py가 30줄로 줄었습니다. 이 파일만 보면 이 서비스가 무엇으로 구성되어 있는지 한눈에 들어옵니다.
include_router가 라우터를 앱에 끼워 넣는 함수입니다. 여기서도 접두사와 태그를 지정할 수 있습니다.
# API 전체에 /api/v1을 붙이고 싶다면
app.include_router(auth.router, prefix="/api/v1")
app.include_router(blogs.router, prefix="/api/v1")
4. 실행하기
main.py가 app 폴더 안으로 들어갔으므로 실행 명령이 조금 달라집니다.
fastapi dev app/main.py
fastapi dev는 인자 없이 실행하면 현재 폴더의 main.py를 찾습니다. 위치가 바뀌었으니 경로를 알려줘야 합니다.
http://127.0.0.1:8000/docs에 접속해보면 5장과 완전히 같은 문서가 나옵니다. 태그로 인증과 블로그가 나뉘어 있는 것도 확인할 수 있습니다.
앞서 만들어둔 .http 파일도 그대로 동작합니다. 경로가 하나도 바뀌지 않았기 때문입니다.
ModuleNotFoundError: No module named 'app'이 나온다면
app/main.py를 실행하는데 from app.database import ...를 찾지 못하는 경우입니다. 명령을 실행하는 위치가 app 폴더 안이 아니라 프로젝트 최상위 폴더여야 합니다.
# 맞는 위치
06_5_structure> fastapi dev app/main.py
# 틀린 위치
06_5_structure/app> fastapi dev main.py
또한 app/__init__.py와 app/routers/__init__.py가 있는지 확인해보세요. 빈 파일이어도 있어야 합니다.
5. 순환 참조 피하기
파일을 나누면 새로 만나는 문제가 있습니다. A가 B를 가져오고 B가 A를 가져오면 파이썬이 에러를 냅니다.
ImportError: cannot import name 'X' from partially initialized module
이 구조에서 파일 사이의 의존 방향은 아래와 같습니다.
화살표가 한 방향으로만 흐릅니다. 아래쪽 파일은 위쪽 파일을 절대 가져오지 않습니다. models.py가 routers/blogs.py를 가져오는 일은 없어야 합니다.
이 규칙을 지키면 순환 참조가 생기지 않습니다. 나눌 때 "이 파일은 누구를 알아야 하나"를 먼저 정하는 것이 좋습니다.
6. 나눈 뒤의 차이
| 파일 하나 | 나눈 뒤 | |
|---|---|---|
| 블로그 기능 수정 | 200줄에서 찾기 | routers/blogs.py만 열기 |
| 두 명이 동시 작업 | 같은 파일 충돌 | 서로 다른 파일 |
| 토큰 로직 재사용 | 어려움 | security.py를 가져다 씀 |
| 테스트 | 앱 전체를 가져와야 함 | 필요한 것만 가져옴 |
| 기능 추가 | 파일이 계속 길어짐 | 라우터 파일 하나 추가 |
댓글 기능을 추가한다고 하면, routers/comments.py를 만들고 main.py에 한 줄을 더하면 됩니다.
from app.routers import auth, blogs, comments
app.include_router(comments.router)
연습문제
- 5장에서 만든 블로그를 이 구조로 직접 옮겨보세요.
.http파일이 그대로 동작하는지 확인해보세요. routers/blogs.py에 검색 엔드포인트를 추가해보세요.GET /blogs/search?q=키워드형태이며, 경로 순서에 주의해야 합니다.- 모든 API 경로 앞에
/api/v1을 붙여보세요. 화면의common.js도 함께 고쳐야 합니다. crud.py파일을 만들어 데이터베이스를 다루는 함수들을 라우터에서 분리해보세요. 라우터는 HTTP 처리만,crud.py는 데이터 처리만 하도록 나눠보세요.- 지금 구조를 계층별이 아닌 기능별(
app/blog/,app/user/)로 바꿔보세요. 어느 쪽이 더 읽기 편한지 비교해보세요.