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

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

[https://fastapi.tiangolo.com/](https://fastapi.tiangolo.com/)

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

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

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

# インストール

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

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

```terminal
$ pip install sqlalchemy uvicorn fastapi
```

# DBの用意

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

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

```python: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とテーブルを作成します。

```terminal
$ python db.py
```

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

```terminal
$ 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定義ファイルを作成します。

```python: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を起動して、動かしてみます。

```terminal
$ uvicorn main:app --reload
```

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

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

![スクリーンショット 2019-03-31 21.41.20.png](https://qiita-image-store.s3.amazonaws.com/0/113133/8402245f-9b1d-5448-979f-cabac2cff0bb.png)

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

![スクリーンショット 2019-03-31 21.41.28.png](https://qiita-image-store.s3.amazonaws.com/0/113133/87b2c0b7-5c36-865a-4ed7-0fece5c4a581.png)

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

```terminal
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/](http://127.0.0.1:8000/todos/)

![スクリーンショット 2019-03-31 21.46.39.png](https://qiita-image-store.s3.amazonaws.com/0/113133/7e258f84-2b91-cffc-8c1d-c09066db6697.png)

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

# 終わりに

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

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

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