DocsPricingBlog
Contact Us
Log inSign up for FREE
Back to Tech Articles

Python 웹 애플리케이션에서 리포지토리 패턴을 사용하여 비즈니스 로직과 데이터 액세스 분리하기

Sep 14, 2025
# Misc

Min-jun Kim

Dev Intern · Leapcell

This post is also available in English, 日本語, and Deutsch
Cover of "Python 웹 애플리케이션에서 리포지토리 패턴을 사용하여 비즈니스 로직과 데이터 액세스 분리하기"

소개

빠르게 변화하는 웹 개발 세계에서 강력하고 확장 가능하며 유지보수 가능한 애플리케이션을 구축하는 것이 무엇보다 중요합니다. 파이썬은 우아한 문법과 방대한 생태계를 갖추고 있어 Django 및 Flask와 같은 웹 개발 프레임워크에서 인기가 높습니다. 그러나 애플리케이션이 복잡해짐에 따라 공통적인 문제가 발생합니다. 바로 비즈니스 로직과 데이터 액세스의 긴밀한 결합입니다. 이러한 결합은 종종 이해, 테스트 및 수정하기 어려운 코드로 이어져 결국 개발 속도를 늦추고 버그 발생 위험을 높입니다.

핵심 비즈니스 규칙이 SQL 쿼리나 ORM 호출과 얽혀 있는 시나리오를 상상해 보세요. 데이터베이스 스키마의 사소해 보이는 변경이라도 애플리케이션의 여러 부분에 파급되어 광범위한 수정과 재테스트가 필요할 수 있습니다. 이것이 바로 리포지토리 패턴이 해결하려는 문제입니다. 리포지토리 패턴은 명확한 추상화 계층을 도입하여 이러한 종속성을 푸는 데 도움을 주므로 파이썬 웹 애플리케이션은 더 복원력이 있고 적응력이 뛰어나며 유지보수가 즐거워집니다. 이 글에서는 리포지토리 패턴이 이 중요한 분리를 어떻게 달성하는지 살펴보고 애플리케이션 아키텍처를 더 강력하고 미래 지향적으로 만들 것입니다.

리포지토리 패턴 이해하기

구현에 들어가기 전에 관련된 핵심 개념을 명확하게 이해해 봅시다.

  • 비즈니스 로직: 이는 애플리케이션이 작동하고 데이터를 조작하는 방식을 정의하는 특정 규칙과 프로세스를 의미합니다. 데이터 저장 또는 검색 "방법"과 독립적으로 애플리케이션이 "무엇"을 하는지에 해당합니다. 예를 들어, 사용자 입력을 검증하거나, 주문 총액을 계산하거나, 할인 규칙을 적용하는 것이 비즈니스 로직의 예입니다.
  • 데이터 액세스: 여기에는 데이터베이스(SQL, NoSQL), 파일 시스템 또는 외부 API와 같은 영구 저장소와 상호 작용하는 메커니즘이 포함됩니다. 데이터가 저장되고 검색되는 "방법"에 해당합니다. SQL 쿼리 실행, SQLAlchemy 또는 Django의 ORM과 같은 ORM(객체 관계형 매퍼) 사용 또는 API 요청 만들기가 예입니다.
  • 리포지토리 패턴: 핵심적으로 리포지토리 패턴은 도메인 객체의 메모리 내 컬렉션 역할을 합니다. 특정 데이터 액세스 기술에서 애플리케이션의 비즈니스 로직을 분리하는 데이터 저장 및 검색을 위한 명확한 추상 인터페이스를 제공합니다. 데이터 영속성 계층에 대한 파사드(facade)라고 생각하면 됩니다. 비즈니스 로직이 데이터와 상호 작용해야 할 때 데이터베이스나 ORM과 직접 통신하는 것이 아니라 리포지토리와 통신합니다.

패턴의 기본 원칙

리포지토리 패턴의 핵심 원칙은 추상 인터페이스 뒤에 데이터 액세스 로직을 캡슐화하는 것입니다. 이 인터페이스는 일반적인 데이터 작업(예: get_by_id, add, update, delete, query)에 대한 메서드를 정의합니다. 비즈니스 로직은 기본 데이터 영속성 메커니즘을 알지 못한 채 이 인터페이스와만 상호 작용합니다. 이는 몇 가지 중요한 이점을 제공합니다.

  1. 분리: 비즈니스 로직은 더 이상 특정 데이터베이스 기술이나 ORM에 결합되지 않습니다. PostgreSQL에서 MongoDB로, 또는 SQLAlchemy에서 다른 ORM으로 전환하기로 결정하면 비즈니스 로직이 아닌 리포지토리 구현만 수정하면 됩니다.
  2. 테스트 용이성: 리포지토리는 비즈니스 로직의 단위 테스트를 훨씬 쉽게 만듭니다. 라이브 데이터베이스 연결이 필요한 대신, 테스트 중에 리포지토리 인터페이스를 메모리 내 구현으로 쉽게 모의(mock)하거나 대체할 수 있습니다. 이렇게 하면 테스트 속도가 빨라지고 외부 종속성에 대한 의존성이 줄어듭니다.
  3. 유지보수성: 데이터 영속성 계층에 대한 변경은 국소적인 영향을 미칩니다. 수정은 리포지토리 구현에 국한되어 코드베이스를 이해하고 유지보수하기 쉽게 만듭니다.
  4. 가독성: 비즈니스 로직은 데이터 액세스의 복잡한 부분에 대해 걱정할 필요가 없으므로 더 명확하고 집중됩니다.

예시: 작업 관리 애플리케이션

간단한 작업 관리 애플리케이션을 고려해 보겠습니다. Task 엔티티를 사용하고 리포지토리 패턴을 적용하는 방법을 시연할 것입니다.

1. 도메인 모델 정의

먼저 애플리케이션의 핵심 엔티티를 나타내는 도메인 모델을 정의합니다. 이것은 데이터베이스 관련 문제를 피할 수 있는 일반 파이썬 객체여야 합니다.

# models.py
import dataclasses
import datetime
from typing import Optional

@dataclasses.dataclass
class Task:
    id: Optional[int] = None
    title: str
    description: Optional[str] = None
    completed: bool = False
    created_at: datetime.datetime = dataclasses.field(default_factory=datetime.datetime.now)
    updated_at: datetime.datetime = dataclasses.field(default_factory=datetime.datetime.now)

    def mark_as_completed(self):
        if not self.completed:
            self.completed = True
            self.updated_at = datetime.datetime.now()
            return True
        return False

    def update_details(self, title: Optional[str] = None, description: Optional[str] = None):
        if title:
            self.title = title
            self.updated_at = datetime.datetime.now()
        if description:
            self.description = description
            self.updated_at = datetime.datetime.now()

2. 리포지토리 인터페이스 정의 (추상 기본 클래스)

다음으로 TaskRepository에 대한 추상 기본 클래스(abc 모듈 사용)를 정의합니다. 이 계약은 모든 실제 작업 리포지토리가 반드시 구현해야 하는 메서드를 지정합니다.

# repositories/interfaces.py
import abc
from typing import List, Optional
from models import Task

class TaskRepository(abc.ABC):
    @abc.abstractmethod
    def add(self, task: Task) -> Task:
        """Adds a new task to the repository."""
        raise NotImplementedError

    @abc.abstractmethod
    def get_by_id(self, task_id: int) -> Optional[Task]:
        """Retrieves a task by its ID."""
        raise NotImplementedError

    @abc.abstractmethod
    def get_all(self, completed: Optional[bool] = None) -> List[Task]:
        """Retrieves all tasks, optionally filtered by completion status."""
        raise NotImplementedError

    @abc.abstractmethod
    def update(self, task: Task) -> Task:
        """Updates an existing task."""
        raise NotImplementedError

    @abc.abstractmethod
    def delete(self, task_id: int) -> None:
        """Deletes a task by its ID."""
        raise NotImplementedError

3. 실제 리포지토리 구현

이제 다양한 데이터 영속성 메커니즘에 대해 TaskRepository의 실제 구현을 만들 수 있습니다.

메모리 내 리포지토리(테스트 및 간단한 경우용)
# repositories/in_memory.py
from typing import List, Optional
from repositories.interfaces import TaskRepository
from models import Task

class InMemoryTaskRepository(TaskRepository):
    def __init__(self):
        self._tasks: List[Task] = []
        self._next_id = 1

    def add(self, task: Task) -> Task:
        task.id = self._next_id
        self._next_id += 1
        self._tasks.append(task)
        return task

    def get_by_id(self, task_id: int) -> Optional[Task]:
        for task in self._tasks:
            if task.id == task_id:
                return task
        return None

    def get_all(self, completed: Optional[bool] = None) -> List[Task]:
        if completed is None:
            return list(self._tasks)
        return [task for task in self._tasks if task.completed == completed]

    def update(self, task: Task) -> Task:
        for i, existing_task in enumerate(self._tasks):
            if existing_task.id == task.id:
                self._tasks[i] = task
                return task
        raise ValueError(f"Task with ID {task.id} not found for update.")

    def delete(self, task_id: int) -> None:
        self._tasks = [task for task in self._tasks if task.id != task_id]
SQLAlchemy 리포지토리 (관계형 데이터베이스용)

SQLAlchemy 및 데이터베이스가 구성되어 있다고 가정하면, 여기 개념적인 예가 있습니다. 간결성을 위해 전체 SQLAlchemy 설정(엔진, 세션 등)은 생략하지만 리포지토리의 논리에 중점을 둡니다.

# repositories/sqlalchemy_repo.py
from typing import List, Optional
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base

from repositories.interfaces import TaskRepository
from models import Task

# --- SQLAlchemy 특정 ORM 모델 매핑 ---
Base = declarative_base()

class SQLAlchemyTask(Base):
    __tablename__ = 'tasks'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String, nullable=False)
    description = Column(String)
    completed = Column(Boolean, default=False)
    created_at = Column(DateTime)
    updated_at = Column(DateTime)

    def to_domain_model(self) -> Task:
        return Task(
            id=self.id,
            title=self.title,
            description=self.description,
            completed=self.completed,
            created_at=self.created_at,
            updated_at=self.updated_at
        )

    @staticmethod
    def from_domain_model(domain_task: Task) -> 'SQLAlchemyTask':
        return SQLAlchemyTask(
            id=domain_task.id,
            title=domain_task.title,
            description=domain_task.description,
            completed=domain_task.completed,
            created_at=domain_task.created_at,
            updated_at=domain_task.updated_at
        )
# --- ORM 모델 매핑 끝 ---

class SQLAlchemyTaskRepository(TaskRepository):
    def __init__(self, session: Session):
        self.session = session

    def add(self, task: Task) -> Task:
        sa_task = SQLAlchemyTask.from_domain_model(task)
        self.session.add(sa_task)
        self.session.commit()
        # 생성된 ID가 있는 경우 도메인 모델 업데이트
        task.id = sa_task.id
        return task

    def get_by_id(self, task_id: int) -> Optional[Task]:
        sa_task = self.session.query(SQLAlchemyTask).filter_by(id=task_id).first()
        if sa_task:
            return sa_task.to_domain_model()
        return None

    def get_all(self, completed: Optional[bool] = None) -> List[Task]:
        query = self.session.query(SQLAlchemyTask)
        if completed is not None:
            query = query.filter_by(completed=completed)
        return [sa_task.to_domain_model() for sa_task in query.all()]

    def update(self, task: Task) -> Task:
        sa_task = self.session.query(SQLAlchemyTask).filter_by(id=task.id).first()
        if not sa_task:
            raise ValueError(f"Task with ID {task.id} not found for update.")
        
        sa_task.title = task.title
        sa_task.description = task.description
        sa_task.completed = task.completed
        sa_task.updated_at = task.updated_at # 도메인에서 업데이트한다고 가정
        
        self.session.commit()
        return task

    def delete(self, task_id: int) -> None:
        sa_task = self.session.query(SQLAlchemyTask).filter_by(id=task_id).first()
        if sa_task:
            self.session.delete(sa_task)
            self.session.commit()

4. 애플리케이션 서비스 / 비즈니스 로직 계층

이제 비즈니스 로직(종종 "서비스" 또는 "유스 케이스"에 배치됨)은 작업이 저장되는 방식을 알지 못한 채 TaskRepository 인터페이스와 상호 작용할 수 있습니다.

# services.py
from typing import List, Optional
from models import Task
from repositories.interfaces import TaskRepository

class TaskService:
    def __init__(self, task_repository: TaskRepository):
        self.task_repository = task_repository

    def create_task(self, title: str, description: Optional[str] = None) -> Task:
        new_task = Task(title=title, description=description)
        return self.task_repository.add(new_task)

    def get_task_by_id(self, task_id: int) -> Optional[Task]:
        return self.task_repository.get_by_id(task_id)

    def list_tasks(self, completed: Optional[bool] = None) -> List[Task]:
        return self.task_repository.get_all(completed=completed)

    def mark_task_complete(self, task_id: int) -> Optional[Task]:
        task = self.task_repository.get_by_id(task_id)
        if task and task.mark_as_completed(): # 도메인 모델에 대한 비즈니스 로직
            return self.task_repository.update(task)
        return None

    def update_task_details(self, task_id: int, title: Optional[str] = None, description: Optional[str] = None) -> Optional[Task]:
        task = self.task_repository.get_by_id(task_id)
        if task:
            task.update_details(title, description) # 도메인 모델에 대한 비즈니스 로직
            return self.task_repository.update(task)
        return None

    def delete_task(self, task_id: int) -> None:
        self.task_repository.delete(task_id)

5. 웹 애플리케이션 계층 (예: Flask)

Flask(또는 Django, FastAPI) 뷰에서는 TaskService(이는 다시 TaskRepository를 가짐)를 주입하게 됩니다.

# app.py (간소화된 Flask 예제)
from flask import Flask, request, jsonify
# 여기서 SQLAlchemy 세션 설정을 가정합니다.
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from repositories.sqlalchemy_repo import SQLAlchemyTaskRepository, Base as SQLBase
from repositories.in_memory import InMemoryTaskRepository
from services import TaskService
from models import Task
import dataclasses

app = Flask(__name__)

# --- 의존성 주입 설정 ---
# 시연을 위해 리포지토리를 쉽게 전환할 수 있습니다.
# 테스트/개발용 메모리 내 사용:
# task_repo_instance = InMemoryTaskRepository()

# 프로덕션용 SQLAlchemy 사용:
DATABASE_URL = "sqlite:///./tasks.db"
engine = create_engine(DATABASE_URL)
SQLBase.metadata.create_all(bind=engine) # 테이블 생성
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db_session() -> Session:
    db_session = SessionLocal()
    try:
        yield db_session
    finally:
        db_session.close()

# 실제 앱에서는 이 설정을 프레임워크의 DI 시스템(예: Flask-Injector, FastAPI Depends)과 통합해야 합니다.
# 단순화를 위해 세션을 수동으로 가져와 전달합니다.
# 리포지토리를 위한 전역 또는 팩토리를 사용할 수 있습니다.
# 예: 리포지토리 제공을 위한 팩토리 사용
def get_task_repository() -> SQLAlchemyTaskRepository:
    # 실제 앱에서는 세션 수명 주기를 관리해야 합니다(예: 각 요청마다 세션 가져오기).
    return SQLAlchemyTaskRepository(next(get_db_session())) 

def get_task_service() -> TaskService:
    return TaskService(get_task_repository())
# --- 의존성 주입 설정 끝 ---


@app.route("/tasks", methods=["POST"])
def create_task_endpoint():
    data = request.json
    service = get_task_service()
    task = service.create_task(title=data["title"], description=data.get("description"))
    return jsonify(dataclasses.asdict(task)), 201

@app.route("/tasks", methods=["GET"])
def get_tasks_endpoint():
    completed_param = request.args.get("completed")
    completed_filter = None
    if completed_param is not None:
        completed_filter = completed_param.lower() == 'true'

    service = get_task_service()
    tasks = service.list_tasks(completed=completed_filter)
    return jsonify([dataclasses.asdict(task) for task in tasks])

@app.route("/tasks/<int:task_id>", methods=["GET"])
def get_task_endpoint(task_id: int):
    service = get_task_service()
    task = service.get_task_by_id(task_id)
    if task:
        return jsonify(dataclasses.asdict(task))
    return jsonify({"message": "Task not found"}), 404

@app.route("/tasks/<int:task_id>/complete", methods=["POST"])
def complete_task_endpoint(task_id: int):
    service = get_task_service()
    task = service.mark_task_complete(task_id)
    if task:
        return jsonify(dataclasses.asdict(task))
    return jsonify({"message": "Task not found or already completed"}), 404

# ... (업데이트, 삭제에 대한 다른 엔드포인트)

if __name__ == "__main__":
    app.run(debug=True)

애플리케이션 시나리오

리포지토리 패턴은 여러 시나리오에서 특히 유익합니다.

  • 복잡한 비즈니스 로직: 애플리케이션에 자주 변경되는 복잡한 비즈니스 규칙이 포함되어 있을 때 데이터 관련 문제로부터 분리하는 것이 중요합니다.
  • 다양한 데이터 소스: 애플리케이션이 다른 데이터베이스, API 또는 파일 시스템에서 데이터를 가져와야 하는 경우 리포지토리는 통합 인터페이스를 제공합니다.
  • 테스트 요구 사항: 높은 테스트 커버리지와 빠른 단위 테스트가 필요한 애플리케이션의 경우 리포지토리를 사용하면 비즈니스 로직을 격리하여 모의(mocking)하고 테스트할 수 있습니다.
  • 레거시 시스템 통합: 오래된 시스템이나 특이한 데이터 액세스 방법이 있는 타사 API와 통합할 때 리포지토리는 이러한 복잡성을 캡슐화합니다.
  • 확장성 및 진화: 애플리케이션이 확장되거나 데이터 저장 기술을 변경할 것으로 예상되는 경우 리포지토리는 리팩토링을 최소화하면서 전환을 용이하게 합니다.

결론

리포지토리 패턴은 많은 파이썬 웹 애플리케이션에서 흔히 볼 수 있는 긴밀하게 결합된 비즈니스 로직 및 데이터 액세스 계층을 푸는 강력한 솔루션을 제공합니다. 명확한 추상 인터페이스를 도입하여 깨끗한 아키텍처를 촉진하고, 테스트 용이성을 향상시키며, 유지보수성을 크게 향상시킵니다. 약간의 초기 설계 및 코드 작업이 필요하지만, 유연성, 안정성 및 개발자 생산성 측면에서 장기적인 이점은 투자가 충분히 가치가 있으며, 변경에 더 탄력적이고 진화하기 쉬운 애플리케이션을 구축할 수 있습니다. 리포지토리 패턴을 채택하여 수년간 강력하고 테스트 가능하며 유지보수 가능한 파이썬 웹 애플리케이션을 구축하십시오.

Share this article

HackerNews

More Posts from Leapcell

Popular Posts