🎑

FastAPIの導入とサービス化

2024/01/05に公開

Python

FastAPI

suporvisor

tech

はじめに

PythonのFastAPIを利用することにしましたのでその基礎の勉強と、supervisorで常時起動化する方法までを実施しました。

FastAPIの導入

次のコマンドでインストールします。

pip install "fastapi[all]"
pip install uvicorn

FastAPIの簡単なサンプル

テスト処理の作成

パスやクエリは簡単に取得可能
JSONの場合は型定義して取得する方法もある。

test.py

from fastapi import Request, FastAPI
from typing import Any, Dict, List, Union
from pydantic import BaseModel

import sys

app = FastAPI()

@app.get("/")
def test0():
    print("test0.stdout")
    print("test0.stderr",file=sys.stderr)
    return {"TEST1": "TEST2"}

@app.get("/test1/{test_item}" )
def test1(test_item: int):
    # curl -v http://127.0.0.1:8000/test1/1
    # curl -v http://127.0.0.1:8000/test1/2
    return {"test_item_id": test_item}

@app.get("/test2/")
def test2(q: Union[str, None] = None):
    # curl -v 'http://127.0.0.1:8000/test2/?q=3&test1=test2'
    # qというパラメータ名のみが対象
    # 他は無視される
    return {"q":q}

@app.get("/test3/")
def test3(req: Request):
    # 全部取得
    # curl -v 'http://127.0.0.1:8000/test3/?q=3&test1=test2'
    # print(req,dict(req))
    request_dict = dict(req.query_params)
    return {"request_dict":request_dict}

@app.post("/test_post1/")
async def test_post1(request: Request):
    # curl -v -X POST -H "Content-Type: application/json" http://127.0.0.1:8000/test_post1/ -d '{"A":"B","C":"D"}'
    return await request.json()

class TEST2(BaseModel):
    A: str
    C: str

@app.post("/test_post2/")
def test_post2(d: TEST2 ):
    # curl -v -X POST -H "Content-Type: application/json" http://127.0.0.1:8000/test_post2/ -d '{"A":"B","C":"D"}'
    # 以下だとエラーになる
    # curl -v  http://127.0.0.1:8000/test_post2/ -d '{"A":"B","C":"D"}'
    # Content-Typeが大切
    return {"テスト":d}

@app.post("/test_post3/")
def test_post3(json_data: dict = {}):
    # curl -v -X POST -H "Content-Type: application/json" http://127.0.0.1:8000/test_post3/ -d '{"A":"B","C":"D"}'
    return json_data

テスト起動方法

uvicornを利用して起動します。
コードの変更は--reloadをつけていると自動で再反映されます。
hostとportはそれぞれ指定します。
portはデフォルトからはずらしておいたほうがよいです。

uvicorn test:app --reload --host=0.0.0.0 --port=8000

test:appの部分は、test.pyのappオブジェクトという意味になる
配下ディレクトリを指定する場合はディレクトリ名.ファイル名:オブジェクト名
stdoutもstderrもそのままコンソールに出力される。

起動シェルを作成

sudo vi /root/start.sh

/root/start.sh

#!/bin/bash

umask 000
cd /root/test_files/ # ソースを配置したディレクトリ
uvicorn test:app --reload --host=0.0.0.0 --port=8000

サービス化

Supervisorを利用する。
ブロックするプロセスの常時起動にはsystemctldより簡易
子プロセスもプロセスグループ単位で停止してくれる。

http://supervisord.org/

インストール

sudo apt update
sudo apt install supervisor

sudo systemctl enable supervisor
sudo systemctl start supervisor
sudo systemctl status supervisor

設定の実施

sudo nano /etc/supervisor/conf.d/test.conf

/etc/supervisor/conf.d/test.conf

[program:test_fastapi]
command=bash /root/start.sh
autostart=true
autorestart=true
stderr_logfile=/root/test_fastapi-stderr.log
stdout_logfile=/root/test_fastapi-stdout.log
stopasgroup=true
killasgroup=true
stopsignal=INT

stopasgroupとkillasgroupを指定しておくと子プロセスも停止対象
未指定の場合は親プロセスでしっかりケアする必要がある。
特に起動シェルを挟む場合は起動シェルしか停止されないので注意が必要となる。

設定のリロード

supervisorctl reload

起動・停止・状態

以下のコマンドで制御可能
supervisor起動時には自動で起動される

開始・再起動

supervisorctl start test_fastapi
supervisorctl restart test_fastapi

停止・状態表示

supervisorctl stop test_fastapi
supervisorctl status test_fastapi

その他

以下の点も調査予定

TODO:ヘッダの取得、ミドルウェア的なヘッダによる認証の実施なども動作確認
TODO:標準出力に出るログのフォーマットについても確認
TODO:X-Forwareded-Forを考慮したリクエスト元IPの取得方法
TODO:テンプレートエンジンを利用する場合の方法、FastAPIとしては関与せずで多くはJinja2の直接利用っぽい？
TODO:多言語対応。こちらもFastAPIとしては関与せずでpython-i18n等を利用？
TODO:複数のpyファイルに分けた場合のEndpointの定義、階層化
TODO:応答時のHTTPステータスの指定
TODO:大きなファイルのアップロード
TODO:/docsや/redocなどの無効化