[FastAPI] 디자인 패턴-레포지토리 패턴(11-6)

 

 

1. 레포지토리 패턴(Repository Pattern)이란?

• 레포지토리 패턴(Repository Pattern)은 데이터베이스(DB) 액세스를 서비스 로직(Service Layer)에서 분리하는 패턴
• 즉, 데이터베이스와 직접 상호작용하는 코드(쿼리)를 레포지토리 파일에 따로 관리
• 서비스 레이어는 이 레포지토리를 호출하여 데이터를 가져오는 방식

장점

• 데이터베이스 종속성 제거 → 서비스 레이어에서 직접 DB 모델을 다루지 않아, DB 변경 시 최소한의 코드 수정
• 재사용성 증가 → 동일한 DB 조회/저장 로직을 여러 서비스에서 재사용 가능
• 테스트가 쉬워짐 → 가짜(faker) DB를 사용해 테스트 가능 (예: SQLite, Mock DB)
• 코드 가독성 향상 → SQLAlchemy 관련 로직이 분리되어 서비스 코드가 더 깔끔해짐

 

---

2. 사용 방법

프로젝트 구조 예시

my_fastapi_project/
│── main.py
│── routers/
│   ├── user.py
│── services/
│   ├── user_service.py
│── repositories/
│   ├── user_repository.py  # <- 데이터베이스 접근을 담당
│── models/
│   ├── user.py  # <- 데이터베이스 모델 (SQLAlchemy)
│── schemas/
│── database.py

 

models/user.py

• SQLAlchemy를 사용해 User 테이블 정의

from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

 

repositories/user_repository.py

• DB 관련 쿼리를 캡슐화하여 서비스 레이어에서 쉽게 호출 가능
• 레포지토리 파일에서는 SQLAlchemy를 사용하여 DB에 접근하는 코드만 담당
• 즉 쿼리만 작성한다.

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_email(db: Session, email: str):
    return db.query(User).filter(User.email == email).first()  # 기존 코드 옮기기

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

def get_all_users(db: Session):
    return db.query(User).all()

def delete_user(db: Session, user_id: int):
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        db.delete(user)
        db.commit()
    return user

 

services/user_service.py

• 서비스 레이어에서 레포지토리를 호출

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

def register_user(db: Session, user: UserCreate):
    existing_user = get_user_by_email(db, user.email)
    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

def list_users(db: Session):
    return get_all_users(db)

def remove_user(db: Session, user_id: int):
    user = delete_user(db, user_id)
    if not user:
        raise ValueError("삭제할 사용자가 없습니다.")
    return user