はじめに
こんにちは！今回はPythonのWebフレームワーク「FastAPI」を使った非同期処理の実装方法について、シンプルな例を用いて解説します。同期処理と非同期処理の比較、そして非同期タスクの結果取得方法も含めて説明していきます。
目次
- FastAPIと非同期処理の概要
- 同期処理と非同期処理の比較
- 非同期タスクの実装と結果取得
- シーケンス図による処理の可視化
- まとめ
1. FastAPIと非同期処理の概要
FastAPIは、Pythonの非同期機能を最大限に活用できるWebフレームワークです。非同期処理を使うことで、I/O待ち時間を効率的に利用し、アプリケーションのパフォーマンスを向上させることができます。
2. 同期処理と非同期処理の比較
まずは、同期処理と非同期処理の違いを見てみましょう。
from fastapi import FastAPI
import asyncio
import time
app = FastAPI()
def sync_operation(operation_id: int, sleep_time: int):
time.sleep(sleep_time)
return f"Sync operation {operation_id} completed after {sleep_time} seconds"
async def async_operation(operation_id: int, sleep_time: int):
await asyncio.sleep(sleep_time)
return f"Async operation {operation_id} completed after {sleep_time} seconds"
@app.get("/sync-operations")
def sync_operations():
start_time = time.time()
results = [
sync_operation(1, 2),
sync_operation(2, 1),
sync_operation(3, 3)
]
end_time = time.time()
return {
"results": results,
"total_time": end_time - start_time
}
@app.get("/async-operations")
async def async_operations():
start_time = time.time()
tasks = [
async_operation(1, 2),
async_operation(2, 1),
async_operation(3, 3)
]
results = await asyncio.gather(*tasks)
end_time = time.time()
return {
"results": results,
"total_time": end_time - start_time
}
この例では、/sync-operationsエンドポイントが同期的に3つの操作を実行し、/async-operationsエンドポイントが非同期的に3つの操作を実行します。
同期処理では、各操作が順番に実行されるため、全体の実行時間は各操作の実行時間の合計（この場合は6秒）になります。
一方、非同期処理では、3つの操作が並行して実行されるため、全体の実行時間は最も時間のかかる操作（この場合は3秒）とほぼ同じになります。
3. 非同期タスクの実装と結果取得
次に、非同期タスクを開始し、後でその結果を取得する方法を実装してみましょう。
import uuid
from pydantic import BaseModel
# タスクの結果を保存する辞書
task_results = {}
class TaskResponse(BaseModel):
task_id: str
class TaskResult(BaseModel):
result: str
status: str
execution_time: float
async def long_running_task(task_id: str, sleep_time: int):
start_time = time.time()
await asyncio.sleep(sleep_time)
end_time = time.time()
execution_time = end_time - start_time
task_results[task_id] = TaskResult(
result=f"Task completed after {sleep_time} seconds",
status="completed",
execution_time=execution_time
)
@app.post("/start-task/{sleep_time}")
async def start_task(sleep_time: int):
task_id = str(uuid.uuid4())
task_results[task_id] = None # タスクが進行中であることを示す
asyncio.create_task(long_running_task(task_id, sleep_time))
return TaskResponse(task_id=task_id)
@app.get("/task-result/{task_id}")
async def get_task_result(task_id: str):
if task_id not in task_results:
raise HTTPException(status_code=404, detail="Task not found")
if task_results[task_id] is None:
return TaskResult(result="Task is still running", status="in_progress", execution_time=0)
return task_results[task_id]
この実装では、/start-task/{sleep_time}エンドポイントで新しい非同期タスクを開始し、タスクIDを返します。クライアントは、このタスクIDを使って/task-result/{task_id}エンドポイントからタスクの結果を取得できます。
4. シーケンス図による処理の可視化
以下のPlantUMLコードは、同期処理と非同期処理、そして非同期タスクの開始と結果取得のシーケンスを視覚化しています。
この図は、同期処理、非同期処理、そして非同期タスクの開始と結果取得の流れを視覚的に表現しています。
6. Windows 11での動作確認
この記事で紹介したAPIの動作確認方法をWindows 11環境を前提に説明します。Windows 11でFastAPIアプリケーションを実行し、各エンドポイントをテストする方法を示します。
アプリケーションの起動
まず、コマンドプロンプトまたはPowerShellを開き、アプリケーションのディレクトリに移動します。その後、以下のコマンドでアプリケーションを起動します：
uvicorn main:app --reload
ここで、mainはPythonファイルの名前（拡張子なし）、appはFastAPIインスタンスの変数名です。
エンドポイントのテスト
Windows 11では、コマンドプロンプトとPowerShellの両方が利用可能です。ここでは両方の例を示します。
コマンドプロンプトでの実行例：
- 同期処理のテスト：
curl -o - http://127.0.0.1:8000/sync-operations
- 非同期処理のテスト：
curl -o - http://127.0.0.1:8000/async-operations
- 非同期タスクの開始：
curl -X POST -o - http://127.0.0.1:8000/start-task/5
- タスク結果の取得（返されたtask_idを使用）：
curl -o - http://127.0.0.1:8000/task-result/[返されたtask_id]
注意: -o - オプションは、レスポンスを標準出力に表示するために使用しています。
PowerShellでの実行例：
- 同期処理のテスト：
Invoke-RestMethod -Uri "http://127.0.0.1:8000/sync-operations"
- 非同期処理のテスト：
Invoke-RestMethod -Uri "http://127.0.0.1:8000/async-operations"
- 非同期タスクの開始：
Invoke-RestMethod -Method Post -Uri "http://127.0.0.1:8000/start-task/5"
- タスク結果の取得（返されたtask_idを使用）：
Invoke-RestMethod -Uri "http://127.0.0.1:8000/task-result/[返されたtask_id]"
実行結果の例
以下は、PowerShellでの実行結果の例です：
- 同期処理のテスト：
PS> Invoke-RestMethod -Uri "http://127.0.0.1:8000/sync-operations"
results total_time
------- ----------
{Sync operation 1 completed after 2 seconds, Sync operation... 6.005197
- 非同期処理のテスト：
PS> Invoke-RestMethod -Uri "http://127.0.0.1:8000/async-operations"
results total_time
------- ----------
{Async operation 1 completed after 2 seconds, Async operati... 3.002893
- 非同期タスクの開始：
PS> Invoke-RestMethod -Method Post -Uri "http://127.0.0.1:8000/start-task/5"
task_id
-------
f47ac10b-58cc-4372-a567-0e02b2c3d479
- タスク結果の取得（開始直後）：
PS> Invoke-RestMethod -Uri "http://127.0.0.1:8000/task-result/f47ac10b-58cc-4372-a567-0e02b2c3d479"
result status execution_time
------ ------ --------------
Task is still running in_progress 0
- タスク結果の取得（5秒後）：
PS> Invoke-RestMethod -Uri "http://127.0.0.1:8000/task-result/f47ac10b-58cc-4372-a567-0e02b2c3d479"
result status execution_time
------ ------ --------------
Task completed after 5 seconds completed 5.001337
これらの例を通じて、同期処理と非同期処理の実行時間の違い、そして非同期タスクの進行状況と結果の取得を観察することができます。
Windows 11環境でこれらのコマンドを実行することで、FastAPIアプリケーションの動作を確認し、非同期処理の効果を実感することができます。
まとめ
この記事では、FastAPIを使用した非同期処理の実装方法について、理論的な説明だけでなく、実際の動作確認方法まで含めて解説しました。Windows 11環境でのテスト方法を示すことで、読者の皆さんがより簡単に非同期処理の効果を体験できるようになりました。
非同期プログラミングは現代のWeb開発において重要なスキルの一つです。この記事で紹介した内容を基に、ぜひ自分のプロジェクトでFastAPIと非同期処理を試してみてください。効率的で高性能なWebアプリケーションの開発に役立つはずです。