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でファイルをダウンロードします。

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

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

ソースコード

from fastapi import FastAPI
from fastapi.responses import FileResponse
import os

app = FastAPI()

@app.get("/download/{filename}")
async def download_file(filename: str):
    file_path = os.path.join("files", filename)
    if os.path.exists(file_path):
        return FileResponse(path=file_path, filename=filename, media_type='application/octet-stream')
    return {"error": "File not found"}

コマンド実行例

$ 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 [1401886] using StatReload
INFO:     Started server process [1401889]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:48454 - "GET /download/hogehoge.txt HTTP/1.1" 200 OK
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [1401889]
INFO:     Stopping reloader process [1401886]

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

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

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

ソースコード

import requests

def download_file(url: str, save_path: str):
    response = requests.get(url, stream=True)
    if response.status_code == 200:
        with open(save_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)
        print(f"ダウンロード成功: {save_path}")
    else:
        print(f"ダウンロード失敗: {response.status_code}")

# 使用例
file_url = "http://localhost:8000/download/hogehoge.txt"
save_as = "downloaded_hogehoge.txt"
download_file(file_url, save_as)

コマンド実行例

$ python3 -B python-fastapi-file-download-client.py 
ダウンロード成功: downloaded_hogehoge.txt

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

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

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

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

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

ソースコード

from fastapi import FastAPI, File, UploadFile
import shutil
import os

app = FastAPI()

UPLOAD_DIR = "uploaded_files"
os.makedirs(UPLOAD_DIR, exist_ok=True)

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"filename": file.filename, "message": "ファイルアップロード成功"}

コマンド実行例

$ 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 [1402122] using StatReload
INFO:     Started server process [1402124]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:58898 - "POST /upload/ HTTP/1.1" 200 OK
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [1402124]
INFO:     Stopping reloader process [1402122]

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

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

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

ソースコード

import requests

def upload_file(url: str, file_path: str):
    with open(file_path, "rb") as f:
        files = {"file": (file_path, f)}
        response = requests.post(url, files=files)
    if response.status_code == 200:
        print(f"アップロード成功: {response.json()}")
    else:
        print(f"アップロード失敗: {response.status_code}")

# 使用例
upload_url = "http://localhost:8000/upload/"
upload_file(upload_url, "hogehoge.txt")

コマンド実行例

$ python3 -B python-fastapi-file-upload-client.py 
アップロード成功: {'filename': 'hogehoge.txt', 'message': 'ファイルアップロード成功'}

クライアント側でファイルアップロードの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コマンドでモジュール名を指定しますが、ソースコード内のオブジェクト名と合わせる必要があります。

まとめ

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

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

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

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