Flask의 뚱뚱한 라우트 줄이기: 서비스 및 리포지토리 계층 가이드

소개

웹 개발의 활기찬 세계에서 Flask는 가볍고 비정형적인 특성으로 인해 개발자에게 엄청난 유연성을 제공합니다.

하지만 이러한 자유는 때때로 흔한 함정, 즉 "뚱뚱한 라우트"로 이어질 수 있습니다. 본질적으로 뚱뚱한 라우트 핸들러는 너무 많은 일을 하는 Flask 뷰 함수입니다. 단일 함수 내에서 HTTP 요청 처리, 입력 유효성 검사, 복잡한 비즈니스 로직 실행, 데이터베이스 직접 상호 작용, 응답 준비까지 모두 수행합니다. 이러한 관심사의 긴밀한 결합은 작은 프로젝트에서는 편리해 보일 수 있지만, 애플리케이션이 성장함에 따라 빠르게 심각한 병목 현상이 됩니다. 읽기 어렵고, 테스트하기 까다로우며, 유지보수하거나 확장하기 어려운 코드로 이어집니다.

다행히도 잘 정립된 아키텍처 패턴을 채택하여 이러한 제어되지 않는 라우트를 제어할 수 있습니다. 이 글에서는 전용 서비스 및 리포지토리 계층을 도입하여 일반적인 "뚱뚱한" Flask 라우트를 보다 구조화되고 유지보수 가능하며 테스트 가능한 디자인으로 리팩터링하는 과정을 안내합니다. 이러한 관심사 분리는 Flask 뷰를 정리할 뿐만 아니라 강력한 애플리케이션 개발을 위한 견고한 기반을 마련합니다.

핵심 개념 설명

코드를 살펴보기 전에 논의할 아키텍처 패턴에 대한 명확한 이해를 확립해 보겠습니다.

• 컨트롤러 (또는 뷰 계층): Flask의 컨텍스트에서, 이것은 일반적으로 라우트 핸들러 함수입니다. 주요 책임은 들어오는 HTTP 요청을 처리하고, 쿼리 매개변수 또는 요청 본문을 구문 분석하며, 다른 계층으로 작업을 위임하고, HTTP 응답을 반환하는 것입니다. 비즈니스 로직을 포함하거나 데이터베이스와 직접 상호 작용해서는 안 됩니다.
• 서비스 계층 (또는 비즈니스 로직 계층): 이 계층은 애플리케이션의 핵심 비즈니스 규칙과 워크플로를 캡슐화합니다. 종종 하나 이상의 리포지토리의 메서드를 호출하고, 유효성 검사를 적용하며, 복잡한 사용 사례를 처리하여 작업을 조정합니다. 서비스 계층은 컨트롤러와 데이터 액세스 계층 사이의 다리 역할을 합니다. 애플리케이션의 무엇이 발생하는 곳입니다.
• 리포지토리 계층 (또는 데이터 액세스 계층): 이 계층은 데이터 저장 및 검색 메커니즘을 추상화하는 역할을 합니다. 데이터베이스(또는 외부 API 또는 파일 시스템과 같은 모든 데이터 소스)와 상호 작용하기 위한 깔끔한 API를 제공하며, 서비스 계층을 기본 지속성 세부 정보로부터 보호합니다. 데이터가 관리되는 방법이 발생하는 곳입니다.

이러한 관심사를 분리함으로써 다음을 달성할 수 있습니다.

• 향상된 유지보수성: 데이터 저장소의 변경이 반드시 비즈니스 로직에 영향을 미치지 않으며, 그 반대도 마찬가지입니다.
• 쉬운 테스트: 각 계층은 의존성에 대한 목을 사용하여 격리하여 테스트할 수 있습니다.
• 향상된 가독성: 각 구성 요소에 하나씩 명확한 책임이 있으므로 코드가 더 집중되고 이해하기 쉬워집니다.
• 향상된 확장성: 모듈식 디자인은 본질적으로 향후 성장과 아키텍처 변경에 더 잘 적응할 수 있습니다.

뚱뚱한 Flask 라우트 리팩터링

일반적인 Flask 애플리케이션 시나리오를 상상해 보겠습니다. 사용자 관리입니다. 새 사용자를 생성하는 일반적인 "뚱뚱한" 라우트는 다음과 같을 수 있습니다.

초기 뚱뚱한 라우트 예시:

# app.py (초기 버전)
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f'<User {self.username}>'

@app.before_first_request
def create_tables():
    db.create_all()

@app.route('/users', methods=['POST'])
def create_user_fat():
    data = request.get_json()
    if not data or not all(key in data for key in ['username', 'email']):
        return jsonify({"message": "Missing username or email"}), 400

    username = data['username']
    email = data['email']

    # 비즈니스 로직 및 데이터 액세스 혼합
    existing_user = User.query.filter_by(username=username).first()
    if existing_user:
        return jsonify({"message": "Username already exists"}), 409
    
    existing_email = User.query.filter_by(email=email).first()
    if existing_email:
        return jsonify({"message": "Email already exists"}), 409

    new_user = User(username=username, email=email)
    db.session.add(new_user)
    db.session.commit()

    return jsonify({"message": "User created successfully", "user_id": new_user.id}), 201

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

이 create_user_fat 함수는 너무 많은 일을 합니다. 요청 구문 분석, 입력 유효성 검사, 기존 사용자 확인, 새 사용자 객체 생성 및 데이터베이스에 지속하는 작업을 처리합니다.

이를 서비스 및 리포지토리 계층으로 리팩터링해 보겠습니다.

1단계: 리포지토리 계층 정의

먼저 User 객체와 관련된 모든 데이터베이스 상호 작용을 담당하는 UserRepository를 생성합니다.

# app/repositories/user_repository.py
from flask_sqlalchemy import SQLAlchemy

class UserRepository:
    def __init__(self, db: SQLAlchemy):
        self.db = db
        self.User = db.Model._decl_class_registry.get('User') # User 모델이 SQLAlchemy에 등록되어 있다고 가정

    def get_by_username(self, username: str):
        return self.User.query.filter_by(username=username).first()

    def get_by_email(self, email: str):
        return self.User.query.filter_by(email=email).first()

    def add(self, user_data: dict):
        new_user = self.User(username=user_data['username'], email=user_data['email'])
        self.db.session.add(new_user)
        self.db.session.commit()
        return new_user

    def get_all(self):
        return self.User.query.all()

    def get_by_id(self, user_id: int):
        return self.User.query.get(user_id)

참고: 잘 구조화된 프로젝트에서는 일반적으로 app/models.py에 User 모델을 정의하고 리포지토리에 가져올 것입니다. 여기서는 단순성을 위해 db.Model._decl_class_registry를 통해 액세스합니다.

2단계: 서비스 계층 정의

다음으로 사용자 관리를 위한 비즈니스 로직을 캡슐화하는 UserService를 생성합니다. 데이터베이스와 상호 작용하기 위해 UserRepository를 사용합니다.

# app/services/user_service.py
from typing import Dict, Union
from app.repositories.user_repository import UserRepository # 올바른 경로라고 가정

class UserService:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo

    def create_user(self, user_data: Dict[str, str]) -> Union[Dict, None]:
        if not all(key in user_data for key in ['username', 'email']):
            raise ValueError("Missing username or email")
        
        username = user_data['username']
        email = user_data['email']

        # 고유 사용자 이름/이메일에 대한 비즈니스 로직
        if self.user_repo.get_by_username(username):
            raise ValueError("Username already exists")
        
        if self.user_repo.get_by_email(email):
            raise ValueError("Email already exists")

        user = self.user_repo.add(user_data)
        return {"id": user.id, "username": user.username, "email": user.email}

    def get_user_by_id(self, user_id: int):
        user = self.user_repo.get_by_id(user_id)
        if user:
            return {"id": user.id, "username": user.username, "email": user.email}
        return None

    def get_all_users(self):
        users = self.user_repo.get_all()
        return [{"id": u.id, "username": u.username, "email": u.email} for u in users]

3단계: Flask 라우트 리팩터링

마지막으로 Flask 라우트가 훨씬 깔끔해집니다. HTTP 요청/응답 로직만 처리하고 실제 작업은 UserService에 위임합니다.

# app.py (리팩터링된 버전)
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

# 새로운 계층 가져오기
from app.repositories.user_repository import UserRepository
from app.services.user_service import UserService

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# 여기에 User 모델을 정의하거나 app/models.py에 정의하고 가져오기
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f'<User {self.username}>'

# 리포지토리 및 서비스 초기화 (의존성 주입)
# 실제 앱의 경우 Flask-Injector 또는 유사한 패턴 사용
user_repository = UserRepository(db)
user_service = UserService(user_repository)

@app.before_first_request
def create_tables():
    db.create_all()

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    if not data:
        return jsonify({"message": "Invalid JSON"}), 400

    try:
        new_user_data = user_service.create_user(data)
        return jsonify({"message": "User created successfully", "user": new_user_data}), 201
    except ValueError as e:
        return jsonify({"message": str(e)}), 400 # 충돌 오류의 경우 409

@app.route('/users', methods=['GET'])
def get_all_users():
    users = user_service.get_all_users()
    return jsonify({"users": users}), 200

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = user_service.get_user_by_id(user_id)
    if user:
        return jsonify({"user": user}), 200
    return jsonify({"message": "User not found"}), 404

if __name__ == '__main__':
    with app.app_context(): # 요청 컨텍스트 외부에서 db.create_all()에 필요
        create_tables()
    app.run(debug=True)

이제 create_user 라우트는 훨씬 깔끔해졌습니다. HTTP 요청 및 응답 흐름을 처리하는 데만 집중합니다. 사용자 생성과 관련된 모든 입력 유효성 검사 및 비즈니스 로직은 UserService에서 처리되며, 데이터베이스 상호 작용은 UserRepository에 위임됩니다.

애플리케이션 시나리오

이 아키텍처 패턴은 다음을 위해 매우 유익합니다.

• 중대형 애플리케이션: 복잡성이 증가함에 따라 명확한 구분이 스파게티 코드를 방지합니다.
• 시스템의 다른 부분을 작업하는 팀: 개발자는 비즈니스 로직(서비스) 또는 데이터 액세스(리포지토리)를 독립적으로 작업할 수 있습니다.
• 높은 테스트 용이성이 필요한 애플리케이션: 각 계층은 의존성에 대한 목을 사용하여 격리하여 테스트할 수 있습니다.
• 데이터 소스를 변경할 수 있는 애플리케이션: 예를 들어 SQL에서 NoSQL로 변경하는 경우 서비스 및 컨트롤러는 그대로 두고 리포지토리 계층만 수정하면 됩니다.

결론

서비스 및 리포지토리 계층을 도입하기 위해 Flask 애플리케이션을 리팩터링함으로써 "뚱뚱하고" 다루기 어려운 라우트 핸들러를 간결하고 집중적이며 유지보수 가능한 구성 요소로 변환합니다. 이 아키텍처 규율은 관심사를 분리하여 가독성을 향상시키고, 테스트 용이성을 증대시키며, 성장하는 모든 Python 웹 프로젝트를 위한 확장 가능한 기반을 마련합니다.

이는 장기적인 개발 효율성과 시스템 복원력에 배당금을 지급하는 깔끔한 코드베이스에 대한 투자입니다.