FastAPIにおける非同期処理の実装方法

## はじめに
こんにちは！今回はPythonのWebフレームワーク「FastAPI」を使った非同期処理の実装方法について、シンプルな例を用いて解説します。同期処理と非同期処理の比較、そして非同期タスクの結果取得方法も含めて説明していきます。

## 目次
1. FastAPIと非同期処理の概要
2. 同期処理と非同期処理の比較
3. 非同期タスクの実装と結果取得
4. シーケンス図による処理の可視化
5. まとめ

## 1. FastAPIと非同期処理の概要

FastAPIは、Pythonの非同期機能を最大限に活用できるWebフレームワークです。非同期処理を使うことで、I/O待ち時間を効率的に利用し、アプリケーションのパフォーマンスを向上させることができます。

## 2. 同期処理と非同期処理の比較

まずは、同期処理と非同期処理の違いを見てみましょう。

```python
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. 非同期タスクの実装と結果取得

次に、非同期タスクを開始し、後でその結果を取得する方法を実装してみましょう。

```python
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コードは、同期処理と非同期処理、そして非同期タスクの開始と結果取得のシーケンスを視覚化しています。

```plantuml
@startuml
actor Client
participant "FastAPI Server" as Server
participant "Sync Operations" as Sync
participant "Async Operations" as Async
participant "Task Manager" as TaskManager

== 同期処理 ==
Client -> Server: GET /sync-operations
activate Server
Server -> Sync: 実行
activate Sync
Sync -> Sync: 操作1 (2秒)
Sync -> Sync: 操作2 (1秒)
Sync -> Sync: 操作3 (3秒)
Sync --> Server: 結果返却
deactivate Sync
Server --> Client: レスポンス (約6秒後)
deactivate Server

== 非同期処理 ==
Client -> Server: GET /async-operations
activate Server
Server -> Async: 実行
activate Async
Async -> Async: 操作1 (2秒)
Async -> Async: 操作2 (1秒)
Async -> Async: 操作3 (3秒)
Async --> Server: 結果返却
deactivate Async
Server --> Client: レスポンス (約3秒後)
deactivate Server

== 非同期タスク開始と結果取得 ==
Client -> Server: POST /start-task/5
activate Server
Server -> TaskManager: タスク作成
activate TaskManager
TaskManager --> Server: タスクID返却
Server --> Client: タスクID
deactivate Server

Client -> Server: GET /task-result/{task_id}
activate Server
Server -> TaskManager: 結果確認
TaskManager --> Server: 進行中
Server --> Client: 進行中ステータス
deactivate Server

...5秒後...

Client -> Server: GET /task-result/{task_id}
activate Server
Server -> TaskManager: 結果確認
TaskManager --> Server: 完了結果
Server --> Client: タスク結果
deactivate Server
deactivate TaskManager

@enduml
```

この図は、同期処理、非同期処理、そして非同期タスクの開始と結果取得の流れを視覚的に表現しています。

## 6. Windows 11での動作確認

この記事で紹介したAPIの動作確認方法をWindows 11環境を前提に説明します。Windows 11でFastAPIアプリケーションを実行し、各エンドポイントをテストする方法を示します。

### アプリケーションの起動

まず、コマンドプロンプトまたはPowerShellを開き、アプリケーションのディレクトリに移動します。その後、以下のコマンドでアプリケーションを起動します：

```
uvicorn main:app --reload
```

ここで、`main`はPythonファイルの名前（拡張子なし）、`app`はFastAPIインスタンスの変数名です。

### エンドポイントのテスト

Windows 11では、コマンドプロンプトとPowerShellの両方が利用可能です。ここでは両方の例を示します。

#### コマンドプロンプトでの実行例：

1. 同期処理のテスト：
```
curl -o - http://127.0.0.1:8000/sync-operations
```

2. 非同期処理のテスト：
```
curl -o - http://127.0.0.1:8000/async-operations
```

3. 非同期タスクの開始：
```
curl -X POST -o - http://127.0.0.1:8000/start-task/5
```

4. タスク結果の取得（返されたtask_idを使用）：
```
curl -o - http://127.0.0.1:8000/task-result/[返されたtask_id]
```

注意: `-o -` オプションは、レスポンスを標準出力に表示するために使用しています。

#### PowerShellでの実行例：

1. 同期処理のテスト：
```powershell
Invoke-RestMethod -Uri "http://127.0.0.1:8000/sync-operations"
```

2. 非同期処理のテスト：
```powershell
Invoke-RestMethod -Uri "http://127.0.0.1:8000/async-operations"
```

3. 非同期タスクの開始：
```powershell
Invoke-RestMethod -Method Post -Uri "http://127.0.0.1:8000/start-task/5"
```

4. タスク結果の取得（返されたtask_idを使用）：
```powershell
Invoke-RestMethod -Uri "http://127.0.0.1:8000/task-result/[返されたtask_id]"
```

### 実行結果の例

以下は、PowerShellでの実行結果の例です：

1. 同期処理のテスト：
```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
```

2. 非同期処理のテスト：
```powershell
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
```

3. 非同期タスクの開始：
```powershell
PS> Invoke-RestMethod -Method Post -Uri "http://127.0.0.1:8000/start-task/5"

task_id
-------
f47ac10b-58cc-4372-a567-0e02b2c3d479
```

4. タスク結果の取得（開始直後）：
```powershell
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. タスク結果の取得（5秒後）：
```powershell
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アプリケーションの開発に役立つはずです。