[FastAPI] Depands와 Dependency-injector DI 비교해보기

목차

 

개요 

최근에는 SQLAlchemy의 session을 이용해 프로젝트 구조를 잡는 편이다. FastAPI가 구동되는 단계에서 DI를 이용해 전역적으로 사용할 Session을 설정하는 방식으로 구성하는데 크게 3가지 정도의 방법을 이용해 Session DI를 구성할 수 있었다. SQLAlchemy의 Session으로 DI를 사용할 경우 어떤 식으로 DI를 시도했는지에 대한 내용 정리와 각 구성별 LoadTest 결과를 기록하려 한다.

 

테스트 셋업

FastAPI에서는 Too Many Open File 이슈가 발생할 수 있다. ulimit 설정은 soft 설정을 5012 정도로 잡았다.

locust를 이용해 테스트를 진행했으며 RPS 가 어느 경우에 더 높은 지를 보는 게 목적이다. locust는 워커를 3개로 하고 다음과 같이 설정했다.

 

프로젝트 

전체적인 프로젝트 구조는 이 링크에서 참고할 수 있다.

 

프로젝트 구조 참고

프로젝트 구조는 계층(layerd)으로 잡았다. 여기서는 간단히 테스트만 하기 때문에 api(controller)가 바로 repository를 호출하는 방식으로 테스트했다. 덧붙여 SQLAlchemy의 Session을 이용할 것이기 때문에 Repository의 구성을 추상화(abstract) 시켰다.

from typing import Protocol, Generic, TypeVar, List

Entity = TypeVar("Entity")

class IRepository(Generic[Entity], Protocol):
     session = None # DI를 통해 사용하게될 SQLAlchemy Session 후술되는 내용에서는 조금씩 설정방법이 다름

class Repository(IRepository):

    def get_session(self):
        self.session.execute("select 1;")

위와 같은 Repository 구성을 api(controller)에서는 다음과 같이 호출한다.

@api_router.post(path="/session/example")
def session_example():
    repo = Repository()
    repo.get_session()

    return JSONResponse(status_code=200, content={})

이후 uvicorn을 이용해 worker를 4개 정도로 주어 실행시킨 뒤 locust를 이용했다.

uvicorn application:demoapplication --workers 4

 

3. DI 테스트

3.1 FastAPI의 DI

먼저 FastAPI의 DI를 이용해 보자. FastAPI에서는 built-in으로 Depands라는 함수 제공된다. 이 Depands 함수를 이용해 SQLALchemy Session을 이용할 때는 다음과 같이 사용한다.

# pseudo
def get_db():
    db = SQLAlchemySession() 
    try:
        yield db
    finally:
        db.close()

@app.post("/", )
def create_user(db: Session = Depends(get_db)): ...

위의 코드를 참조해 다음과 같이 구성했다.

db_engine = engine.create_engine(
    DevDataBaseConnection.get_url(),
    pool_pre_ping=True,
    pool_recycle=3600,
    pool_size=5,
    max_overflow=5,
    pool_timeout=10,
    echo=True
)

def get_db():
    _session = scoped_session(sessionmaker(
        bind=db_engine,
        expire_on_commit=False,
        autocommit=False,
        autoflush=False
    ))
    try:
        yield _session
    finally:
        _session.commit()
        _session.close()

# api
@api_router.post(path="/session/example02")
async def repository_session_example2(session=Depends(get_db)):
    session.execute("select 1;")
    session.commit()
    session.close()
    return JSONResponse(status_code=200, content={})

위 코드를 locust를 이용해 테스트한 결과 RPS는 다음과 같이 나온다.

 

3.2 dependency-injector의 Factory Provider

이 방법은 애용하는 방법인데 프로젝트에 아키텍처적인 부분을 감안할 때 사용해 먹는 편이다. dependency-injector의 container를 이용해 SQLALchemy의 engine과 Session을 미리 초기화해 두고 FastAPI가 실행될 때 IRepsitory의 Session에 DI 시키는 방식이다. 문장으로 써보자니 조금 어렵게 느껴지는데 코드 자체는 다음과 같이 구성할 수 있다.

from dependency_injector import containers, providers

from settings.dev import get_session, get_engine

class DataBaseContainer(containers.DeclarativeContainer):
    _engine = providers.Singleton(get_engine)

    session = providers.Singleton(get_session, sa_engine=_engine.provided)

from typing import Protocol, Generic, TypeVar, List
from dependency_injector.wiring import Provide
from settings.dependency import DataBaseContainer

Entity = TypeVar("Entity")

class IRepository(Generic[Entity], Protocol):
    session = Provide[DataBaseContainer.session]

api에서는 IRepository의 concreate class인 Repository를 사용한다.

@api_router.post(path="/session/example")
def repository_session_example2():
    repo = Repository()
    repo.get_session()

    return JSONResponse(status_code=200, content={})

테스트 결과는 다음과 같다.

3.3 dependency-injector의 Resource Provider

2번 방법에서는 dependency-injector의 container를 이용해 sqlalchemy engine과 session을 초기화한뒤 IRepository Classs Variable에 DI를 주입한 뒤 사용했다. 그러나 이 방법은 명시적인 session commit()과 close()를 적용해주어야 한다는 특징이 있다.

 

“1”번 방법에서 알 수 있듯 dependency-injector에서도 generator를 사용해 session을 관리하는 아이디어를 적용해 보자. dependency-injector에서는 Resource Provider가 이러한 기능을 수행해 준다.

from dependency_injector import containers, providers

from settings.dev import get_session, get_engine, get_db

class DataBaseContainer(containers.DeclarativeContainer):
    _engine = providers.Singleton(get_engine)

    _session = providers.Singleton(get_session, sa_engine=_engine.provided)

    session = providers.Resource(get_db, session=_session.provided)

상술한 코드를 IRepository에 적용하려면 다음과 같이 변경해야 한다.

from functools import cached_property
from typing import Protocol, Generic, TypeVar

from dependency_injector.wiring import Provide, Closing, inject
from sqlalchemy.orm import Session

from settings.dependency import DataBaseContainer

Entity = TypeVar("Entity")

class IRepository(Generic[Entity], Protocol):

    @cached_property
    @inject
    def session(self, _session: Session = Closing[Provide[DataBaseContainer.session2]]):
        return _session

class Repository(IRepository):

    def get_session(self):
        self.session.execute("select 1;")
        self.session.commit()
        self.session.close()

api 코드는 변경사항이 없다 테스트를 수행해 보자.

4. 정리 

다음은 DI 별 Request 발생과 그에 따른 RPS이다.

1. FastAPI의 Depands - 124181(Requests), 1456.1(RPS)
2. Dependency-injector의 Singleton Provider - 131437(Requests), 2168.8(PRS)
3. Dependency-injector의 Resource Provider - 129938(Requests), 2080.7(RPS)

이 이후로도 몇 번 더 같은 환경에서 RPS를 측정했지만 FastAPI의 Depands를 사용하는 경우가 RPS가 더 낮았다. 어떤 설정값과 환경을 설정했느냐에 따라 이 수치의 변화는 오차가 있겠지만 dependency-injector의 DI를 사용하는 케이스가 조금 더 좋은 것으로 보인다.