FastAPI サンプル - you1025/my_something_flagments GitHub Wiki

インストール

アプリケーションの実行

参考: 最初のステップ

main.app

下記を参考に GET メソッドを受け取る main.py を作成する。
python の dict を返すようにすると JSON で結果が返却される。

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
def hello():
    return { "message": "Hello, World!" }

Uvicorn

uvicorn を起動する。
--reload オプションを指定すると再起動なしでコードの変更が反映されるようになる。

カレントディレクトリに main.py が存在する場合は下記でよいが、 カレントディレクトリの hoge ディレクトリ配下に main.py が存在する場合は hoge.main:app を指定する必要がある事に注意。

$ uvicorn main:app --reload
INFO:     Will watch for changes in these directories: ['/Users/shimajiro/prog/python/fastapi_sample']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [2414] using statreload
INFO:     Started server process [2416]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

API の呼び出し

$ curl -X GET localhost:8000/hello
{"message":"Hello, World!"}

パラメータおよびデータの指定

パスパラメータ

参考: パスパラメータ

URL の一部をパラメータとして取得する。

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/hello/{name}")
def hello(name: str):
    return { "message": f"Hello, {name}!" }

$ curl localhost:8000/hello/shimajiro
{"message":"Hello, shimajiro!"}

パラメータ未指定の場合

単純に存在しない URL を指定した事になる。

$ curl localhost:8000/hello/
{"detail":"Not Found"}

クエリパラメータ

参考: クエリパラメータ

パスパラメータでない引数は自動的にクエリパラメータとして解釈される。

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
def hello(name: str):
    return { "message": f"Hello, {name}!" }

$ curl "http://localhost:8000/hello?name=shimajiro"
{"message":"Hello, shimajiro!"}

パラメータ未指定の場合

必須項目が未指定というエラーが返る。

$ curl "http://localhost:8000/hello"
{"detail":[{"loc":["query","name"],"msg":"field required","type":"value_error.missing"}]}

パラメータ未指定への対応

typing::Optional を用いて未指定を可能とする。
未指定のままで良い場合は下記のようにデフォルト値として None を指定するが、これ以外の文字列を設定する事も可能。

# main.py
from fastapi import FastAPI
from typing import Optional

app = FastAPI()

@app.get("/hello")
def hello(name: Optional[str] = None):
    return { "message": f"Hello, {name}!" }

未指定の場合でもエラーにならず None が指定されている。
文字列として "Hello, None!" と表記されているのは python の仕様で f"{None}" により "None" 文字列が生成される事による。

$ curl "http://localhost:8000/hello"
{"message":"Hello, None!"}

リクエストボディ

参考: リクエストボディ

下記に記載するように Pydantic::BaseModel と FastAPI::Body を用いた 2 通りの方法が存在する。

@app.post により POST メソッドを指定している事に注意。

Pydantic::BaseModel

リクエストボディに特定の型を指定する。
必須項目の指定や細かい指定が可能でこちらの方が高機能。

# main.app
from fastapi import FastAPI
from pydantic import BaseModel

class Data(BaseModel):
  message: str

app = FastAPI()

@app.post("/send")
def receive(data: Data):
    return data.message

$ curl -X POST -H "Content-Type: application/json" --data '{"message": "Hello, World!"}' localhost:8000/send
"Hello, World!"

Field による制約の付与

参考: Body - Fields

pydantic::Field を用いてリクエストボディの項目ごとに制約を追加する事ができる。
下記は Data::message 項目に最小文字数の制約を追加し、空文字の送付を弾いている。

from fastapi import FastAPI
from pydantic import BaseModel, Field

class Data(BaseModel):
  message: str = Field(min_length=1)

app = FastAPI()

@app.post("/send")
def receive(data: Data):
    return data.message

message に空文字を指定するとちゃんとエラーになる。

$ curl -X POST -H "Content-Type: application/json" --data '{"message": ""}' localhost:8000/send
{"detail":[{"loc":["body","message"],"msg":"ensure this value has at least 1 characters","type":"value_error.any_str.min_length","ctx":{"limit_value":1}}]}

FastAPI::Body

参考: Body - Multiple Parameters

リクエストボディに特定の型を指定しない。

# main.py
from fastapi import FastAPI, Body

app = FastAPI()

@app.post("/send")
def receive(body: dict = Body(...)):
    return body["message"]

$ curl -X POST -H "Content-Type: application/json" --data '{"message": "Hello, World!"}' localhost:8000/send
"Hello, World!"

データ未送信の場合

おこられる。

$ curl -X POST -H "Content-Type: application/json" localhost:8000/send
{"detail":[{"loc":["body"],"msg":"field required","type":"value_error.missing"}]

バックグラウンド処理

参考: Background Tasks

API として呼び出される関数の引数に FastAPI::BackgroundTasks のインスタンスを含め、 実行したいバックグラウンド処理を記載した関数を add_task で追加する。

# main.py
import time
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def heavy_task(message):
    print("start heavy task.")
    time.sleep(10)
    with open("./messages.txt", mode="a") as f:
        f.write(f"message: {message}\n")
    print("heavy task has done.")

@app.post("/kick_heavy_process")
async def receive_heavy_process(background_tasks: BackgroundTasks):
    background_tasks.add_task(heavy_task, message="がんばれ")
    return { "message": "receive your heavy task." }

API を呼び出すと即座にタスク受け入れのメッセージが返却され、約 10 秒後にカレントディレクトリの messages.txt にメッセージが追加される。

$ curl -X POST localhost:8000/kick_heavy_process

複数回の同時呼び出しでも即座に受け入れメッセージが返却される。

$ curl -X POST localhost:8000/kick_heavy_process; curl -X POST localhost:8000/kick_heavy_process; curl -X POST localhost:8000/kick_heavy_process
{"message":"receive your heavy task."}{"message":"receive your heavy task."}{"message":"receive your heavy task."}

uvicorn のログ。
まず 3 回呼び出しを受け入れてその後に順次処理を実行している事が分かる。

INFO:     127.0.0.1:56506 - "POST /kick_heavy_process HTTP/1.1" 200 OK
start heavy task.
INFO:     127.0.0.1:56508 - "POST /kick_heavy_process HTTP/1.1" 200 OK
start heavy task.
INFO:     127.0.0.1:56510 - "POST /kick_heavy_process HTTP/1.1" 200 OK
start heavy task.
heavy task has done.
heavy task has done.
heavy task has done.

Cloud Run での API 作成例

参考: クイックスタート: ビルドとデプロイ

ビルド用のソースコード

ソースコード用のディレクトリとして src を作成。名前は何でも良い。

$ mkdir src

src 以下に下記 4 ファイルを作成

  • requirements.txt
  • main.py
  • Dockerfile
  • .dockerignore

requirements.txt

fastapi
uvicorn

main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
def hello():
    return { "message": "Hello, World!" }

Dockerfile

# Use the official lightweight Python image.
# https://hub.docker.com/_/python
FROM python:3.9-slim

# Allow statements and log messages to immediately appear in the Knative logs
ENV PYTHONUNBUFFERED True

# Copy local code to the container image.
ENV APP_HOME /app
WORKDIR $APP_HOME
COPY . ./

# Install production dependencies.
RUN pip install --no-cache-dir -r requirements.txt

# Run the web service on container startup.
ENV PORT 8080
CMD exec uvicorn --host 0.0.0.0 --port $PORT main:app

.dockerignore

Dockerfile
README.md
*.pyc
*.pyo
*.pyd
__pycache__
.pytest_cache

Docker Image の作成

src ディレクトリを指定して Cloud Build でイメージを作成する。

  • [PROJECT-ID] には GCP のプロジェクト ID を指定する
  • [IMAGE-NAME] は(恐らく対象プロジェクト内で一意であれば)何でも良いはずなので適当な名前(ex. helloworld)を付ける
$ gcloud builds submit ./src --tag gcr.io/[PROJECT-ID]/[IMAGE-NAME]

詳細なオプションは gcloud builds submit を参照。

Cloud Run へのデプロイ

上記で作成したイメージを用いて Cloud Run 上にコンテナを追加する。

  • [サービス名] に指定した名前で API 呼び出しの URL が決まる(適当で良い。今回は helloservice とする)
  • --image には上記ビルドで指定した --tag の値を指定する

リージョンと未認証での呼び出しを聞かれるので適当に回答する。

$ gcloud run deploy [サービス名] --image gcr.io/[PROJECT-ID]/[IMAGE-NAME]

詳細なオプションは gcloud run deploy を参照。

API の呼び出し

URL の helloservice がデプロイ時に指定したサービス名となっている。

$ curl https://helloservice-qxtt2lxm3x-an.a.run.app/hello
{"message":"Hello, World!"}