FastAPIの起動時にファイルの読み込みをしておきたいとか、終了時に終わらせておきたい処理などを実行したい場合は、lifespanを使用する。
asynccontextmanagerを使用して非同期関数を作成し、FastAPIのオブジェクト生成時に関数を渡す。
作成した関数で処理を実装する。起動後にやりたいことが済んだら、yieldをコールして処理を中断する。
FastAPIの終了時に処理が再開する。
@asynccontextmanager async def lifespan(app: FastAPI): # 起動時にやりたいこと logger.info("Start up!!!") yield # 終了時にやりたいこと logger.info("Shutdown!!!") app = FastAPI(lifespan=lifespan)
ファイルの中身を読み込んでリクエストを受ける
ユースケース
ファイルの中にステータスコードとメッセージを定義したファイルを起動時に読み込んで、リクエストの受信数で返す内容を変更する。
動作結果
起動時にファイルの読み込みと終了時にログ出力を設定して動作できた。 設定したコンフィグの内容を使って応答を返せている。
INFO: Will watch for changes in these directories: ['/Users/kotaro/github/pythonStudy/src/fastapi']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [7535] using WatchFiles
INFO: Started server process [7537]
INFO: Waiting for application startup.
INFO: Start up!!!
INFO: messages = [{'status': 200, 'message': 'Success'}, {'status': 400, 'message': 'Error!!!'}, {'status': 500, 'message': 'ServerError!!!'}]
INFO: Application startup complete.
# リクエスト受信1回目
INFO: cnt = 0
INFO: statuscode = 200, body = b'{"result":"Success"}'
INFO: 127.0.0.1:55179 - "GET /message HTTP/1.1" 200 OK
# リクエスト受信2回目
INFO: cnt = 1
INFO: statuscode = 400, body = b'{"result":"Error!!!"}'
INFO: 127.0.0.1:55241 - "GET /message HTTP/1.1" 400 Bad Request
# リクエスト受信3回目
INFO: cnt = 2
INFO: statuscode = 500, body = b'{"result":"ServerError!!!"}'
INFO: 127.0.0.1:55241 - "GET /message HTTP/1.1" 500 Internal Server Error
INFO: Shutting down
INFO: Waiting for application shutdown.
INFO: Shutdown!!!
INFO: Application shutdown complete.
INFO: Finished server process [8733]
INFO: Stopping reloader process [8731]
最後に
FastAPIの起動／終了時に処理を実行する方法を確認した。 起動までに読み込んでおきたいデータや起動するために前提条件として必要なデータを取得できる。 sqlalchemyのエンジンオブジェクトの生成などもここでやるべきなのかな？
参考
errormessages: - status: 200 message: "Success" - status: 400 message: "Error!!!" - status: 500 message: "ServerError!!!"
from fastapi import FastAPI from logging import getLogger from contextlib import asynccontextmanager import yaml from fastapi.responses import JSONResponse logger = getLogger("uvicorn.app") messages = [] cnt = 0 @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Start up!!!") with open("config.yaml", "r") as yml: config = yaml.safe_load(yml) for item in config["errormessages"]: messages.append(item) logger.info("messages = " + str(messages)) yield logger.info("Shutdown!!!") app = FastAPI(lifespan=lifespan) @app.get("/message") async def get_data(): global cnt logger.info("cnt = " + str(cnt)) if cnt == 0: idx = cnt cnt += 1 elif cnt == 1: idx = cnt cnt += 1 elif cnt == 2: idx = cnt cnt = 0 response = JSONResponse( status_code=messages[idx]["status"], content={"result": messages[idx]["message"]}, ) logger.info( "statuscode = {}, body = {}".format(response.status_code, response.body) ) return response
