Python과 FastAPI로 지속 성장 가능한 웹서비스 개발하기 - 3. Applications, Testing, Code Style

March 10, 2024

python

이전 글에 이어서 작성된 글입니다.

Test

테스트 커버리지는 잘 체크하지 않지만, 중요한 로직에 대한 테스트 코드는 최대한 챙기려고 노력하는 편이다. 본인은 개발하기 전에 꼭 지켜야하는 몇가지 개발 원칙이 있는데,

목표를 정확히 기술합니다. 명확한 이정표가 없다면 소중한 자원을 낭비하게 될 것입니다.
작업을 분리합니다. 하위 작업 역시 목표를 정확하게 기술합니다.

1과 2가 되어야 정확한 요구사항이 정해질 수 있고, 정확한 요구사항이 정해져야 유의미한 테스트 코드가 나올 수 있다고 생각한다. 미처 생각하지 못한 오류의 발생은 테스트 코드를 무조건 쓴다고 해서 잡을 수 있는게 아니라, 정확한 요구사항 정리가 되어야 잡을 수 있는 것이라 생각한다.

Mocking

지난 글을 통해 설명했듯이, 모든 컴포넌트들이 Dependency Injector로 Explicit하게 선언된 IoC 컨테이너를 통해 관리되고 있다. 즉 테스트하고자 하는 컴포넌트(sut; system under test)가 어떤 의존을 갖고 있던, IoC 컨테이너의 오버라이딩 기능을 통해 쉽게 테스트 대역을 주입할 수 있다.

# conftest.py
@pytest.fixture(scope="session")
async def test_container():
    container = TestContainer()
    yield container

pytest의 fixture 중 test container를 세션 스코프로 사용한다. 테스트 코드는 아래와 같이 작성할 수 있다.

@pytest.mark.asyncio
async def test_create_product(test_container, product_service):
    # arrange
    product_repository = Mock(spec=ProductRepository)
    product_repository.return_value = Product(id=1, name="Apple")

    # act
    with test_container.override_providers(product_repository=product_repository):
        product = await product_service.create_product(name="Apple")

    # assert
    assert product.name = "Apple"

간단하게 모킹이 가능하다. 여러 의존성을 같이 주입할 수도 있다.

# act
with test_container.override_providers(
    product_repository=product_repository,
    product_option_repository=product_option_repository,
    ...
):
    product = await product_service.create_product(**kwargs)

이런 방법으로 테스트 더블을 쉽게 주입해서 테스트할 수 있다.

참고로, 테스트 더블이란 테스트의 복잡성을 줄이기 위해 프로덕션 객체들처럼 행동하는 Simplified 버전의 대체할 수 있는 객체를 의미한다.

Test-double

거의 20년이나 되가는 Martin Fowler의 글을 발췌해보면…

Dummy objects are passed around but never actually used. Usually they are just used to fill parameter lists.
Fake objects actually have working implementations, but usually take some shortcut which makes them not suitable for production (an InMemoryTestDatabase is a good example).
Stubs provide canned answers to calls made during the test, usually not responding at all to anything outside what’s programmed in for the test.
Spies are stubs that also record some information based on how they were called. One form of this might be an email service that records how many messages it was sent.
Mocks are pre-programmed with expectations which form a specification of the calls they are expected to receive. They can throw an exception if they receive a call they don’t expect and are checked during verification to ensure they got all the calls they were expecting.

적절히 # arrange 영역에서 테스트 더블을 잘 구현해 override_providers 컨텍스트 매니저에 주입해주면 쉽게 테스트할 수 있다.

pytest fixture 관리

간단하게 IoC 컨테이너에 있는 의존성들은 pytest의 fixture로 불러올 수 있게 구현했다. dependency injector 라이브러리에서는 String identifier로 컴포넌트를 지정하여 가져올 수 있게 되어있다.

# conftest.py
@pytest.fixture(autouse=True)
def inject_components(test_container, request) -> None:
    """
    Components (services, repositories) 를 테스트의 fixture로 주입합니다.

    String Identifier 의 룰을 그대로 적용합니다, 다만 변수 이름으는로 `.`을 사용할 수 없기 때문에, `__`로 대체합니다.

    :param test_app: 테스트 앱 fixture
    :param request: pytest request fixture
    :return: None
    """
    item = request._pyfuncitem  # noqa
    fixture_names = getattr(item, "fixturenames", request.fixturenames)
    for arg_name in fixture_names:
        try:
            request.getfixturevalue(arg_name)
        except FixtureLookupError as e:
            if arg_name == "inject_components":
                continue
            provided = test_container
            for seg in arg_name.split("__"):
                try:
                    provided = getattr(provided, seg)()
                except AttributeError:
                    raise e
            item.funcargs[arg_name] = provided

이렇게 구현이 되어있다면, 쉽게 테스트할 대상을 아래처럼 갖고올 수 있다.

@pytest.mark.asyncio
async def test_some_service_some_method(some_service):  # <- `some_service`
    ...

hypothesis

Hypothesis라는 라이브러리를 사용하고 있다. 모든 테스트에 사용하고 있지는 않고, 더 잘쓰고싶은 라이브러리다. 간단하게 예시를 소개하자면,

@given(
    order_items=strategies.lists(
        strategies.builds(
            OrderItem,
            product=strategies.builds(
                Product,
                price=strategies.integers(min_value=100, max_value=10000),
                name=strategies.text(),
            )
        ),
        min_size=2,
    )
)
async def test_calculate_total_price(checkout_service: CheckoutService, products: list[Product]):
    # act
    total_price = chekout_service.calculate_total_price(products=products)

위와 같이 테스트 코드를 구현하게 되면, 많은 테스트 셋으로 해당 테스트를 실행하도록 자동화되게 된다. 파라미터로 인한 테스트 실패를 발견할 수도 있어서 꽤 좋은 테스트 도구라 생각한다.

Assembly Applications

이제 어플리케이션을 구성해보자. 모든 컴포넌트들이 잘 모듈화되어있다면 어떤 어플리케이션 레이어를 사용하더라도 쉽게 붙일 수 있는 구조를 갖고 있을 것이다. Layered Architecture

HTTP API (FastAPI)

def create_app() -> FastAPI:
    app = FastAPI(
        title="your-app",
        middleware=create_middleware(stage=application_settings.stage),
        openapi_url=None if application_settings.stage == "prod" else "/api/openapi.json",
        docs_url=None if application_settings.stage == "prod" else "/api/docs",
        redoc_url=None,
    )
    app.container = MainContainer()

    app.add_event_handler("startup", redis_async_pool_manager.init_redis_pool)
    app.add_event_handler("shutdown", redis_async_pool_manager.close_redis_pool)

기본적으로 위와 같은 factory 형태를 취한다. gunicorn.conf.py에서는 이렇게 할 수 있다.

workers = 3
worker_class = "uvicorn.workers.UvicornWorker"
timeout = 30  # default = 30
graceful_timeout = 30  # default = 30
threads = 1  # gunicorn with UvicornWorker doesn't affect.
wsgi_app = "application:create_app()"
pythonpath = "app"

몇 가지 생략되었지만, FastAPI 문서까지 잘 참고하기 바란다.

CLI or Batch (Typer)

개발자를 위한 도구나, 운영을 위한 도구들을 CLI로 만들어두는 경우가 있다. FastAPI의 개발자 tiangolo가 만들어둔 Typer라는 도구는 FastAPI의 CLI 도구로 사용할 수 있도록 구현되어 있다. 하지만, 이미 asyncio 라이브러리들을 통해 구현된 의존성을 실행시키기 위해 비지니스 로직 또한 코루틴으로 작성되어있을 확률이 매우 높고 Typer는 코루틴을 실행하기엔 약간 어려움이 있다.

app = typer.Typer()


@app.command()
def hello(name: str):
    print(f"Hello {name}")

커맨드를 정의하는 @app.command()는 synchronous function만 지원하기 때문이다. 간단한 래핑을 통해 코루틴을 실행시키는 환경, async-typer를 만들어뒀다.

app = AsyncTyper()

app.add_event_handler("startup", redis_async_pool_manager.init_redis_pool)
app.add_event_handler("shutdown", redis_async_pool_manager.close_redis_pool)

@app.async_command()
async def foo():
    await service.work_async()

더불어 FastAPI와 같은 환경을 구현하려면 FastAPI의 add_event_handler와 같은 역할을 하는 기능도 필요할 것 같기에 구현해두었다.

Code style

간단하게 코드 스타일 관련 도구들을 소개한다.

isort

코드, 의존성의 설계가 잘못된 상황이라면 임포트문에서부터 잘못됨을 느낄 수 있다고 믿고 있다. 그렇기 때문에 임포트문을 잘 정리해주는 도구인 isort는 절대로 없어서는 안되는 도구라고 생각한다. pyproject.toml 레퍼런스 찾는 것도 귀찮은 일이기에 글에 정리해둔다.

[tool.isort]
src_paths = ["app"]
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 0
use_parentheses = true
float_to_top = true
line_length = 120
ensure_newline_before_comments = true
sections = ["FUTURE", "STDLIB", "THIRDPARTY", "FIRSTPARTY", "LOCALFOLDER"]
force_alphabetical_sort_within_sections = true
skip_glob = ["alembic"]

ruff

rust로 구현된 엄청 빠른 파이썬 린터, 포매터다. Github Repository의 README를 참고하자면, flake8보다 100배는 빠르다고 하고 있다. 파이썬 라이브러리들이 요즘은 rust로 내부 구현을 하는게 대세인가 싶지만, 아무튼 같은 결과를 빨리 내주는게 최고라고 생각하기에.. 아무튼, 별 문제 없이 잘 쓰고 있다. 참고로, Astral이 제공해주는 uv도 유심히 눈여겨보고 있다. 파이썬 생태계가 계속 발전하고 있음에 안도감을 느낀다.

Black

늘 서오던 포매터이다. 작년부터 PyCharm에 기본으로 통합할 수 있다. ruff가 좀 더 괜찮다고 생각할 땐 안 쓸 수도 있을 것 같다.

읽어보면 좋을 글

Unit Test