[fastapi] lifespan

Lifespan

fastapi를 사용하면서, 앱이 시작할 때 혹은 앱이 종료될 때에 실행되어야 하는 로직이 분명 존재하게 됩니다.

저는 모니터링 관련 세팅이 앱 시작할 때 주로 실행되도록 구성을 많이 하고 있는데요.

이런 세팅들을 명시적으로 실행 위치를 정해주는 기능이 fastapi에 존재합니다.

lifespan이라는 기능인데, 오늘은 fastapi의 lifespan을 소개해보겠습니다.

 

먼저 위에서 언급했듯이, 앱이 시작하기 전 혹은 종료 시 실행되어야 하는 로직을 정하고 lifespan 함수를 선언해서 함수 안에 로직들을 작성하고 연결만해주면 됩니다.

바로 예시 코드를 보면서 소개드리겠습니다. 아래 코드는 fastapi 공식 문서에 소개된 코드입니다. (https://fastapi.tiangolo.com/advanced/events/)

Lifespan Events - FastAPI

from contextlib import asynccontextmanager

from fastapi import FastAPI


def fake_answer_to_everything_ml_model(x: float):
    return x * 42


ml_models = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load the ML model
    ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
    yield
    # Clean up the ML models and release the resources
    ml_models.clear()


app = FastAPI(lifespan=lifespan)


@app.get("/predict")
async def predict(x: float):
    result = ml_models["answer_to_everything"](x)
    return {"result": result}

 

주의해야할 점은 lifespan이라는 함수를 async로 선언하실 때, asynccontextmanager를 붙여주어야 합니다.

yield 구문 위에 존재하는 로직은 app이 실행되기 전에, 그리고 yield 구문 아래에 존재하는 로직은 app이 종료될 때 실행되게 됩니다.

 

위의 예제에서는 머신러닝 모델을 불러오고 종료하도록 로직을 선언하고 있는데요.

공식문서의 예제에서도 머신러닝이 언급될 만큼 머신러닝 분야에서 fastapi가 많이 사용되고 있다는 반증이 아닐까 싶습니다.

 

또한 위의 예제에서는 ml model을 초기화하는 코드가 async로 되어있지만, DB에 연결해서 무언가를 받아오는 것과 같은 로직이 필요하다면 async 함수로 구현할 것이고 그런 코드는 lifespan에 집어넣어서 실행해야 eventloop가 꼬이지 않게 됩니다.

 

sync로 구현되는 로직의 경우에는 함수로 선언하지 않고, fastapi app을 선언하기 전에 실행되도록만 코드를 구현한다면 앱이 실행되기 전에 로직을 실행할 수 있고 또 앱이 종료 될 때는 실행되는 로직은 어차피 서버가 꺼지는 상황이라면 굳이 실행이 될 필요가 없으므로 lifespan을 굳이 활용하지 않아도 상관은 없습니다.

 

하지만 코드를 좀 더 명시적으로 구현한다는 관점에서 앱이 실행되고 종료될 때 실행되어야 하는 로직이라는 것을 알 수 있도록 lifespan을 활용하는 습관을 들여보시면 좋을 것 같습니다.

 

기존 방식

참고로 위의 lifespan 사용 방식은 최근 업데이트로 인한 사용방법입니다.

기존의 방식도 소개를 드리자면, 생성된 fastapi app에 startup 혹은 shutdown 이벤트를 등록하는 방식입니다.

바로 코드 예시를 보도록 하겠습니다.

from fastapi import FastAPI

app = FastAPI()

items = {}


@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}


@app.on_event("shutdown")
def shutdown_event():
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]

위와 같이 @app.on_event("startup"), @app.on_event("shutdown") 이렇게 등록하면 됩니다.

기존 방식이 api endpoint 추가하는 방식과 비슷해서 코드가 깔끔해 보이긴 하지만 변경된 방식이 하나의 함수로 처리할 수 있다는 장점이 있는 것 같습니다.

기존의 방식은 fastapi의 버전이 upgrade 됨에 따라서, deprecated 될 예정이니 기존 방식을 사용하시던 분들도 빠르게 새로운 방식으로 변경하시는 것을 추천드립니다.