こんにちは、こんばんは、おはようございます。
K.S.ロジャース の島袋です。
以前PythonのWebフレームワークの次期覇権はコレ みたいな記事を書きましたが、それからしばらくPythonから離れたエンジニア生活を過ごしていたところ、なんと FastAPI なるフレームワークが登場しているではないですか！
正直、何番煎じかわからない遅れ方してますが、ぜひとも紹介させてください。
FastAPIの特徴
公式トップページ に書いてあるには、
- 高速 - NodeJSやGoなみに早い
- コード書くの早い - 2倍3倍で書ける！
- バグが減る - 40％くらい減る！
- 直感的 - デバッグが簡単
- 簡単 - 使いやすい。ドキュメント読むのも簡単！
- 短い - 重複が少ない。パラメータ宣言で複数の機能を呼べる
- 堅牢 - 自動生成あるからほんと楽
- スタンダード - OpenAPI標準だから互換性あるよ（Swaggerと互換性あるよ）
とありますが一番下の OpenAPI標準だから互換性あるよ（Swaggerと互換性あるよ） これが素晴らしすぎる。
FastAPI、実はSwagger（ドキュメント）の自動生成ができるのです！
ドキュメント自動生成
pipで fastapi と uvicorn をインストールしたら早速触ってみます。
ちなみに標準のpipには環境切り分けなどがないので、pipenv を利用することをおすすめします。
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
上記を main.py で作成後に uvicorn main:app --reload で起動。
http://127.0.0.1:8000/docs を確認すると…
はい！素敵！！
試しに path parameter を設定すると…
Path parameters
@app.get("/{name}")
async def root():
return {"message": "Hello " + name}
自動で反映されます。
ドキュメントが自動で作られていくとか最強です。
とくにドキュメントを重要視している弊社では便利すぎる機能です。
Query Parameters
@app.get("/")
async def read_item(name: str = ""):
return {"message": "Hello " + name}
Requests
Modelを使います。
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
app = FastAPI()
@app.post("/items/")
async def create_item(item: Item):
return item
さらにModelクラス内でのプロパティにバリデーションを指定できるので、取り回しが凄く良いです。
ここらへんは Django のModelと似たような構成ですね。
class Item(BaseModel):
name: str = Field(None, min_length=2, max_length=5)
description: str = Field(None, max_length=100)
CRUD (Crate, Read, Update, Delete)
基本のCRUDをざっくり書いてみます。
一部抜粋して書いているので、詳しくは書きリンクをご参考ください。
DB Connection & Migration
DBへの接続とマイグレーションは databases と sqlalchemy を使います。
今回は取り回しのよい SQLite をDBに使います。
さくっとコネクション周りを書いて、
import databases
import sqlalchemy
DATABASE_URL = "sqlite:///./db.sqlite3"
database = databases.Database(DATABASE_URL)
engine = sqlalchemy.create_engine(DATABASE_URL, echo=False)
metadata = sqlalchemy.MetaData()
Schemas
テーブル定義を記述して、同ファイルを実行しておきます。
import sqlalchemy
from db import metadata, engine
items = sqlalchemy.Table(
"items",
metadata,
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
sqlalchemy.Column("name", sqlalchemy.String),
sqlalchemy.Column("description", sqlalchemy.String),
)
metadata.create_all(engine)
Main
main.py も改修
from fastapi import FastAPI
from db import database
from starlette.requests import Request
from routers import items
app = FastAPI()
@app.on_event("startup")
async def startup():
# DBコネクション開始
await database.connect()
@app.on_event("shutdown")
async def shutdown():
# DBコネクション切断
await database.disconnect()
# routersを登録
app.include_router(items.router)
# ミドルウェアでDBコネクション埋め込んでおく（routerで取得できるように）
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
request.state.connection = database
response = await call_next(request)
return response
Routers
router もファイル分割してエンドポイント毎に記述できるようにします。
from fastapi import APIRouter, Depends, HTTPException
from typing import List
from databases import Database
from starlette.status import HTTP_204_NO_CONTENT
from utils import get_db_connection
from schemas import items
from models.item import ItemModel
router = APIRouter()
@router.get("/items", tags=["items"], response_model=List[ItemModel])
async def list_item(database: Database = Depends(get_db_connection)):
query = items.select()
return await database.fetch_all(query)
@router.post("/items", tags=["items"], response_model=ItemModel)
async def create_item(data: ItemModel, database: Database = Depends(get_db_connection)):
query = items.insert()
await database.execute(query, data.dict())
return {**data.dict()}
@router.patch("/items/{item_id}", tags=["items"], response_model=ItemModel)
async def update_item(item_id: int, data: ItemModel, database: Database = Depends(get_db_connection)):
query = items.update().where(items.columns.id==item_id)
ret = await database.execute(query, data.dict())
if not ret:
raise HTTPException(status_code=404, detail="Not Found")
return {**data.dict()}
@router.delete("/items/{item_id}", tags=["items"], status_code=HTTP_204_NO_CONTENT)
async def delete_item(item_id: int, database: Database = Depends(get_db_connection)):
query = items.delete().where(items.columns.id==item_id)
ret = await database.execute(query)
if not ret:
raise HTTPException(status_code=404, detail="Not Found")
まとめ
どうでしょう？ドキュメント自動生成はかなり魅力的な機能ではないでしょうか？
Responder もかなり良いフレームワークだと思いますが、FastAPI のほうが癖が少なく書きやすく感じました。
あとづけ
ちなみに弊社、Tech系以外にも会社ブログも掲載してますので、気になった方は是非どうぞ。
https://www.wantedly.com/companies/ks-rogers