Python으로 클린 아키텍처 적용하기

Jepeto·2020년 4월 20일

Implementing the Clean Architecture

📝 Implementing the Clean Architecture in Python 을 읽고 정리한 글입니다.
소스코드: https://github.com/Enforcer/clean-architecture
작성 일자: 2020-04-21

도메인 주도 설계 (DDD)가 필요하게 된 이유

1. CRUD 방식(일반적 설계 모델)의 문제점:

Invariant Span(생명 주기)가 single model을 넘어서는 경우, 로직이 복잡해짐.

code가 서로 강하게 결합(tightly coupled) 되어있는 경우, business logic을 테스트 하기 위해서 미리 많은 작업을 셋팅해야하는 경우가 생긴다. ex) database 불러오기, web framework code 실행 시키기 등등
=> 이는 테스트 시간을 증가 시키고, 테스트 코드 작성의 복잡성을 야기시킨다.

3rd party services와의 결합이 어려워짐. ex) 어떤 결제 모듈과 처음에 코드를 결합 시켰을 때, 강한 결합으로 인해, 다른 결제 모듈로 대체해야하는 경우 수정하기 어려워진다.

2. Clean Architecture의 장점:

1. Framework의 독립성: 프레임워크를 업그레이드 하거나, 교체하기 쉬워짐.
2. Testability: 모든 비즈니스 로직을 unit test할 수 있음.
3. Independance of UI/API
4. Independance of Database: 데이터를 다양한 방식으로 저장할 수 있음.
5. Independance of any 3rd party: eg) 비즈니스 로직은 어떤 결제 모듈을 사용하는지 알 필요가 없음
6. Flexibility
7. Extensibility: 쉽게 확장 가능해짐. CQRS, Event-Sourcing, DDD방식을 필요하다면 적용가능.

Layers of Clean Architecture

1. External World: code 바깥쪽 외부 세계. DB, API, Network..
2. Infrastructure: DB, API와 같은 것들에 대한 adapter역할.
3. Application:
1) Use Case: Single Operation(하나의 비즈니스 로직 - user의 하나의 action)으로 구성됨. ex) 입찰하기 , 입찰 취소하기
2) Interface: 추상화된 use case, infrastructure을 추상화
# application/interfaces/email_sender.py
import abc
class EmailSender(abc.ABC):
       @abc.abstractmethod
       def send(self, message: EmailMessage) -> None:
           pass
# infrastructure/adapters/email_sender.py
import smtplib
from application.interfaces.email_sender import EmailSender
class LocalhostEmailSender(EmailSender):
    def send(self, message: EmailMessage) -> None:
        server = smtplib.SMTP('localhost', 1025)
        # etc
4. Domain: identity를 가진 Entity로 구성됨. 비즈니스룰을 담고 있음.

Layers of Clean Architecture

External World -> Infrastructure -> Application -> Domain

Dependency Rule: 화살표 방향으로 의존성을 가지고 있음.
ex) Application은 Domain을 사용할 수 있어도, Domain은 Application을 사용 못함.
ex) infrastructure은 Application을 사용할 수 있어도, Application은 Infrastructure을 사용 못함.
Business Rule과 Processes 들은 Application layer와 Domain layer상에서 이루어짐. External World를 전혀 모른채, 쉽게 Business Logic을 테스트 코드 작성할 수 있음.
Boudary는 set of Interfaces (여러개의 Interfaces) 들로 정의됨. 모든 detail은 boundary 뒤에 숨어있음.
Input DTO, Output DTO, Input Boundary 는 Application Layer안에 Detail을 숨기는 Boundary를 형성함.
- Input DTO: Boundary에 진입할 수 있는 input argument(Request)
- Output DTO: (Response)
- Input Boundary: Use Cases를 추상한 Interface

@dataclass
class PlacingBidInputDto:
    bidder_id: int
    auction_id: int
    amount: Decimal   
    
@dataclass
class PlacingBidOutputDto:
    is_winning: bool
    current_price: Decimal                                   
    
class PlacingBidInputBoundary:
    @abc.abstractmethod
    def execute(self, request: PlacingBidInputDto) -> None:

주의할 Anti Pattern: anemic entities이 안되게 조심할 것!. entity 안에 entity 관련 method를 만들기(올바른 방법). entity 밖의 method로 entity값을 변경하는 것(잘못된 방법)은 금물.
ex) entity안에 있는 관련된 method를 통해 entity 데이터를 조작.
class PlacingBidUseCase:
    def execute(self, ...):
       # Tell, don't ask violated
       # if auction.current_price < new_bid.amount:
       #    auction.winners = [new_bid.bidder_id]
       #    auction.current_price = new_bid.amount
       # let Auction handle it! It knows best
         auction.place_bid(...)

Control Flow in Architecture

control flow in the clean architecture

Controller은 HTTTP request data를 repack해서 Input DTO로 보내고, Input DTO를 Input Boundary에 input으로 전달됨. 이때, Input Boundary를 Use Case의 Interface이다.
Input Boundary는 Input DTO를 이용하여 Input DTO의 데이터를 사용하여 필요한 Entities를 Data Access Interface를 이용하여 Database로부터 fetch함.
Entites가 Business Logic을 수행함. Data Access Interface를 통해 변경사항을 저장함.
Use Case는 Output DTO를 return함. Presenter의 Interface 역할을 하는 Output Boundary에 Output DTO를 전달함. 이때, Presenter의 역할은 Output DTO의 data를 reformat하여 View에 최종적으로 보여주는 역할을 함.

Input DTO, Input Boundary, Output Boundary, Output DTO, Use Case, Data Access Interface 는 Application layer에 속함.
Entities는 Domain Layer에 속함.

ex) 경매(Auction) 을 예로 들어 보자.

Business Requriements:

Bidders can place bids on auctions to win them

An auction has a current price that is visible for all bidders

current price is determined by the amount of the lowest winning bid

to become a winner, one has to offer a price higher than the current price

Auction has a starting price. New bids with an amount lower than the starting price must not be accepted

Sequence Diagram


# Input Boundary로 전달할 Input DTO
@dataclass
class PlacingBidInputDto:
    bidder_id: int
    auction_id: int
    amount: Decimal
 
 # Use Case를 Abstract한 Input Boundary
class PlacingBidInputBoundary(abc.ABC):
    @abc.abstractmethod
    def execute(self, input_dto: PlacingBidInputDto, 
    		presenter: PlacingBidOutputBoundary) -> None:
	pass

# Use Case: 모든 비즈니스 로직이 담김.
# 중요한 점: 오직 Higher Layer(Interface)들과 코드를 구성하고 결합함. 
# 뒷단에서 이루어지는 concrete Implementation은 모름.(Dependency Injection을 통해 Mapping 해줬기 때문에 몰라도됨)
class PlacingBidUseCase(PlacingBidInputBoundary):
"""
1. Auction entity를 data access를 통해 retrieve
2. entity의 place bid 메소드 호출 (비즈니스 로직 수행)
3. Command 역할: 업데이트 상태변경 및 저장, return 값 없는 것이 특징
4. Output DTO 생성
5. Output DTO를 Output Boundary에 전달
"""
    def __init__(self, data_access: AuctionsDataAccess, output_boundary:
       PlacingBidOutputBoundary) -> None:
        self._data_access = data_access
        self._output_boundary = output_boundary
        
    def execute(self, input_dto: PlacingBidInputDto) -> None:
        auction = self._data_access.get_auction(input_dto.auction_id)
        auction.place_bid(input_dto.bidder_id, input_dto.amount)
        self._data_access.save_auction(auction)
        output_dto = PlacingBidOutputDto(
            input_dto.bidder_id in auction.winners, auction.current_price
        )
        self._output_boundary.present(output_dto)

@dataclass
class PlacingBidOutputDto:
    is_winning: bool
	current_price: Decimal

# Presenter를 Abstarct한 Output Boundary
class PlacingBidOutputBoundary(abc.ABC):
    @abc.abstractmethod
    def present(self, output_dto: PlacingBidOutputDto) -> None:
    	pass
        
    @abc.abstractmethod
    def get_presented_data(self) -> dict:
    	pass

# Presenter: 최종적으로 view에 보여줄 data를 reformat 시키는 역할
class PlacingBidWebPresenter(PlacingBidOutputBoundary):
    def present(self, output_dto: PlacingBidOutputDto) -> None:
        self._formatted_data = {
            'current_price': f'${output_dto.current_price.quantize(".01")}',
            'is_winning': 'Congratulations!' if output_dto.is_winning else ':('
        }
        
    def get_presented_data(self) -> dict:
        return self._formatted_data

import inject

# Dependency Rule을 만들어줌: Abstract class 와 특정 Implementation을 연결해줌.(Mapping 역할) 
# 이를 통해 Abstract class를 사용할때, 이 특정 Implementation이 사용 되도록하는 역할을 함.
def di_config(binder: inject.Binder) -> None:
    binder.bind(AuctionsDataAccess, DbAuctionsDataAccess())
    binder.bind_to_provider(PlacingBidOutputBoundary, PlacingBidWebPresenter)

inject.configure(di_config)

# DataAccess: Database의 Interface 역할함.
class AuctionsDataAccess(abc.ABC):
    @abc.abstractmethod
    def get(self, auction_id: int) -> None:
        pass
        
    @abc.abstractmethod
    def save(self, auction: Auction) -> None:
    	pass

# Entities - Bid
@dataclass
class Bid:
    id: Optional[int]
    bidder_id: int
    amount: Decimal

# Entities - Auction
class Auction:
    def __init__(self, id: int, starting_price: Decimal, bids: typing.List[Bid]):
       self.id = id
       self.starting_price = starting_price
       self.bids = bids
       
    def place_bid(self, user_id: int, amount: Decimal) -> None:
       pass
       
    @property
    def current_price(self) -> Decimal:
       pass
       
    @property
    def winners(self) -> typing.List[int]:
       pass

Clean Architecture 변형할 수 있는 형태 소개

1. Presenter에서 get 함수 사용.

mobile 용도가 아닌 web framework 사용시 Presenter가 View에 return 값을 전달하도록 변경 할 수 있다.
ex)

변경 전

def index(request: HttpRequest) -> HttpResponse:
       # Django will not forgive you if you do not return HttpResponse from a view
       return HttpResponse(f'Hello, world!')

변경 후

# presenter에 return 시킬 수 있는 get 함수를 만듬.
def index(request: HttpRequest) -> HttpResponse:
       ...
       return presenter.get_html_response()

2. CQRS 방식

Output DTO와 Output Boundary(&& Presenter)를 사용하지않고, CQRS 방식으로 처리하는 방법이 있다.

Presenter를 사용하지 않고, Query를 사용하는 방식

3. Input Boundary를 없앨 수 있다.

Controller 와 Use Case가 coupled되는 것이 커다란 문제를 야기시키지는 않는다. 이렇게 해야할 이유가 있다면 해도 된다.

User Case에 대한 대안

1. Facade 패턴

User Case가 복잡하지 않은 형태라면 Facade 패턴을 사용하여 디자인할 수 있다.

ex) application은 호텔이라고 생각하면, reception에 말해서 모든 일을 처리해야지, 골프 치고 싶다고 골프 keepeer 찾고, spa를 하고 싶다고 spa treatment keepr를 찾는 것은 비효율.

Facade 패턴: 순차적인 메소드들을 single function안에 넣어서 구현.
get an Entity using Data Access Interface, call an Entity’s method, then save it back. 이와 같은 복잡하지 않은 형태의 경우 Facade 패턴을 사용하면 굿!

2. Mediator(Command Bus)

만약 Output DTO를 사용하지 않을 것이라면, Input DTO를 CQRS의 Commands로 대체하고, Command Bus라고 불리는 Mediator를 도입하는 것도 좋은 대안이다. (Input Boundary도 삭제)

이때, User Case는 Command Handlers 가 된다.
이때, Command는 DTO (왜냐하면 easy to serialize, message-driven architecture 설계하기에 좋음)

def placing_bid_view(...) -> None:
    command = PlaceBid(...)
    # command_bus passes command to appropriate handler
    command_bus.dispatch(command)

장점: handler로 부터 완전히 decouple 시킬 수 있다. 우리는 Command classes와 Command Bus만 알고 있으면 된다.

Dependency Injection

추상화 그것에 대응되는 어떤 Implementation을 mapping

Abstractions & classes everywhere!
losse coupling이란 Interface만 고려되는 것.

class CreditCardPaymentGateway(PaymentGateway):
    pass
    
# tight coupling (Implementation을 그대로 사용)
class Order:
    def finalise(self) -> None:
        payment_gateway = CreditCardPaymentGateway(
            settings.payment['url'], settings.payment['credentials']
        )
        payment_gateway.pay(self.total)
        
# loose coupling (payment_gateway로 Abstarct 한 것을 사용)
class Order:
    def __init__(self, payment_gateway: PaymentGateway) -> None:
        self._payment_gateway = PaymentGateway
    def finalise(self) -> None:
        self._payment_gateway.pay(self.total)

참고: Configuration을 통해 Dependency Injection을 관리한다면, if문 같은 것으로 코드를 더럽히지 않아도 됨. Inversion of Control Container 이 그 역할을 해줌.

CQRS

시스템의 상태를 변경하는 코드와 상태 변경 없이 data를 read하는 코드를 분리한다.

CQRS allows us to use normalized data for writing and denormalized data for querying with a mechanism to allow write data to keep updating the read data at regular intervals.
Event Sourcing 방식과 CQRS 방식의 궁합이 좋음.

Commands:

시스템의 상태를 변화시킴.

business requirements 변화에 영향을 받음.

Commands 실행 순서가 중요함.
ex) adding an item to a cart, removing delivery address

Queries:

simple하고 안전함(시스템의 상태를 변경하지 않음)
ex) displaying a list of available products, getting user`s delivery addresses

왼쪽: Commands, 오른쪽: Queries

1. Command

Commands는 DTO 역할을 하고, 이는 Command Handlers에 의해 실행된다. 개발자는 application에서 services들을 사용하기 위해 Command Handlers의 존재를 알 필요 없다. Use Case를 sending a Command로 대체해서 생각하면 된다.
=> Command(DTO) - Command Bus - Command Handler

2. Query

Query를 다루는 여러 방법
1. Query as DTO: Command를 다루는 방식과 똑같이 동작한다. 각각의 쿼리를 single class (DTO) 형태로 나타낸다. 이는 Query Handler에 의해 실행 된다. 개발자는 Query Handlers의 존재를 알 필요 없다. Query class는 Application Layer에 속하고, Query Handler는 Infrastructure layer에 속한다.
=> Query(DTO) - Query Bus - Query Handler
ex)

class GetListOfDeliveryAddresses(Query):
       Dto = List[DeliveryAddress]
       def __init__(self, user_id: int) -> None:
             self.user_id = user_id
       def query_handler(query: GetListOfDeliveryAddresses) ->
GetListOfDeliveryAddresses.Dto:
...
# using via QueryBus
query = GetListOfDeliveryAddresses(user_id=1)
result = query_bus.dispatch(query)

2. Queries as separate classes: Query Bus 나 Query Handler는 사용하지 않고, 각각의 Query는 abstract class로 Application Layer에 있고, concrete implementation은 Infrastructure Layer에 위치한다.
ex)

# somewhere in Application layer
class GettingListOfDeliveryAddresses(Query):
    Dto = List[DeliveryAddress]
    def __init__(self, user_id: int) -> None:
        self.user_id = user_id
    @abc.abstractmethod
    def execute(self) -> Dto:
        pass
# in Infrastructure layer
class SqlGettingListOfDeliveryAddresses(GettingListOfDeliveryAddresses):
    def execute(self) -> GettingListOfDeliveryAddresses.Dto:
        models = self.session.query(Address).filter(
            (Address.type == Address.DELIVERY) &
            (Address.user_id = self.user_id)
        )
        return [self._to_dto(model) for model in models]

이때 Dependency Injection을 통해 Abstract Class와 Concrete Implementation을 맵핑 시켜줘야 한다.
ex)

# dependency injection configuration
@inject(config=Config)
def configure(binder, config):
    binder.bind(
        GettingListOfDeliveryAddresses,
        to=SqlGettingListOfDeliveryAddresses,
)
# in web view
@app.route("/auction/bids", methods=["POST"])
def auction_bids(query: GettingListOfDeliveryAddresses) -> Response:
    result = query.execute()
    ...