FastAPI 백엔드 개발 가이드: API 구현 순서와 모던 개발 방법론 [2탄]

4. API 개발 순서

Step 1: Pydantic Schemas 정의

# app/domain/users/schemas/user_schemas.py
  from pydantic import BaseModel, EmailStr, Field
  from datetime import datetime
  from typing import Optional

  class UserBase(BaseModel):
      email: EmailStr
      username: str = Field(..., min_length=3, max_length=50)

  class UserCreate(UserBase):
      password: str = Field(..., min_length=8)

  class UserUpdate(BaseModel):
      email: Optional[EmailStr] = None
      username: Optional[str] = None

  class UserResponse(UserBase):
      id: int
      is_active: bool
      created_at: datetime

      class Config:
          from_attributes = True  # SQLAlchemy 모델 지원

  class UserListResponse(BaseModel):
      items: list[UserResponse]
      total: int
      page: int
      size: int

왜 먼저 Schema를 정의하는가?

• 계약 우선 개발: API 인터페이스를 먼저 정의
• 타입 안전성: Pydantic이 자동으로 검증 및 직렬화
• 문서 자동화: FastAPI가 자동으로 OpenAPI 스펙 생성

Step 2: Repository Layer 구현

# app/domain/users/repositories/user_repository.py
  from sqlalchemy.orm import Session
  from sqlalchemy import and_, or_
  from typing import List, Optional
  from app.domain.users.models.user import User
  from app.domain.users.schemas.user_schemas import UserCreate, UserUpdate

  class UserRepository:
      def __init__(self, db: Session):
          self.db = db

      def create(self, user_data: UserCreate) -> User:
          """사용자 생성"""
          db_user = User(**user_data.dict(exclude={'password'}))
          db_user.hashed_password = hash_password(user_data.password)

          self.db.add(db_user)
          self.db.commit()
          self.db.refresh(db_user)
          return db_user

      def get_by_id(self, user_id: int) -> Optional[User]:
          """ID로 사용자 조회"""
          return self.db.query(User).filter(User.id == user_id).first()

      def get_by_email(self, email: str) -> Optional[User]:
          """이메일로 사용자 조회"""
          return self.db.query(User).filter(User.email == email).first()

      def get_list(self, skip: int = 0, limit: int = 100) -> List[User]:
          """사용자 목록 조회 (페이징)"""
          return self.db.query(User).offset(skip).limit(limit).all()

      def update(self, user_id: int, user_data: UserUpdate) -> Optional[User]:
          """사용자 정보 업데이트"""
          db_user = self.get_by_id(user_id)
          if not db_user:
              return None

          update_data = user_data.dict(exclude_unset=True)
          for field, value in update_data.items():
              setattr(db_user, field, value)

          self.db.commit()
          self.db.refresh(db_user)
          return db_user

      def delete(self, user_id: int) -> bool:
          """사용자 삭제 (소프트 삭제)"""
          db_user = self.get_by_id(user_id)
          if not db_user:
              return False

          db_user.is_active = False
          self.db.commit()
          return True

Repository 패턴을 사용하는 이유

• 데이터 액세스 추상화: 데이터베이스 구현체와 비즈니스 로직 분리
• 테스트 용이성: Mock Repository로 단위 테스트 가능
• 변경 용이성: 데이터베이스 변경 시 Repository만 수정

Step 3: Service Layer 구현

# app/domain/users/services/user_service.py
  from sqlalchemy.orm import Session
  from typing import List, Optional
  from app.domain.users.repositories.user_repository import UserRepository
  from app.domain.users.schemas.user_schemas import (
      UserCreate, UserUpdate, UserResponse, UserListResponse
  )
  from app.core.exceptions import NotFoundException, ConflictException

  class UserService:
      def __init__(self, db: Session):
          self.repository = UserRepository(db)

      def create_user(self, user_data: UserCreate) -> UserResponse:
          """사용자 생성 비즈니스 로직"""
          # 비즈니스 규칙 검증
          existing_user = self.repository.get_by_email(user_data.email)
          if existing_user:
              raise ConflictException("이미 존재하는 이메일입니다.")

          # 사용자 생성
          user = self.repository.create(user_data)

          # 추가 비즈니스 로직 (이메일 발송, 로깅 등)
          self._send_welcome_email(user.email)

          return UserResponse.from_orm(user)

      def get_user(self, user_id: int) -> UserResponse:
          """사용자 조회"""
          user = self.repository.get_by_id(user_id)
          if not user or not user.is_active:
              raise NotFoundException("사용자를 찾을 수 없습니다.")

          return UserResponse.from_orm(user)

      def get_users(self, page: int = 1, size: int = 20) -> UserListResponse:
          """사용자 목록 조회"""
          skip = (page - 1) * size
          users = self.repository.get_list(skip=skip, limit=size)
          total = self.repository.count()

          return UserListResponse(
              items=[UserResponse.from_orm(user) for user in users],
              total=total,
              page=page,
              size=size
          )

      def update_user(self, user_id: int, user_data: UserUpdate) -> UserResponse:        
          """사용자 정보 업데이트"""
          # 권한 검증 로직
          if user_data.email:
              existing_user = self.repository.get_by_email(user_data.email)
              if existing_user and existing_user.id != user_id:
                  raise ConflictException("이미 사용 중인 이메일입니다.")

          user = self.repository.update(user_id, user_data)
          if not user:
              raise NotFoundException("사용자를 찾을 수 없습니다.")

          return UserResponse.from_orm(user)

      def delete_user(self, user_id: int) -> bool:
          """사용자 삭제"""
          success = self.repository.delete(user_id)
          if not success:
              raise NotFoundException("사용자를 찾을 수 없습니다.")

          return True

      def _send_welcome_email(self, email: str):
          """환영 이메일 발송 (내부 메서드)"""
          # 이메일 발송 로직
          pass

Service Layer의 역할:

• 비즈니스 로직 집중: 도메인 규칙 구현
• 트랜잭션 관리: 복잡한 비즈니스 연산의 원자성 보장
• 외부 서비스 호출: 이메일, 알림 등 부가 기능
• 검증 및 변환: 데이터 검증과 응답 형식 변환

Step 4: Controller Layer 구현

# app/domain/users/controllers/user_controller.py
  from fastapi import APIRouter, Depends, HTTPException, Query, Path
  from sqlalchemy.orm import Session
  from typing import List

  from app.core.database import get_db
  from app.core.dependencies import get_current_user
  from app.domain.users.services.user_service import UserService
  from app.domain.users.schemas.user_schemas import (
      UserCreate, UserUpdate, UserResponse, UserListResponse
  )

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

  @router.post("/", response_model=UserResponse, status_code=201)
  async def create_user(
      user_data: UserCreate,
      db: Session = Depends(get_db)
  ):
      """
      사용자 생성

      - **email**: 유효한 이메일 주소
      - **username**: 3-50자 사용자명
      - **password**: 8자 이상 비밀번호
      """
      service = UserService(db)
      return service.create_user(user_data)

  @router.get("/{user_id}", response_model=UserResponse)
  async def get_user(
      user_id: int = Path(..., gt=0, description="사용자 ID"),
      db: Session = Depends(get_db),
      current_user = Depends(get_current_user)  # 인증 필요
  ):
      """사용자 상세 조회"""
      service = UserService(db)
      return service.get_user(user_id)

  @router.get("/", response_model=UserListResponse)
  async def get_users(
      page: int = Query(1, ge=1, description="페이지 번호"),
      size: int = Query(20, ge=1, le=100, description="페이지 크기"),
      db: Session = Depends(get_db)
  ):
      """사용자 목록 조회 (페이징)"""
      service = UserService(db)
      return service.get_users(page=page, size=size)

  @router.put("/{user_id}", response_model=UserResponse)
  async def update_user(
      user_id: int = Path(..., gt=0),
      user_data: UserUpdate,
      db: Session = Depends(get_db),
      current_user = Depends(get_current_user)
  ):
      """사용자 정보 업데이트"""
      # 권한 검증: 본인 또는 관리자만 수정 가능
      if current_user.id != user_id and not current_user.is_admin:
          raise HTTPException(status_code=403, detail="권한이 없습니다.")

      service = UserService(db)
      return service.update_user(user_id, user_data)

  @router.delete("/{user_id}", status_code=204)
  async def delete_user(
      user_id: int = Path(..., gt=0),
      db: Session = Depends(get_db),
      current_user = Depends(get_current_user)
  ):
      """사용자 삭제"""
      if current_user.id != user_id and not current_user.is_admin:
          raise HTTPException(status_code=403, detail="권한이 없습니다.")

      service = UserService(db)
      service.delete_user(user_id)

Controller Layer의 역할

• HTTP 요청/응답 처리: 요청 파라미터 검증, 응답 직렬화
• 라우팅: URL 경로와 HTTP 메서드 매핑
• 인증/인가: 사용자 권한 검증
• 예외 처리: HTTP 상태 코드와 에러 메시지 변환

Step 5: Exception Handling

# app/core/exceptions.py
  class BaseCustomException(Exception):
      def __init__(self, message: str):
          self.message = message
          super().__init__(self.message)

  class NotFoundException(BaseCustomException):
      pass

  class ConflictException(BaseCustomException):
      pass

  class ValidationException(BaseCustomException):
      pass

  # app/core/exception_handlers.py
  from fastapi import Request, HTTPException
  from fastapi.responses import JSONResponse

  async def not_found_exception_handler(request: Request, exc: NotFoundException):       
      return JSONResponse(
          status_code=404,
          content={"error": "Not Found", "message": exc.message}
      )

  async def conflict_exception_handler(request: Request, exc: ConflictException):        
      return JSONResponse(
          status_code=409,
          content={"error": "Conflict", "message": exc.message}
      )

  # main.py에서 등록
  app.add_exception_handler(NotFoundException, not_found_exception_handler)
  app.add_exception_handler(ConflictException, conflict_exception_handler)