PythonでWeb APIを開発する場合はFlaskを選択していましたが、うちのメンバーから「FastAPI」を現場で使っていてなかなか良いと評判だったので簡単なWeb APIを作成しながら検証して行こうと思います。
Flaskでよく利用していた機能を中心にまとめています。
一番最初に感じたのが、Flaskの公式ドキュメントよりFastAPIの公式ドキュメントの方が分かりやすかった点です
https://fastapi.tiangolo.com/ja/
FastAPIとは?
FastAPIは、以下の特徴を持ったPython製のWebフレームワークです:
- 非同期処理が標準対応(
async/await) - 自動ドキュメント生成(OpenAPI, Swagger UI)
- Pythonの型ヒント(
pydantic)を活用してデータ検証(バリデーション・ドキュメント・エディタ補完)が可能。 - Flaskと比較して高速なレスポンス性能
インストールも簡単で、Flaskと同様に軽量なAPI開発が可能です。
開発環境の準備
$ pip install fastapi uvicorn
-
fastapi:フレームワーク本体 -
uvicorn:ASGIサーバ(FastAPIアプリを実行するために必要)
FlaskではFlask自体にサーバー機能も備わってましたがFastAPIではuvicornを利用して実行していきます。
nginxとの連携でもuWSGIの代わりとして使います。ちなみにsocketでも連携できます(こちらの方が安全かも)。
最小構成のFastAPIアプリ
まずは main.py というファイルを作成します。
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Hello, World!"}
ルート / にアクセスした際に Hello, World! を返すエンドポイントを定義しています。
サーバの起動
以下のコマンドで開発サーバを起動できます:
$ uvicorn main:app --reload
-
main:appはmain.pyの中にあるappオブジェクトを指定しています -
--reloadオプションはコード変更時に自動リロードしてくれるので開発時に便利です
起動後、curlで http://localhost:8000 にアクセスすると、以下のJSONが返ってくるはずです:
{
"message": "Hello, World!"
}
APIドキュメントの確認
FastAPIでは、OpenAPI仕様に基づいたインタラクティブなドキュメントが自動生成されます。
Swaggerのyaml編集が意外と苦痛だったりするのでこれば助かりますね
(OpenAPI仕様のYAML/JSONを手で編集する負担が減ります)
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
GETでクエリパラメータの取得方法
# main.py
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/get_name")
def get_name(name: str = Query(...)):
return {"name": name}
name: str = Query(...)
?name=xxx という形式のクエリパラメータを取得し、文字列としてnameに代入します。
...は「必須」を意味します(指定しなかった場合、422エラーになります)。
curlで検証
curl http://localhost:8000/get_name?name=yamada
結果
{
"name": "yamada"
}
必須ではなく任意のパラメータにする場合
# main.py
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/get_age")
def get_age(age: int = Query(None)):
return {"age": age}
指定しなかった場合は、ageにNoneが入ります
必須と任意のパラメータが混在する場合
# main.py
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/get_profile")
def get_profile(name: str = Query(...), age: int = Query(None)):
return {"name": name, "age": age}
指定しなかった場合は、ageにNoneが入ります
curlで検証
curl http://localhost:8000/get_profile?name=yamada
結果
{
"name": "yamada",
"age": null
}
###URLパラメーター
# main.py
@app.get('/get_name2/{name2}/myname')
def get_name2(name2: str):
return {"name": name2}
curlで検証
curl -i http://localhost:8000/get_name2/yamada/myname
結果
{
"name": "yamada"
}
まとめ
@app.get("/get_profile/{name2}/myname")
def get_profile(
name2: str, # パスパラメータは非デフォルトを先に
name: str = Query(...), # 必須クエリ
age: int | None = Query(None) # 任意クエリ
):
return {"name": name, "name2": name2, "age": age}
curl -i http://localhost:8000/get_profile/tarou/myname?name=yamada&age=18
結果
{
"name": "yamada",
"name2": "tarou",
"age": 18
}
CSVファイルや画像などのファイルの受取
formのmultipart/form-dataで送れば簡単に受け取ることが出来ました
from fastapi import FastAPI, File, UploadFile
app = FastAPI()
@app.post("/upload")
def upload_file(file: UploadFile = File(...)):
# file.fileに実体が入ってくる
data = file.file.read() # 読み出すとポインタが末尾になる点に注意(再利用するならseek(0)が必要)
return {
"filename": file.filename,
"content_type": file.content_type,
"size": len(data)
}
(大きなファイルの場合は async def と await file.read() の利用を推奨)
URLルーティングを別ファイルにまとめたい
ビジネスロジックのルーティングをファイル毎に分けられるので、ここで整理します
# main.py
from fastapi import FastAPI
import user, item
app = FastAPI()
# ルーターを登録
app.include_router(user.router)
app.include_router(item.router, prefix="/items", tags=["items"])
# user.py
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/user")
def get_users():
return {"name": "yamada"}
# item.py
from fastapi import APIRouter
router = APIRouter()
@router.get("/item")
def get_item():
return {"item": "Book"}
- Flaskと大きな違いはなさそうです
- prefixを呼び出し元か呼び出し先のrouterに書くかは好みがわかれそう
- OpenAPI ドキュメントにタグが反映されて、Swagger UI でグルーピング出来るのでこれば便利そうですね。
非同期処理とBackgroundTasks
FastAPI では async def を使う事で I/O待ち(DB / 外部API / sleep など)の間に他リクエストを処理できる。
from fastapi import FastAPI
import asyncio
from datetime import datetime
app = FastAPI()
@app.get("/async_demo")
async def async_demo():
for i in range(1, 11):
print(f"{datetime.now()}, {i}: 即時出力")
asyncio.create_task(delayed_line()) # ノンブロッキングで後続を予約
return {"message": f"{datetime.now()}, すぐレスポンスを返すよ"}
async def delayed_line():
await asyncio.sleep(5)
print(f"{datetime.now()}, 5秒後に出力されるよ")
BackgroundTasks を使う別パターン:
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
@app.get("/bg_task")
async def bg_task(bg: BackgroundTasks):
bg.add_task(time_sleep)
return {"message": "すぐレスポンスを返すよ"}
def time_sleep():
import time
time.sleep(5) # ここは同スレッド実行のため重い処理は避ける
print("5秒後に出力されるよ")
LINEのMessageAPIなどのレスポンスを返した後で裏で処理を実行したい場合はBackgroundTasksがいいのかな
async は「レスポンスに必要な処理」を 並行で処理する
BackgroundTasks は「レスポンス後でも良い処理」を 裏で処理する
Flaskでは大量アクセスの場合、DBのコネクションプーリングで接続の返却がうまくいかずに、
threadingを使ったので、負荷試験での検証は必要かな
リクエスト前後に処理を挿入する
middleware("http")は、ログイン後の認証やログの処理、レスポンスの加工処理に利用出来ます
HTTP Statusが404や500なんかでも処理が実行されます。
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def simple_middleware(request: Request, call_next):
# リクエスト前処理
# request.url.pathにアクセスしたpathが入ってくるので、ログイン処理などに利用可能
# 不必要なpathはここで弾くことも可能
print(f"リクエスト開始: {request.url.path}")
# 非同期ミドルウェアでは await が必要
response = await call_next(request)
# レスポンス後処理
print("レスポンス完了")
return response
@app.get("/menu")
def menu():
return {"message": "menuです"}
CORSの設定(CORSMiddleware)
SPA(例: React / Vue / Angular)やチャットボットの開発などで、フロントとFastAPIを別ドメインやポートで動かす際にCORSの制御が必要になったときに
FastAPIでは簡単に設定ができます
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# 許可するオリジン(フロントエンドのURL)
origins = [
"http://localhost:3000", # ローカルのReact/Vue開発環境
"https://example.com", # 本番環境のフロントエンド
]
# CORSMiddleware の追加
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # 許可するオリジン
allow_credentials=True, # Cookieや認証情報を含めるか
allow_methods=["*"], # 許可するHTTPメソッド(GET, POSTなど)
allow_headers=["*"], # 許可するHTTPヘッダー
)
@app.get("/menu")
def menu():
return {"message": "menuです"}
リクエスト(JSON)のバリデーションチェック
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class Profile(BaseModel):
name1: str
name2: str = Field(..., min_length=5, max_length=20, description="5文字以上20文字以下")
age: int = Field(..., gt=0, description="0より大きい整数")
@app.post("/get_profile")
def get_profile(profile: Profile):
return {"json": profile.model_dump()}
curl -i -X POST http://127.0.0.1:8000/get_profile \
-H 'Content-Type: application/json' \
-d '{"name1":"yamada","name2":"tarou","age":18}'
結果
{"json":{"name1":"yamada","name2":"tarou","age":18}}
ボディがjson以外だったり、バリデーションのエラーは全てHTTPが422で返されました
FastAPI が 422 (Unprocessable Entity) で自動レスポンスするようです
エラー内容はmodel_dump()で取得した JSON の detail 配列で確認できます。
カスタムバリデーションでも記述できるようですが、pydanticは別の記事にしたいと思います
サーバの起動/停止時に処理を追加する
ローカルのLLMを起動時に読み込んだり、一時ファイルの掃除に使ったり
from fastapi import FastAPI
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
print("アプリの起動処理を実行中...")
yield # アプリ実行中
# --- shutdown 相当 ---
print("アプリの停止処理を実行中...")
app = FastAPI(lifespan=lifespan)
@app.get("/")
def health():
return {"message": "OK"}
その他の主要機能
- 国際化(i18n)やロケール対応により多言語対応APIの構築が可能
- フロントとのWebSocket通信も @app.websocket() で標準で備わっている
- fastapi.securityでトークンベース認証対応 OAuth2パスワードフロー、JWT認証が簡単に?実装が可能
- エラーハンドリングもHTTPException, RequestValidationError などでエラー処理を統一化
FastAPIはFlaskと同様に学習コストが低く、Web APIの開発では有力な選択肢の一つとなりました
pydanticやRequestValidationErrorを上手に使って、ドキュメントから生成AIにコードを書かせることも視野にいれると次の開発はFastAPIで進めようと思います