MENU
やすひら
やすひらと申します。
長靴を履いたタヌキ(ITエンジニア)です。
モノ作りの楽しさを発信中。
X(旧Twitter)のフォローもお願いします。

[Python]FastAPIの概要と使い方

FastAPIは、Restful APIを作成するためのフレームワークです。

やすひら

FastAPIの概要と使い方を紹介します

この記事でわかること
  • FastAPIとは
  • FastAPIでできること
  • FastAPIの使い方
目次

FastAPIとは

FastAPIは、Restful APIを作成するためのフレームワークです。
FastAPIでは、名前の通り高速にAPI制御を実行することができます。
APIを自作することができるので、クライアントからサーバー側を操作したい場合に、FastAPIを利用すると良いです。

FastAPIでできること

FastAPIは、以下の特徴があります。

  • 高速処理によるAPI制御
  • 非同期処理対応
  • ドキュメント自動生成

FastAPIの使い方

FastAPIの使い方を紹介します。

FastAPIのインストール

FastAPIをインストールします。

コマンドライン

pip install fastapi uvicorn python-multipart

pipコマンドにより、関連ライブラリをインストールします。

python-multipartはファイル転送時に必要なライブラリです。

FastAPIによるシンプルなAPI開発

FastAPIでシンプルなAPIを開発します。

FastAPIによるシンプルなAPI開発(サーバー側)

サーバー側でFastAPIでシンプルなAPIを開発します。

ソースコード

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Message(BaseModel):
    user: str
    text: str

@app.post("/chat/")
def post_message(message: Message):
    return {"response": f"ユーザー名:{message.user} メッセージ:{message.text}"}

コマンド実行例

$ uvicorn python-fastapi-simple-api-server:app --reload
INFO:     Will watch for changes in these directories: ['/home/user']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1215025] using StatReload
INFO:     Started server process [1215027]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:44922 - "POST /chat/ HTTP/1.1" 200 OK
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [1215027]
INFO:     Stopping reloader process [1215025]

サーバー側でAPIを作成します。
上記は、チャットAPIで、ユーザー名とメッセージを応答します。

シンプルなAPIのリクエスト(クライアント側)

クライアント側でシンプルなAPIのリクエストを送信します。

ソースコード

import requests

url = "http://127.0.0.1:8000/chat/"
data = {
    "user": "やすひら",
    "text": "Hello"
}

response = requests.post(url, json=data)
print(response.json())

コマンド実行例

$ python3 -B python-fastapi-simple-api-client.py
{"response": "ユーザー名:やすひら メッセージ:Hello"}

クライアント側でAPIを操作します。
チャットAPIで、ユーザー名とメッセージが応答されました。

FastAPIによるAPI開発

FastAPIでAPIを開発します。

FastAPIによるAPI開発(サーバー側)

サーバー側でFastAPIでAPIを開発します。

ソースコード

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

messages = {}

class Message(BaseModel):
    user: str
    text: str

# POST: 新規登録
@app.post("/messages/{msg_id}")
def create_message(msg_id: int, message: Message):
    messages[msg_id] = message
    return {"status": "created", "message": message}

# GET: 取得
@app.get("/messages/{msg_id}")
def get_message(msg_id: int):
    if msg_id in messages:
        return {"message": messages[msg_id]}
    return {"error": "Message not found"}, 404

# PUT: 更新
@app.put("/messages/{msg_id}")
def update_message(msg_id: int, message: Message):
    if msg_id in messages:
        messages[msg_id] = message
        return {"status": "updated", "message": message}
    return {"error": "Message not found"}, 404

# DELETE: 削除
@app.delete("/messages/{msg_id}")
def delete_message(msg_id: int):
    if msg_id in messages:
        del messages[msg_id]
        return {"status": "deleted"}
    return {"error": "Message not found"}, 404

コマンド実行例

$ uvicorn python-fastapi-api-server:app --reload 
INFO:     Will watch for changes in these directories: ['/home/user']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1215983] using StatReload
INFO:     Started server process [1215985]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:46088 - "POST /messages/1 HTTP/1.1" 200 OK
INFO:     127.0.0.1:46096 - "GET /messages/1 HTTP/1.1" 200 OK
INFO:     127.0.0.1:46112 - "PUT /messages/1 HTTP/1.1" 200 OK
INFO:     127.0.0.1:46126 - "DELETE /messages/1 HTTP/1.1" 200 OK
INFO:     127.0.0.1:46136 - "GET /messages/1 HTTP/1.1" 200 OK
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [1215985]
INFO:     Stopping reloader process [1215983]

サーバー側でAPIを作成します。
上記は、チャットAPIで、ユーザー名とメッセージを制御します。

APIのリクエスト(クライアント側)

クライアント側でAPIのリクエストを送信します。

ソースコード

import requests

url = "http://localhost:8000/messages/1"
data = {"user": "やすひら", "text": "これは初回メッセージです"}

# POST:新規作成
res = requests.post(url, json=data)
print("POST:", res.json())

# GET:取得
res = requests.get(url)
print("GET:", res.json())

# PUT:更新
update_data = {"user": "やすひら", "text": "メッセージを更新しました"}
res = requests.put(url, json=update_data)
print("PUT:", res.json())

# DELETE:削除
res = requests.delete(url)
print("DELETE:", res.json())

# GET:再取得(削除後)
res = requests.get(url)
print("GET after delete:", res.json())

コマンド実行例

$ python3 -B python-fastapi-api-client.py 
POST: {'status': 'created', 'message': {'user': 'やすひら', 'text': 'これは初回メッセージです'}}
GET: {'message': {'user': 'やすひら', 'text': 'これは初回メッセージです'}}
PUT: {'status': 'updated', 'message': {'user': 'やすひら', 'text': 'メッセージを更新しました'}}
DELETE: {'status': 'deleted'}
GET after delete: [{'error': 'Message not found'}, 404]

クライアント側でAPIを操作します。
APIで、ユーザー名とメッセージが制御されました。

FastAPIによる非同期通信のAPI開発

FastAPIで非同期通信のAPIを開発します。

FastAPIによる非同期通信のAPI開発(サーバー側)

サーバー側でFastAPIで非同期通信のAPIを開発します。

ソースコード

from fastapi import FastAPI
from pydantic import BaseModel
import asyncio

app = FastAPI()

class Message(BaseModel):
    user: str
    text: str

messages = {}

# 非同期でPOST
@app.post("/messages/{msg_id}")
async def create_message(msg_id: int, message: Message):
    await asyncio.sleep(1)  # 疑似的な非同期処理
    messages[msg_id] = message
    return {"status": "created", "message": message}

# 非同期でGET
@app.get("/messages/{msg_id}")
async def get_message(msg_id: int):
    await asyncio.sleep(0.5)  # 疑似的な待機
    if msg_id in messages:
        return {"message": messages[msg_id]}
    return {"error": "Message not found"}

コマンド実行例

$ uvicorn python-fastapi-async-api-server:app --reload
INFO:     Will watch for changes in these directories: ['/home/user']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1216321] using StatReload
INFO:     Started server process [1216323]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:54850 - "POST /messages/1 HTTP/1.1" 200 OK
INFO:     127.0.0.1:54850 - "GET /messages/1 HTTP/1.1" 200 OK
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [1216323]
INFO:     Stopping reloader process [1216321]

サーバー側でAPIを作成します。
上記は、チャットAPIで、ユーザー名とメッセージを応答します。

非同期通信のAPIのリクエスト(クライアント側)

クライアント側で非同期通信のAPIのリクエストを送信します。

ソースコード

import httpx
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        # POST
        post_res = await client.post(
            "http://localhost:8000/messages/1",
            json={"user": "やすひら", "text": "Hello"}
        )
        print("POST:", post_res.json())

        # GET
        get_res = await client.get("http://localhost:8000/messages/1")
        print("GET:", get_res.json())

asyncio.run(main())

コマンド実行例

$ python3 -B python-fastapi-async-api-client.py

POST: {'status': 'created', 'message': {'user': 'やすひら', 'text': 'Hello'}}
GET: {'message': {'user': 'やすひら', 'text': 'Hello'}}

クライアント側で非同期通信のAPIを操作します。
非同期通信のAPIで、ユーザー名とメッセージが応答されました。

FastAPIによるファイルダウンロード

FastAPIでファイルをダウンロードします。

本記事では、ファイルの中身をbase64でエンコードしてファイルダウンロードを実現します。

  1. クライアント側でAPIのリクエストを実行
  2. サーバー側で、ファイルの中身をエンコード
  3. サーバー側で、エンコードデータを応答
  4. クライアント側で、エンコードデータをデコード
  5. クライアント側で、デコードデータをファイル保存

FastAPIによるファイルダウンロード(サーバー側)

サーバー側のファイルをFastAPIでダウンロードできるAPIを開発します。

ソースコード

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import base64

app = FastAPI()

def encode_base64(path: str) -> str:
    with open(path, "rb") as f:
        data = f.read()
    return base64.b64encode(data).decode("utf-8")

@app.post("/download")
async def download(request: Request):
    body = await request.json()
    filename = body.get("filename", "hogehoge.txt")

    # エンコード
    encoded = encode_base64(filename)

    # エンコードデータを応答する
    return JSONResponse(content={"filename": filename, "data": encoded})

コマンド実行例

$ ls
hogehoge.txt  python-fastapi-file-download-server.py
$ cat hogehoge.txt 
hogehoge
$ uvicorn python-fastapi-file-download-server:app --reload
INFO:     Will watch for changes in these directories: ['/home/user']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [2353931] using StatReload
INFO:     Started server process [2353933]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:58074 - "POST /download HTTP/1.1" 200 OK
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [2353933]
INFO:     Stopping reloader process [2353931]

サーバー側でAPIを作成します。
上記は、ファイルダウンロードのAPIです。

FastAPIによるファイルダウンロード(クライアント側)

クライアント側のファイルをFastAPIでダウンロードできるAPIを開発します。

ソースコード

import base64
import requests

def save_base64(b64_data: str, out_path: str):
    data = base64.b64decode(b64_data.encode("utf-8"))
    with open(out_path, "wb") as f:
        f.write(data)

if __name__ == "__main__":
    # REST APIに送信
    url = "http://localhost:8000/download"
    response = requests.post(url, json={"filename": "hogehoge.txt"})

    if response.status_code == 200:
        # サーバーから受け取ったデータをデコードしてファイル保存
        result = response.json()
        print(response.json())
        save_base64(result["data"], "hogehoge.txt")
        print("ファイルをダウンロードしました")
    else:
        print("エラー")

コマンド実行例

$ ls
python-fastapi-file-download-client.py
$ python3 -B python-fastapi-file-download-client.py 
{'filename': 'hogehoge.txt', 'data': 'aG9nZWhvZ2UK'}
ファイルをダウンロードしました
$ ls
hogehoge.txt  python-fastapi-file-download-client.py
$ cat hogehoge.txt 
hogehoge

クライアント側でファイルダウンロードのAPIを操作します。
APIでファイルをダウンロードできました。

FastAPIによるファイルアップロード

FastAPIでファイルをアップロードします。

本記事では、ファイルの中身をbase64でエンコードしてファイルアップロードを実現します。

  1. クライアント側で、ファイルの中身をエンコード
  2. クライアント側でのリクエストを実行
  3. サーバー側で、エンコードデータを受信
  4. サーバー側で、エンコードデータをデコード
  5. サーバー側で、デコードデータをファイル保存

FastAPIによるファイルアップロード(サーバー側)

サーバー側のファイルをFastAPIでアップロードできるAPIを開発します。

ソースコード

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import base64

app = FastAPI()

def save_base64(b64_data: str, out_path: str):
    data = base64.b64decode(b64_data.encode("utf-8"))
    with open(out_path, "wb") as f:
        f.write(data)

@app.post("/upload")
async def upload(request: Request):
    body = await request.json()
    filename = body.get("filename", "hogehoge.txt")
    b64_data = body["data"]

    # サーバー側でデコードしてファイル保存
    save_base64(b64_data, filename)

    return

コマンド実行例

$ ls
python-fastapi-file-upload-server.py
$ uvicorn python-fastapi-file-upload-server:app --reload
INFO:     Will watch for changes in these directories: ['/home/user']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [2354111] using StatReload
INFO:     Started server process [2354113]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:47322 - "POST /upload HTTP/1.1" 200 OK
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [2354113]
INFO:     Stopping reloader process [2354111]
$ ls
hogehoge.txt  __pycache__  python-fastapi-file-upload-server.py
$ cat hogehoge.txt 
hogehoge

サーバー側でAPIを作成します。
上記は、ファイルアップロードのAPIです。

FastAPIによるファイルアップロード(クライアント側)

クライアント側のファイルをFastAPIでアップロードできるAPIを開発します。

ソースコード

import base64
import requests

def encode(path: str) -> str:
    with open(path, "rb") as f:
        data = f.read()
    return base64.b64encode(data).decode("utf-8")

if __name__ == "__main__":
    # ファイル内容をエンコード
    b64_data = encode("hogehoge.txt")

    # REST APIに送信
    url = "http://localhost:8000/upload"
    response = requests.post(url, json={"filename": "hogehoge.txt", "data": b64_data})

    if response.status_code == 200:
        print("サーバーに送信しました")
    else:
        print("エラー")

コマンド実行例

$ ls
hogehoge.txt  python-fastapi-file-upload-client.py
$ cat hogehoge.txt 
hogehoge
$ python3 -B python-fastapi-file-upload-client.py 
サーバーに送信しました

クライアント側でファイルアップロードのAPIを操作します。
APIでファイルをアップロードできました。

トラブルシューティング

筆者が経験したトラブルについて紹介します。

モジュールなしエラー(ファイル名指定誤り)

ファイル名指定誤りによるモジュールがない場合はエラーとなります。

コマンド実行例

$ uvicorn python-fastapi-api-server.py:app --reload
INFO:     Will watch for changes in these directories: ['/home/user']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1221691] using StatReload
ERROR:    Error loading ASGI app. Could not import module "python-fastapi-api-server.py".
INFO:     Stopping reloader process [1221691]

ファイル名の指定が誤っているため、エラーとなっています。

uvicronコマンドでモジュール名を指定しますが、ファイル名の拡張子は不要です。

モジュールなしエラー(オブジェクト名指定誤り)

オブジェクト名指定誤りによるモジュールがない場合はエラーとなります。

コマンド実行例

$ uvicorn python-fastapi-api-server:hogehoge --reload
INFO:     Will watch for changes in these directories: ['/home/user']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [1221727] using StatReload
ERROR:    Error loading ASGI app. Attribute "hogehoge" not found in module "python-fastapi-api-server".
INFO:     Stopping reloader process [1221727]

オブジェクト名の指定が誤っているため、エラーとなっています。

uvicronコマンドでモジュール名を指定しますが、ソースコード内のオブジェクト名と合わせる必要があります。

外部からAPIにリクエストできない

外部からAPIにリクエストできない場合は、以下の可能性があります。

IPアドレスやポート番号が誤っている

IPアドレスやポート番号が誤っている場合は、エラーとなります。
クライアント側で正しいIPアドレスとポート番号を指定して、リクエストを送信します。

ファイアウォールで通信が遮断されている

ファイアウォールで通信が遮断されている場合は、エラーとなります。
Ubuntuの場合は、ufwコマンド等でファイアウォールを設定しています。
ファイアウォールの設定を見直して、リクエストを送信します。

サーバー側のアドレス指定が誤っている

uviconコマンドによるサーバー起動において、アドレス指定が誤っている場合はエラーとなります。

コマンドライン

uvicorn python-fastapi-api-server:app --host 0.0.0.0 --port 8000

IPアドレスを”0.0.0.0″で指定する必要があります。

まとめ

Pythonを用いたFastAPIの概要と使い方を紹介しました。

FastAPIは
  • API開発できる
  • Pythonで開発できる
  • 非同期で処理できる

FastAPIを利用すれば、PythonでAPIを開発することができます。
サーバークライアント構成で通信する場合や、マイクロサービスで開発する場合に有効なので、利用できると良いです。

  • URLをコピーしました!
目次