Clean Architecture

Clean Architecture

회사에서 FastAPI로 자사의 서비스를 구성하면서 아키텍처에 흥미가 갔다.

 

따라서 현재 회사의 구조를 클린아키텍처로 바꿔봐야겠다는 생각이 들었다. 왜냐하면 결제와 주문 부분의 비즈니스 로직이 여러개의 도메인에 묶여있었기 때문에 유지보수와 에러가 발생하는 경우 코드를 파악하기 힘들었고, 내가 작성한 FastAPI코드는 전체적으로 객체지향 개발이 아닌 약간의 스크립트형식의 코드였기 때문이다. 또한 테스트 코드를 작성해보려고 하니, 비즈니스 로직이 너무 길고 다양한 외부 모듈에 의존을 하고 있어 테스트하기 어려웠다.

 

물론 자바와 스프링부트로 자사의 레거시 코드를 마이그레이션 한 적은 있지만, FastAPI를 처음부터 설계할 때는 생각보다 모르는 것들이 많았기 때문에 빠른 기능 구현을 목적으로 개발을 했고, 현재 실 서비스가 운영되는 상태에서 이제는 유지보수의 단계와 비즈니스 로직 중심으로 유지보수와 확장성을 확보하기 위해서 클린아키텍처를 공부했다.

 

또한 운영이 될 정도의 시스템을 만들어 놓고 보니, API 테스트 코드의 중요성을 많이 느끼게 되어 테스트 코드를 작성하려면 의존성이 적은 형태의 아키텍처가 완성이 되어야 하기 때문이다.

 

---

클린 아키텍처의 핵심 목표는 비즈니스 로직을 외부로부터 보호를 하기위한 이유가 있고, 각 계층별 독립적인 테스트가 가능하다. 왜냐하면 컨테이너로부터 의존성을 주입받기 때문에 Mock Container를 만들어서 의존성을 가짜로 주입하면 외부 의존성을 고려하지 않고 비즈니스 로직에 집중한 테스트코드 작성이 가능하다.

또한 Infra 계층에서 외부 라이브러리를 사용해서 언제든지 교체가 가능하다. 마지막으로 개인적으로 중요하게 생각하는 의존성의 방향이 내부로만 향하도록 되어 있다.

Inteface 계층

• API, Controller, Web 의 영역
• 사용자 요청을 Application Layer로 전달한다.
• 개인적으로는 Interface 계층에서 DTO 변환 책임을 부여하도록 하는 것이 좋은 방법이라고 생각한다.
  • 왜냐하면 Interface 계층 이후 Application 계층에서는 오로지 비즈니스 로직에만 집중을 해야하기 때문에 비즈니스 로직 이후 DTO 변환 작업까지 책임을 지면 테스트코드와 유지보수 측면에서 많이 좋지 않다고 생각하기 때문이다.

Application 계층

• Use-Case 정의
• 도메인 엔티티를 조합해서 동작을 수행
• 도메인 계층에만 의존 가능
• 인터페이스로 리포지토리 등을 정의 - 구현은 Infra 계층에서 구체화
• Application 계층에서는 Domain을 가지고만 비즈니스 로직이 수행되어야 한다.
  • 왜냐하면 비즈니스 규칙 보호와 의존성 방향 제어 때문이다.
  • Application 계층은 Domain(VO/Entity)를 사용해야 내부 비즈니스 로지이 외부 입출력 형태(DTO)에 종속되지 않고 테스트 가능하며, 확장성과 유지보수성이 높아지기 때문이다.
  • 이런 이유 때문에 Domain 계층에만 의존이 가능하다는 것이다.
  • 따라서 return 형태 또한 Domain 형태로 변경을 하고 해당 Domain을 Interface 계층에서 DTO로 변환하는 것이다.

Domain 계층

• 시스템의 핵심 비즈니스 로직과 규칙이 위치한 곳이다.
• 의존성이 없다. = 가장 안쪽에 위치한다.
• 인프라 -> 애플리케이션 -> 도메인 순서로 구성되어 있고, 인프라 계층의 경우 구현체에 불과하기 때문에 Domain 계층에서 Infra 계층은 절대로 알 수 없다. 따라서 Domain 계층은 의존하고 있는 것이 없어야 한다.

Infra 계층

• DB 연동, 외부 API 호출, 파일 시스템, 이메일 전송 등등
• 인터페이스 계층이나 애플리케이션 계층에서 정의한 추상화(인터페이스)의 실제 구현체 제공
• 기술 세부사항 구현
• 즉, 이 곳에서는 외부 라이브러리를 연동하거나 추상화한 것을 구체화하는 부분이다.

---

폴더 구조

/project
│
├── domain/
│   └── user/
│       ├── entity.py
│       ├── repository.py (interface)
│
├── application/
│   └── user/
│       ├── service.py
│       ├── dto.py
│
├── interface/
│   └── api/
│       └── user_controller.py
│
├── infrastructure/
│   └── user/
│       └── supabase_user_repository.py
│
├── main.py

예시 코드

project/
├── domain/
│   └── user.py                      # User Entity
│   └── user_repository.py          # IUserRepository (인터페이스)
├── application/
│   └── user_service.py             # Application Service (IUserRepository 의존)
├── infrastructure/
│   └── user_repository_memory.py   # 메모리 기반 레포지토리 구현체
├── dto/
│   └── user_dto.py                 # DTO 정의 및 변환
├── interface/
│   └── api/
│       └── user_controller.py      # FastAPI Router

---

• Interface 계층

from fastapi import APIRouter
from dto.user_dto import UserRequestDto, UserResponseDto
from application.user_service import UserService
from infrastructure.user_repository_memory import InMemoryUserRepository

router = APIRouter()

# 의존성 주입 수동으로 구성 (예시)
user_repo = InMemoryUserRepository()
user_service = UserService(user_repo)

@router.post("/users", response_model=UserResponseDto)
def create_user(request: UserRequestDto):
    user = request.to_domain()
    saved = user_service.register_user(user)
    return UserResponseDto.from_domain(saved)

@router.get("/users", response_model=list[UserResponseDto])
def get_all_users():
    users = user_service.list_users()
    return [UserResponseDto.from_domain(u) for u in users]

• Application Layer

from domain.user import User
from domain.user_repository import IUserRepository

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

    def register_user(self, user: User) -> User:
        if "@" not in user.email:
            raise ValueError("Invalid email")
        return self.user_repo.save(user)

    def list_users(self):
        return self.user_repo.find_all()

• Domain Layer

class User:
    def __init__(self, name: str, email: str):
        self.name = name
        self.email = email
        self.is_active = True

    def deactivate(self):
        self.is_active = False


######################################################

from abc import ABC, abstractmethod
from typing import List
from domain.user import User

class IUserRepository(ABC):

    @abstractmethod
    def save(self, user: User) -> User:
        pass

    @abstractmethod
    def find_all(self) -> List[User]:
        pass

• Infra Layer

from typing import List
from domain.user import User
from domain.user_repository import IUserRepository

class InMemoryUserRepository(IUserRepository):
    def __init__(self):
        self.users: List[User] = []

    def save(self, user: User) -> User:
        self.users.append(user)
        return user

    def find_all(self) -> List[User]:
        return self.users