Help us understand the problem. What is going on with this article?

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

今朝起きたら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で見れます。

スクリーンショット 2019-03-31 21.41.20.png

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

スクリーンショット 2019-03-31 21.41.28.png

では、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/

スクリーンショット 2019-03-31 21.46.39.png

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

終わりに

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

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

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

Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
No comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
ユーザーは見つかりませんでした