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
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のリクエストを送信します。
ソースコード
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"}
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のリクエストを送信します。
ソースコード
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]
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のリクエストを送信します。
ソースコード
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'}}
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
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]
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': 'ファイルアップロード成功'}
トラブルシューティング
筆者が経験したトラブルについて紹介します。
モジュールなしエラー(ファイル名指定誤り)
ファイル名指定誤りによるモジュールがない場合はエラーとなります。
コマンド実行例
$ 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]
モジュールなしエラー(オブジェクト名指定誤り)
オブジェクト名指定誤りによるモジュールがない場合はエラーとなります。
コマンド実行例
$ 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]
まとめ
Pythonを用いたFastAPIの概要と使い方を紹介しました。
- API開発できる
- Pythonで開発できる
- 非同期で処理できる
FastAPIを利用すれば、PythonでAPIを開発することができます。
サーバークライアント構成で通信する場合や、マイクロサービスで開発する場合に有効なので、利用できると良いです。