Python과 FastAPI로 지속 성장 가능한 웹서비스 개발하기 - 1. 의존성 주입

새로운 커머스 서비스 프로젝트를 FastAPI로 개발하고 런칭 후 운영한지 1년 정도 된 시점이 되었다. 현업에서 FastAPI를 사용한지도 거의 3년이 되가고 있기도 하다. 팀 빌딩을 위해 컨퍼런스 발표 준비도 할 겸, 한국에서는 파이썬 생태계나 커뮤니티가 JVM 언어에 비해 많이 부족하다고 생각하여 작게나마 나의 경험담을 공유하고자 한다.

프로젝트 구조

나는 Layered Architecture 기반으로도 왠만한 복잡도 있는 요구사항을 충분히 개발하고 운영하는데 문제가 없을 거라 생각한다. 모든 구성원이 아래와 같은 원칙을 갖고 제품을 개발하도록 했다.

  1. 레이어는 위에서 아래로 순방향으로만 참조되어야 한다.
  2. 역류 참조를 절대 금지한다.
  3. 건너뛰는 참조도 금지한다.

적용하고 있는 아키텍쳐

의존성 관리

계층화된 모듈을 잘 관리하고 사용하는데 도움을 준 가장 큰 프레임워크는 Python 진영의 거의 유일한 라이브러리인 Dependency Injector 다. Spring 프레임워크만 보더라도 기본적으로 숨쉬듯 제공하는 것들이지만, 파이썬 생태계나 커뮤니티에서 마음에 드는 의존성 주입 프레임워크를 찾는 건 쉬운 일이 아니었다. 의존성 주입 프레임워크가 제공해주는 유연하고 안전한 개발 환경과 쉽고 빠르게 개발할 수 있는 파이썬 개발 환경이 적절히 잘 합쳐지면 상당한 임팩트가 있다고 생각한다.

의존성 주입을 통해 우리가 얻을 수 있는건 협업하는 모듈 간 커플링이 낮아지고, 높은 응집도가 생긴다는 점이다. 커플링이 높은 코드 또는 응집도가 낮은 코드는 덩치가 커졌을 때 분리해내기 매우 어렵다. 모듈 간 커플링이 낮고 높은 응집도를 갖게된다면, 일반적으로 코드의 단순화 및 유지보수의 더 높은 수준의 자유를 얻을 수 있다.

또, 테스트 가능성(Testability)을 얻을 수도 있다. 테스트 코드를 작성할 때, 실제 데이터베이스 모듈이 아닌 Mock을 사용하는 등, 테스트 대상 객체를 다른 의존성으로부터 격리시킬 수 있는 아주 쉬운 구조가 될 수 있다.

나는 이 Dependency Injector 라이브러리가 파이썬으로 웹 서비스를 개발하는 엔지니어들에게 더 많이 사용되길 바라고 있다. 장점을 소개하자면, IoC (Inversion of Control) 컨테이너를 훨씬 더 명시적(Explicit)으로 구성할 수 있다. 아래의 예시를 통해 확인해보자.

from dependency_injector.containers import DeclarativeContainer, WiringConfiguration
from dependency_injector.providers import Configuration, Container, Factory

from containers.core import CoreContainer
from containers.infra import InfraContainer
from containers.system import SystemContainer
from domains.owner.services.admin_service import AdminService


class MainContainer(DeclarativeContainer):
    # Configurations
    config = Configuration(yaml_files=["config.yml"])  # <-- 1
    wiring_config = WiringConfiguration(
        packages=[
            "applications",
            "contexts",
        ]  # <-- 2
    )
    core = Container(CoreContainer, config=config.core)
    system = Container(SystemContainer, config=config.system)
    infra = Container(InfraContainer, config=config.infra)

    admin_service = Factory(   # <-- 3
        AdminService,
        admin_repository=infra.admin_repository  # <-- 4
    )

하나의 컨테이너 클래스 정의 안에 어플리케이션 구조가 한 눈에 보이는 특징을 갖고 있다. 이 구현을 보고 아래와 같은 사실을 쉽게 이해할 수 있다.

  1. 이 IoC 컨테이너는 config.yml을 통해 설정값들이 관리되고 있다.
  2. 이 IoC 컨테이너에서 관리되는 모듈들은 applications, contexts 패키지들에 wire되고 있다.
  3. admin_service 가 domains.owner.services.admin_service.AdminService 를 통해 Factory 프로바이더를 통해 어플리케이션에 주입되고, 주입될 때마다 새로운 인스턴스가 만들어진다.
  4. admin_service 가 infra 컨테이너의 admin_repository를 의존하고 있다.

이렇게 명시적으로 정의된 컨테이너로 모듈 간의 의존성을 조감도를 보듯 쉽게 파악할 수 있기 때문에 어플리케이션을 확장하기 아주 쉽다. 순환 참조같은 설계의 실수도 쉽게 피할 수도 있다.

또 다른 장점으로, 적절한 전략으로 의존성을 주입할 수 있다는 점이다. 이 라이브러리는 다양한 Providers가 구현되어 있고, 다루고 있는 인프라, 도메인의 특성 또는 상황에 따라 다양한 전략을 취할 수 있다. 주로 사용하는 Providers 몇 가지만 소개한다. 그 외의 프로바이더들은 이 문서를 통해 확인할 수 있다.

providers.Factory

말그대로 Factory. 주입될 때마다 새로운 인스턴스를 생성한다.

providers.Singleton

첫 주입 때만 인스턴스를 만들고, 그 이후에는 컨테이너가 살아있는 동안 그 인스턴스를 계속 주입하게 된다.

providers.Resource

initialization과 shutdown이 필요한 경우 사용한다. 컨테이너에 Resource로 등록된 경우, 컨테이너 인스턴스의 .init_resources(), shutdown_resources()를 통해 리소스 관리가 가능한 싱글턴이다. FastAPI의 이벤트 핸들러와 사용되면 매우 편하다.

from fastapi import FastAPI

from containers import MainContainer

def create_app():
    app = FastAPI()
    app.container = MainContainer()

    app.add_event_handler("startup", container.init_resources)
    app.add_event_handler("shutdown", container.shutdown_resources)
    ...

위와 같이create_app()을 구성하게 되면, FastAPI 어플리케이션의 생애주기와 동기화되어 모든 어플리케이션 모듈이 관리될 수 있다. 이렇게 관리된 IoC 컨테이너는 API에서 아래와 같이 쓰인다. (응답 정의나 모델 변환 같은 마이너한 코드들은 생략)

from dependency_injector.wiring import inject, Provide
from fastapi import APIRouter, Depends

router = APIRouter()

@router.get("/list")
@inject
async def get_user_list(
    auth: Auth = Depends(auth_admin_user),
    user_service: UserService = Depends(Provide["user_service"])
):
    users = await user_service.find_users()
    return users

다시, 우리 프로젝트 구조로 돌아와서. 이 라이브러리를 통해 위 컴포넌트들을 어떻게 구현했는지 설명해보면,

적용하고 있는 아키텍쳐

Services: 비지니스 모델이 주 관심사인 컴포넌트다. 인프라(쿼리, aws, 네트워크 등…)의 관심사를 전혀 담지 않은, 오직 의존성의 인터페이스에만 의존하고 있는 순수 비지니스 모델을 구현할 수 있다.

from enums.exhibition import ExhibitionStatusEnum
from models.exhibition import Exhibition
from repositories.exceptions import DoesNotExistException, IntegrityException
from repositories.exhibition import ExhibitionRepository
from repositories.product import ProductRepository
from services.exceptions import BadRequestException
from services.exhibition.commands import CreateExhibitionCommand, UpdateExhibitionCommand
from services.exhibition.exceptions import ExhibitionDoesNotExist


class ExhibitionService:
    exhibition_repository: ExhibitionRepository
    image_repository: ImageRepository
    product_repository: ProductRepository

    def __init__(
        self,
        exhibition_repository: ExhibitionRepository,
        product_repository: ProductRepository,
    ):
        self.exhibition_repository = exhibition_repository
        self.product_repository = product_repository

    ...

실제 운영 중인 코드의 일부를 갖고 왔다. 관심사의 분리가 잘못 되었다면 코드 파일의 임포트 문부터 잘못됐음을 인지할 수 있다. 해당 프로젝트에서 사용했던 나중에 소개할, SQLAlchemy (MySQL), redis-py, boto3 등의 라이브러리들이 임포트문에 존재한다면, 관심사의 분리가 잘못되었다는 걸 이해해야 한다. 이 ExhibitionService가 의존하고 있는 ExhibitionRepository와 ProductRepository가 어떤 Database를 통해, 또는 어떤 API를 통해 데이터를 저장하는 지는 이 구현의 관심사가 아니다. 오직 이 두 Repository가 어떤 인터페이스를 갖고 있는지만 의존하고 있으면 충분하다. 느슨한 커플링이 잘 구현된 경우라 할 수 있다.

이 라이브러리는 django와도 쉽게 사용할 수 있다. 만약 비지니스 로직을 순수한 python 서비스 객체로 작성하고 싶은 의지가 있다면, django와의 예시도 함께 보면 좋을 것 같다.

Infra, externals, systems: Infra는 aws, redis, mysql 등의 인프라를 사용하기 위해 필요한 컴포넌트들이고, externals는 시스템 외부의 API등의 것을 위한, systems는 시스템 내부와의 통신을 위한 컴포넌트들이다.

from contextlib import asynccontextmanager

import aiohttp
from aiohttp import ClientSession

from external.delivery_tracker.schemas import TrackResponseSchema


class DeliveryTrackerAPI:
    @asynccontextmanager
    async def get_session(self):
        conn = aiohttp.TCPConnector(
            limit_per_host=5,
        )
        session = aiohttp.ClientSession(connector=conn)
        try:
            yield session
        finally:
            await session.close()

    async def track_delivery(
        self,
        courier_code: CourierCode,
        tracking_number: str
    ):
        url = f"https://example.io/carriers/{delivery.courier_code.value}/tracks/{delivery.tracking_number}"
        async with self.get_session().get(url) as resp:
            if 200 <= resp < 300:
                return TrackResponseSchema.model_validate(await resp.json())
            elif 400 <= resp < 500:
                raise TrackException(message=await resp.json()['message'])
            else:
                raise TrackOperationalException(extra=await resp.text())

배치 앱에서 사용되는 택배 추적 API 클래스를 가져와봤다. systems에 해당하는 컴포넌트다. 서비스 레이어에서는 DeliveryTrackerAPI 의 인터페이스만을 의존해야하고, 이 API 클래스가 aiohttp 라이브러리로 구현되었는지는 관심이 없다. 그러므로 예외 처리까지 생각해본다면, aiohttp가 발생하는 예외를 바로 상위 레이어에 노출하지 않고, 직접 정의한 예외 클래스를 사용하는게 좋다. TrackException, TrackOperationalException을 정의하고 상위 레이어에서 다시 핸들링하도록 한다.

다음 글로 이어집니다.

읽어보면 좋을 글