FastAPI｜DB接続してCRUDするPython製APIサーバーを構築

17840 ワード
api Python FastAPI sqlalchemy Python テキストリンク

今朝起きたらFastAPIなるPython製のAPIサーバー構築ライブラリがバズっていた。

https://fastapi.tiangolo.com/

簡易的かつレスポンスの早いAPIを構築するときはGoとかNode.jsが有力候補な所、このFastAPIならそれらと同じくらいの速度が出るとのこと。

ということでFastAPIのドキュメントを見ながらDB接続してCRUDできるAPIを作ってみたので成果物をアップしてみます。APIの実装とかほとんど経験が無いのにアップしています。これだから無知は恐ろしい。

サンプルアプリとして、TODOの作成や閲覧、更新、削除ができるようなAPIを作成しました。

インストール

必要なPythonのバージョンはPython3.6以上。
今回のコードはPython3.6.5で動かしています。

必要なライブラリをインストールします。

$ pip install sqlalchemy uvicorn fastapi

DBの用意

API作るならDBと接続したいので、tutorialを見ながらDBを用意。今回はSQLiteで簡易的に試してみています。ORMはこれまたtutorial通りにSQLAlchemyを使っています。（tutorialがかなり整っているんです）
普段DjangoのORMを使うことが多いので、このSQLAlchemyの使い方を調べるのに時間がかかりました。

DB作成用のスクリプトは下記の通りです。

db.py
from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base


SQLALCHEMY_DATABASE_URI = "sqlite:///./test.db"
# SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URI, connect_args={"check_same_thread": False}, echo=True
)

Base = declarative_base()

# Todoテーブルの定義
class Todo(Base):
    __tablename__ = 'todos'
    id = Column('id', Integer, primary_key = True)
    title = Column('title', String(200))
    done = Column('done', Boolean, default=False)

# テーブル作成
Base.metadata.create_all(bind=engine)

このファイルを実行してDBとテーブルを作成します。

$ python db.py

試しにSQLiteに入って今作成したものを見てみます。

$ sqlite3 test.db
SQLite version 3.19.3 2017-06-27 16:48:08
Enter ".help" for usage hints.
sqlite> .ta
todos  ← 先程作成されたテーブルが表示される

FastAPIでAPIを作成

本題のAPI定義ファイルを作成します。

main.py
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session, sessionmaker
from starlette.requests import Request
from pydantic import BaseModel
from db import Todo, engine

# DB接続用のセッションクラス インスタンスが作成されると接続する
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Pydanticを用いたAPIに渡されるデータの定義 ValidationやDocumentationの機能が追加される
class TodoIn(BaseModel):
    title: str
    done: bool

# 単一のTodoを取得するためのユーティリティ
def get_todo(db_session: Session, todo_id: int):
    return db_session.query(Todo).filter(Todo.id == todo_id).first()

# DB接続のセッションを各エンドポイントの関数に渡す
def get_db(request: Request):
    return request.state.db

# このインスタンスをアノテーションに利用することでエンドポイントを定義できる
app = FastAPI()

# Todoの全取得
@app.get("/todos/")
def read_todos(db: Session = Depends(get_db)):
    todos = db.query(Todo).all()
    return todos

# 単一のTodoを取得
@app.get("/todos/{todo_id}")
def read_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = get_todo(db, todo_id)
    return todo

# Todoを登録
@app.post("/todos/")
async def create_todo(todo_in: TodoIn,  db: Session = Depends(get_db)):
    todo = Todo(title=todo_in.title, done=False)
    db.add(todo)
    db.commit()
    todo = get_todo(db, todo.id)
    return todo

# Todoを更新
@app.put("/todos/{todo_id}")
async def update_todo(todo_id: int, todo_in: TodoIn, db: Session = Depends(get_db)):
    todo = get_todo(db, todo_id)
    todo.title = todo_in.title
    todo.done = todo_in.done
    db.commit()
    todo = get_todo(db, todo_id)
    return todo

# Todoを削除
@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = get_todo(db, todo_id)
    db.delete(todo)
    db.commit()

# リクエストの度に呼ばれるミドルウェア DB接続用のセッションインスタンスを作成
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    request.state.db = SessionLocal()
    response = await call_next(request)
    request.state.db.close()
    return response

@app.get("/todos/") 部分からの5つの関数が各種エンドポイントの定義になっています。
レスポンスコードなんかはFastAPIがよしなに決定してくれるので楽にすっきりと書けますね。

それではASGI serverであるuvicornを起動して、動かしてみます。

$ uvicorn main:app --reload

起動が成功すると、APIサーバーが動くだけでなく、Swaggerドキュメントが見れたりReDocドキュメントが見れます。

Swaggerドキュメントはhttp://127.0.0.1:8000/docsで見れます。

ReDocドキュメントはhttp://127.0.0.1:8000/redocで見れます。

では、TODOデータを登録してみます。

curl -X POST "http://127.0.0.1:8000/todos/" -H  "accept: application/json" -H  "Content-Type: application/json" -d "{\"title\":\"hoge\",\"done\":false}"

登録したものを確認したいので、次はブラウザからアクセスしてみます。

http://127.0.0.1:8000/todos/

おお、登録できています。
ちなみにmain.py定義したデータ閲覧・登録・更新・削除といった各種機能は全てhttp://127.0.0.1:8000/docsから実行できます。

終わりに

コード量は少ないながら、APIができました。

軽量で速度の出るAPIサーバーを立てる時の有力な選択肢としては最近は特にGoが人気のようですが、Pythonありきの場合だとFastAPIを利用するのも良さそうです。Djangoは大きすぎるのでまだしも、これまでFlaskで作成することが多かった簡易的なAPIサーバーに対する対抗馬になりそうです。

例えば機械学習のモデルがあり、API経由で分類予測の結果を返すだけ、みたいな用途であればWebフレームワークの必要は無いので、効果的に使えるんじゃないでしょうか。