LoginSignup
536
486

More than 3 years have passed since last update.

@bee2

[FastAPI] Python製のASGI Web フレームワーク FastAPIに入門する

Last updated at Posted at 2020-01-03

FastAPI

PythonのWeb frameworkで、Flaskのようなマイクロフレームワークにあたります。
パフォーマンスの高さ、書きやすさ、本番運用を強く意識した設計、モダンな機能などが強みです。

FastAPIはStarletteの肩に乗る形で書かれており、非同期処理が扱いやすいです。
特に、以下の様な特徴があります。

  • ASGI
  • websocketのサポート
  • GraphQLのサポート
  • バックグラウンドプロセスが扱いやすい
  • python type hintによる自動ドキュメント生成 (Swagger UI)
  • pydanticをベースとしたdata validation

率直に言って、responderに非常に似ています。(でた時期も近いですし、responderもStarletteがベースなので)
ですが、下の2つはFastAPIの方がよっぽど使いやすく設計されています。
以下の観点から総合的に見てFastAPIの方が本番運用向けだけだと思います。（個人的にはサクッと自由に書くならresponderの方が使いやすいと思います）

  • ドキュメントが丁寧 (DBとの連携、認証、https化なども紹介されている)
  • 自動ドキュメント生成機能が手厚いのでフロントエンドの開発者との連携が向上しそう
  • 本番運用のためのdocker imageまである

また、いくつかのPythonのframeworkとのパフォーマンスを比較しましたが、FastAPIは確かにパフォーマンスが高いと言えそうでした。（参考: PythonのWeb frameworkのパフォーマンス比較 (Django, Flask, responder, FastAPI, japronto)）

さらに、FastAPIは公式のドキュメント翻訳プロジェクトが動いています。私も微力ながら参加しています。以下、翻訳のガイドです。興味のある方はぜひ一緒にコントリビューションしましょう。

本記事の目的

FastAPIのありがたみを感じようとすると公式tutorialが適切かと思います。内容が充実しているのですごくわかりやすいです。しかし、その反面、使い始めるだけのために参照するのは少し量的に重いです。
そこで、必要最低限でFastAPIを使えるようになるための内容にまとめ直して紹介したいと思います。

また、本記事は、以下を想定して書いています。

  • pythonの何らかのmicroframeworkの基本的な記法が分かる
  • 基本的なpythonの型ヒント (mypy) の記法が分かる

ここで紹介する内容に相当するコード例をこちらにまとめています。Swaggerだけさわってみたいなどの場合にご利用下さい。

目次

  • intro
  • requestの扱い
  • responseの扱い
  • error handling & status code管理
  • background process
  • unittest
  • deployment
  • その他 (CORS問題への対処、認証)

intro

install FastAPI

fastapiとそのASGI serverとなるuvicornをinstallします。

$ pip install fastapi uvicorn

intro code

GETするとjsonで{"text": "hello world!"}が返ってくるAPIをたててみます。

intro.py
from fastapi import FastAPI

app = FastAPI()

@app.get('/') # methodとendpointの指定
async def hello():
    return {"text": "hello world!"}

Pythonのmicroframeworkの中でも簡潔に書けるほうだと思います。

run server

以下でサーバーが起動します。（--reloadとするとファイルの変更の度にサーバーが更新されるので開発時には便利です） intro:appの部分はfile名:FastAPI()のインスタンス名です。適宜置き換えて下さい。

$ uvicorn intro:app --reload

自動生成ドキュメント(Swagger UI)を確認

http://127.0.0.1:8000/docs にアクセスします。すると、Swagger UIが開きます。ここでAPIを叩くことができます。
f895d438c0b57a8272939ee4e3521af3.gif

また後述の方法でrequestとresponseのスキーマを確認したりできるようになります。FastAPIの大きな強みの一つがこのドキュメントが自動生成される点です。普通に開発していれば勝手にドキュメントが生成されていきます。

requestの扱い

以下の項目を扱います。

  • GET method:
    • path parameterの取得
    • query parameterの取得
    • validation
  • POST method:
    • request bodyの取得
    • validation

GET method

path parameter & query parameterの取得

parameterの取得はparameter名を引数に入れるだけで実現できます。
一旦、

  • endpointに/{param}のように宣言したparameter名はpath parameter
  • それ以外はquery parameterを指す

という理解をして下さい。また、引数の順番は関係ありません。そして、デフォルト値を宣言するか否かで引数に入っているparameterがGET時に入っていない場合の処理がかわります。

  • not required: デフォルト値を宣言すると、parameterがきていない場合にはデフォルト値が使われる
  • required: 一方、デフォルト値を宣言していないparameterがこないときは{"detail": "Not Found"}を返す

そして、引数は以下の様にpythonの型ヒントをつけるのがFastAPIの特徴です。

@app.get('/get/{path}')
async def path_and_query_params(
        path: str, 
        query: int, 
        default_none: Optional[str] = None):
    return {"text": f"hello, {path}, {query} and {default_none}"}

こうすることで、parameterの取得時に、pythonの型ヒントを考慮してFastAPIが、

  • 変換: データを指定した型に変換した状態で引数に入る
  • 検証: 指定した型に変換できない場合は、{"detail": "Not Found"}を返す
  • 自動ドキュメント生成： swagger UIに型情報を追記

を行います。実際にSwaggerを確認すると、以下の様にparameterの型情報が確認できます。　
5136078ab0a27e2f274d116438395bc2.png

validation

上記に加えて以下のQuery, Pathを使うと多少高度なことができます。Queryはquery parameter用で、Pathはpath parameter用です。

from fastapi import Query, Path

以下の様に使用します。QueryとPathの引数は基本的に同じものが使えて、

  • 第一引数はデフォルト値を指定。デフォルト値なし (required)にしたい場合は、...を渡す
  • alias: parameter名を指定します。引数名とparameter名を別にしたい時に使います。pythonの命名規則に反している場合用です
  • その他: 文字長、正規表現、値の範囲を指定して受け取る値を制限できます
@app.get('/validation/{path}')
async def validation(
        string: str = Query(None, min_length=2, max_length=5, regex=r'[a-c]+.'),
        integer: int = Query(..., gt=1, le=3),  # required
        alias_query: str = Query('default', alias='alias-query'),
        path: int = Path(10)):

    return {"string": string, "integer": integer, "alias-query": alias_query, "path": path}

Swaggerから制限内容も確認できます。APIが叩けるので、色々と値を変えてみて正しくvalidationがなされているか確認してみて下さい。

POST method

request bodyの取得

基本形

post dataの受け取り方を説明します。まず、基本は以下の様に、pydantic.BaseModelを継承した上で、attributesに型ヒントをつけたクラスを別途用意し、それをrequest bodyの型として引数で型ヒントをつければよいです。

from pydantic import BaseModel
from typing import Optional, List

class Data(BaseModel):
    """request data用の型ヒントがされたクラス"""
    string: str
    default_none: Optional[int] = None
    lists: List[int]

@app.post('/post')
async def declare_request_body(data: Data):
    return {"text": f"hello, {data.string}, {data.default_none}, {data.lists}"}

ここで、上記のコードは、以下のようなjsonがpostされてくる想定です。

requestBody
{
    "string": "string",
    "default_none": 0,
    "lists": [1, 2]
}

もしもfieldが足りなければstatus code 422が返ります。（余分なfieldが入っている場合は正常に動いているようです）
また、ここまでの処理を行うと、想定しているrequest bodyのデータ構造がSwagger UIから確認できるようになっています。
c21c87c01835cab42629eb3e88e30201.png

embed request body

先程の例と少しかわって以下の様なデータ構造の場合のための記法を説明します。

requestBody
{
    "data": {
        "string": "string",
        "default_none": 0,
        "lists": [1, 2]
    }
}

このような構造の場合は、Data classは先程のと同じものを使います。fastapi.Bodyを使うことで構造だけかえることができます。fastapi.BodyはGET methodのvalidationで紹介したpydantic.Queryの仲間です。同じく第一引数はデフォルト値です。pydantic.Queryなどにはなかったembedという引数を利用します。以下の微小な変更で構造の変更が実現できます。

from fastapi import Body

@app.post('/post/embed')
async def declare_embedded_request_body(data: Data = Body(..., embed=True)):
    return {"text": f"hello, {data.string}, {data.default_none}, {data.lists}"}

nested request body

次は、以下のようにリストや辞書がネストした構造の扱いを説明します。
subDataの構造は先程のembed request bodyの形ですが、異なる書き方を紹介します。

{
    "subData": {
        "strings": "string",
        "integer": 0
    },
    "subDataList": [
        {"strings": "string0", "integer": 0},
        {"strings": "string1", "integer": 1},
        {"strings": "string2", "integer": 2}
    ]
}

pythonの型ヒントであればネスト構造の型宣言は多くの場合、すごく大雑把にしかできません。(もし、大雑把でいいならば、以下のsubDataListはList[Any]とかList[Dict[str, Any]などの型をつけるだけで十分です)
一方、FastAPI (というかpydantic) だとネストした複雑な構造でも対応できます。
以下の様にネスト構造に沿って忠実にサブクラスを定義して型ヒントをつけていけばいいです。

class subDict(BaseModel):
    strings: str
    integer: int

class NestedData(BaseModel):
    subData: subDict
    subDataList: List[subDict]

@app.post('/post/nested')
async def declare_nested_request_body(data: NestedData):
    return {"text": f"hello, {data.subData}, {data.subDataList}"}

validation

GET methodとやること、できることはほぼ同じです。違いと言えば、fastapi.Queryなどではなく、pydantic.Fieldを使用する点です。しかし引数は差異がないです。
nested request bodyで使用した各クラスにpydantic.Fieldを導入しただけです。また、fastapi.Queryなどでも使えますが、引数exampleを利用しています。この引数に渡したデータがSwagger上からAPIを叩くときのデフォルト値になります。

from pydantic import Field

class ValidatedSubData(BaseModel):
    strings: str = Field(None, min_length=2, max_length=5, regex=r'[a-b]+.')
    integer: int = Field(..., gt=1, le=3)  # required

class ValidatedNestedData(BaseModel):
    subData: ValidatedSubData = Field(..., example={"strings": "aaa", "integer": 2})
    subDataList: List[ValidatedSubData] = Field(...)

@app.post('/validation')
async def validation(data: ValidatedNestedData):
    return {"text": f"hello, {data.subData}, {data.subDataList}"}

responseの扱い

responseにもrequest bodyで定義したようなクラスを定義してvalidationを行うことができます。

基本形

response_modelに渡すと、デフォルトで、

  • returnした辞書について、attributesに一致する名前が存在しないkeyは破棄される
  • returnした辞書には含まれないが、attributesにはデフォルト値がある場合はその値が補填される

ここで、以下のように書くと、returnしている辞書のうちintegerは捨てられ、auxが補われてjsonを返します。　(非常にシンプルな例を挙げていますが、ネストしていたり、少し複雑なvalidationが必要な場合は「requestの扱い」で挙げたような型ヒントについての記法をそのまま流用すればよいです)

class ItemOut(BaseModel):
    strings: str
    aux: int = 1
    text: str

@app.get('/', response_model=ItemOut)
async def response(strings: str, integer: int):
    return {"text": "hello world!", "strings": strings, "integer": integer}

この段階でSwaggerからresponse dataのschemaが確認できるようになります。
bb16c30d6110d5ec387b8e8edca89fc8.png

派生形

response_modelの利用はいくつかのオプションがあります。

# 辞書に存在しない場合にresponse_modelのattributesのデフォルト値を"いれない"
@app.get('/unset', response_model=ItemOut, response_model_exclude_unset=True)
async def response_exclude_unset(strings: str, integer: int):
    return {"text": "hello world!", "strings": strings, "integer": integer}

# response_modelの"strings", "aux"を無視 -> "text"のみ返す
@app.get('/exclude', response_model=ItemOut, response_model_exclude={"strings", "aux"})
async def response_exclude(strings: str, integer: int):
    return {"text": "hello world!", "strings": strings, "integer": integer}

# response_modelの"text"のみ考慮する -> "text"のみ返す
@app.get('/include', response_model=ItemOut, response_model_include={"text"})
async def response_include(strings: str, integer: int):
    return {"text": "hello world!", "strings": strings, "integer": integer}

error handling & status code管理

status code管理は3段階あります。

  • defaultのstatus codeを宣言: decoratorで宣言する
  • error handlingで400番台を返す: 適切な場所でfastapi.HTTPExceptionをraiseする
  • 柔軟にstatus code変更してreturnする: starletteを直接さわる
    • 引数にstarlette.responses.Responseを型としたものを追加
    • response.status_codeを書き換えると出力のstatus codeを変更できる
    • 通常通り返したいdataをreturnする
from fastapi import HTTPException
from starlette.responses import Response
from starlette.status import HTTP_201_CREATED

@app.get('/status', status_code=200) # default status code指定
async def response_status_code(integer: int, response: Response):
    if integer > 5:
        # error handling
        raise HTTPException(status_code=404, detail="this is error messages")
    elif integer == 1:
        # set manually
        response.status_code = HTTP_201_CREATED
        return {"text": "hello world, created!"}
    else:
        # default status code
        return {"text": "hello world!"}

background process

background processを用いれば、重い処理が完了する前にレスポンスだけ返すことができます。
WSGI (Djangoなど) 系だとこの処理は結構たいへんです。しかし、StarletteベースのASGIだとこの処理が非常に簡潔に扱えます。

手順は、
1. fastapi.BackgroundTasksを型とする引数を宣言
2. .add_taskでタスクを投げる

何が起きているのかこれだけでは予想もつかないですが、記述自体は簡単と言えるレベルかと思います。

重い処理の例として、受け取ったpath parameter秒だけスリープし、その後にprintするようなbackground processを実行してみます。

from fastapi import BackgroundTasks
from time import sleep
from datetime import datetime

def time_bomb(count: int):
    sleep(count)
    print(f'bomb!!! {datetime.utcnow()}')

@app.get('/{count}')
async def back(count: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(time_bomb, count)
    return {"text": "finish"} # time_bombの終了を待たずにレスポンスを返す

結果は、次の順序で処理されています
1. Response headersのdate=17時37分14秒
2. printに出力されたdate=17時37分25秒

なので、ちゃんとバックグラウンドで処理されていそうです。
スクリーンショット 2020-01-03 2.39.19.png

unittest

StarletteのTestClientというのが優秀で、unittestのために簡単にapiを叩けます。
今回は、tutorial通りに、pytestでunittestを行ってみます。

install

$ pip install requests pytest

directory配置

├── intro.py
└── tests
    ├── __init__.py
    └── test_intro.py

ここで、以下のunittestを行うことにします。

intro.py
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional, List
app = FastAPI()

@app.get('/')
async def hello():
    return {"text": "hello world!"}

class Data(BaseModel):
    string: str
    default_none: Optional[int] = None
    lists: List[int]

@app.post('/post')
async def declare_request_body(data: Data):
    return {"text": f"hello, {data.string}, {data.default_none}, {data.lists}"}

unittest

以下の様にstarlette.testclient.TestClientで簡単にGETとPOSTが叩けて、レスポンスのassertができるというのが売りです。

test_intro.py
from starlette.testclient import TestClient
from intro import app

# get and assign app to create test client
client = TestClient(app)

def test_read_hello():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"text": "hello world!"}

def test_read_declare_request_body():
    response = client.post(
        "/post",
        json={
            "string": "foo",
            "lists": [1, 2],
        }
    )
    assert response.status_code == 200
    assert response.json() == {
        "text": "hello, foo, None, [1, 2]",
    }

pytest実行

$ pytest
========================= test session starts =========================
platform darwin -- Python 3.6.8, pytest-5.3.2, py-1.8.1, pluggy-0.13.1
rootdir: ***/***/***
collected 1 items

tests/test_intro.py .                                            [100%]
========================== 1 passed in 0.44s ==========================

deployment

以下のような選択肢があります。シンプルなアプリケーションなので、インフラで困るようなことは少ないと思います。

  • pip installができて、uvicornさえ起動できればローカルと同じ用に動く
  • Docker image (Official): パフォーマンスのチューニングがされているそうです。何より公式なので信頼感があります。

基本的にdockerが使える場合は後者、それ以外に (PaaSでサクッとAPIたてるなど)の場合は前者の方法がいいと思います。

具体的なことに関しては、特にFastAPI特有の処理はなく、その他のmicroframeworkと何らかわりない手続きなので今回は省きます。
参考:

その他 (CORS問題への対処、認証)

他にtutorialとして書くまでもないけれども頻出の設定や、コンテキストに強く依存する事柄のリファレンスをまとめます。

まとめ

以上でミニマムのtutorialは終了です。
これで一通りのAPIサーバーの開発 -> deploymentまでできるようになると思います。

今回扱った内容に加えて、databaseとの連携、htmlのレンダリング、websocket、GraphQLなどを扱いたい場合はそれに関連するチャプターだけ参照すれば十分だと思います。

とにかく自動でSwaggerが生成されるのが便利なので自分で手を動かしながら試してみていただきたいです！

最後に、本記事の内容とはほぼ関係ないですが、FastAPIの公式ドキュメントで一番おもしろかったチャプターを紹介します。開発の経緯と他のframeworkとの差別化点が挙げられています。

Refs

536
486
2

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
536
486