[FastAPI] 디자인 패턴-서비스 레이어 패턴(11-5)

 

1. 서비스 레이어 패턴 (Service Layer Pattern) 이란

서비스 레이어 패턴(Service Layer Pattern)은 비즈니스 로직을 엔드포인트(FastAPI 라우터)와 데이터 액세스(Repository) 레이어로부터 분리하는 아키텍처 패턴

 

FastAPI에서

라우터는 요청을 받고,

서비스 레이어는 핵심 로직을 수행하며,

리포지토리는 데이터베이스와 직접 상호작용

 

장점

• 비즈니스 로직과 API 엔드포인트 분리 → 코드가 깔끔
• 재사용성 증가 → 같은 로직을 여러 엔드포인트에서 사용 가능
• 유지보수 용이 → 데이터베이스 변경 시 API 코드 변경 없이 서비스 레이어에서만 수정
• 유닛 테스트 용이 → 데이터베이스 없이 서비스 레이어 단독으로 테스트 가능

 

---

2. 서비스 레이어 패턴 사용 방법

프로젝트 구조 예시

각 역할을 분리하여 관리하며, service/ 폴더에 비즈니스 로직을 담당하는 user_service.py를 배치

my_fastapi_project/
│── main.py
│── routers/
│   ├── user.py
│── services/
│   ├── user_service.py
│── repositories/
│   ├── user_repository.py
│── models/
│── schemas/
│── database.py

 

 

services/user_service.py

• Service Layer에서 데이터베이스 액세스를 직접 하지 않고, Repository Layer를 호출하여 데이터를 가져옴
• 해당 페이지에서 비즈니스 로직을 작성
• 비즈니스 로직: 핵심 기능, 예외처리 같은 내용

 

예외발생 시, 적절한 에러를 발생시킴 (raise 사용), 예외 처리는 하지 않음

• Service Layer는 FastAPI에 종속되지 않아야 함
• HTTPException은 FastAPI의 예외 객체로, Service Layer는 이를 직접 알 필요가 없음
• 서비스 로직이 FastAPI와 강하게 결합됨 → 테스트 & 재사용성이 낮아짐

from sqlalchemy.orm import Session
from repositories.user_repository import create_user, get_user_by_id
from schemas.user import UserCreate

def register_user(db: Session, user: UserCreate):
    # 비즈니스 로직 예시: 중복 이메일 방지
    existing_user = db.query(User).filter_by(email=user.email).first()
    if existing_user:
        raise ValueError("이미 존재하는 이메일입니다.")
    
    return create_user(db, user)

def retrieve_user(db: Session, user_id: int):
    user = get_user_by_id(db, user_id)
    if not user:
        raise ValueError("사용자를 찾을 수 없습니다.")
    return user

 

repositories/user_repository.py

• Repository Layer
• 데이터베이스 접근을 담당함

from sqlalchemy.orm import Session
from models.user import User
from schemas.user import UserCreate

def create_user(db: Session, user: UserCreate):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user_by_id(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

 

routers/user.py

• 라우터는 서비스 레이어를 호출하여 실제 작업을 수행
• Service Layer에서 발생한 예외를 try-except로 감싸고, HTTPException으로 변환하여 적절한 HTTP 응답 반환

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db
from schemas.user import UserCreate, UserResponse
from services.user_service import register_user, retrieve_user

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

@router.post("/", response_model=UserResponse)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    try:
        return register_user(db, user)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: Session = Depends(get_db)):
    try:
        return retrieve_user(db, user_id)
    except ValueError as e:
        raise HTTPException(status_code=404, detail=str(e))