[fastapi] fastapi utils를 활용한 주기적인 작업(periodic task) 실행하기

백엔드 서버를 띄운 상황에서, 주기적인 작업(periodict task, repeated task)을 실행하고 싶은 경우가 생길 수 있습니다.

예를 들면, DB에 주기적으로 polling을 해서 어떤 데이터를 가져와서 서버에서 들고 있게 하는 등의 작업이 필요할 수 있는데요.

fastapi 자체적으로 제공하는 것은 없지만, fastapi-utils(https://fastapi-utils.davidmontague.xyz/)를 활용하면 이런 작업을 수행할 수 있습니다.

 

설치는 pip로 간단하게 하시면 됩니다.

pip install fastapi-utils

 

fastapi-utils를 활용하여 주기적인 작업을 실행하는 방법은 아래와 같습니다.

from fastapi import FastAPI
from sqlalchemy.orm import Session

from fastapi_utils.session import FastAPISessionMaker
from fastapi_utils.tasks import repeat_every

database_uri = f"sqlite:///./test.db?check_same_thread=False"
sessionmaker = FastAPISessionMaker(database_uri)

app = FastAPI()


def remove_expired_tokens(db: Session) -> None:
    """Pretend this function deletes expired tokens from the database"""


@app.on_event("startup")
@repeat_every(seconds=60 * 60)  # 1 hour
def remove_expired_tokens_task() -> None:
    with sessionmaker.context_session() as db:
        remove_expired_tokens(db=db)

 

fastapi 서버의 lifecycle을 활용하여 @app.on_event("startup") 을 같이 활용해야하고, @repeat_every() 데코레이터를 활용하면 쉽게 사용할 수 있는데요.

seconds로 몇 초 주기로 실행이 될 지 지정하면 됩니다.

 

이렇게 fastapi 서버에서 주기적인 작업을 해야하는 경우가 필요하다면, fastapi-utils 라이브러리를 활용하여 쉽고 간편하게 구현하시면 됩니다.

 

하나 주의해야 할 점은, 글 작성 시점인 2023년 6월 18일 기준으로 fastapi-utils의 버전은 0.2.1이고 sqlalchemy를 활용하는데 1.4.48 버전을 설치하게 됩니다.

sqlalchemy는 얼마 전 2.0 버전을 release했는데, fastapi-utils가 현재 기준 sqlalchemy 2.0을 지원하고 있지 않습니다.