async-python-patterns

非同期 Python パターン

asyncio、並行プログラミングのパターン、async/await を使って、高性能でノンブロッキングなシステムを構築する非同期 Python アプリケーションを実装します。

呼び出すタイミング

• 非同期 Web API（FastAPI、aiohttp、Sanic）を構築するとき
• 並行する I/O 操作（データベース、ファイル、ネットワーク）を実装するとき
• 並行リクエストを行う Web スクレイパーを作成するとき
• リアルタイムアプリケーション（WebSocket サーバー、チャットシステム）を開発するとき
• 独立した複数のタスクを同時に処理するとき
• 非同期通信を行うマイクロサービスを構築するとき
• I/O バウンドなワークロードを最適化するとき
• 非同期のバックグラウンドタスクやキューを実装するとき
• スレッド、マルチプロセス、asyncio のいずれを使うか判断するとき

基本概念

イベントループ

• 単一スレッドの協調型マルチタスク
• コルーチンの実行をスケジュールします
• ブロッキングせずに I/O 操作を処理します
• コールバックと Future を管理します

コルーチン

async def で定義され、実行を一時停止して再開できる関数です。

async def my_coroutine():
    result = await some_async_operation()
    return result

タスク

イベントループ上で並行して実行されるようスケジュールされたコルーチンです。

Future

非同期操作の最終的な結果を表す低レベルのオブジェクトです。

非同期コンテキストマネージャー

適切なクリーンアップのために async with をサポートするリソースです。

非同期イテレーター

非同期データソースを反復処理するために async for をサポートするオブジェクトです。

クイックスタート

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(main())

基本パターン

基本的な Async/Await

import asyncio

async def fetch_data(url: str) -> dict:
    await asyncio.sleep(1)  # Simulate I/O
    return {"url": url, "data": "result"}

async def main():
    result = await fetch_data("https://api.example.com")
    print(result)

asyncio.run(main())

gather() による並行実行

import asyncio
from typing import List

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.5)
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    tasks = [fetch_user(uid) for uid in user_ids]
    return await asyncio.gather(*tasks)

asyncio.run(fetch_all_users([1, 2, 3, 4, 5]))

タスクの作成と管理

import asyncio

async def background_task(name: str, delay: int):
    await asyncio.sleep(delay)
    return f"Result from {name}"

async def main():
    task1 = asyncio.create_task(background_task("Task 1", 2))
    task2 = asyncio.create_task(background_task("Task 2", 1))

    # Do other work while tasks run
    await asyncio.sleep(0.5)

    result1 = await task1
    result2 = await task2
    print(f"Results: {result1}, {result2}")

asyncio.run(main())

エラー処理

import asyncio
from typing import Optional

async def safe_operation(item_id: int) -> Optional[dict]:
    try:
        return await risky_operation(item_id)
    except ValueError as e:
        print(f"Error: {e}")
        return None

async def process_items(item_ids):
    tasks = [safe_operation(iid) for iid in item_ids]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successful = [r for r in results if r is not None and not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]
    return successful

タイムアウト処理

import asyncio

async def with_timeout():
    try:
        result = await asyncio.wait_for(slow_operation(5), timeout=2.0)
    except asyncio.TimeoutError:
        print("Operation timed out")

レート制限のための Semaphore

import asyncio
from typing import List

async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:
        await asyncio.sleep(0.5)  # Simulate API call
        return {"url": url, "status": 200}

async def rate_limited_requests(urls: List[str], max_concurrent: int = 5):
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [api_call(url, semaphore) for url in urls]
    return await asyncio.gather(*tasks)

選択の指針

asyncio を使う場合

• I/O バウンドなタスク（ネットワーク、データベース、ファイル）
• 多数の同時接続
• Web サーバーや API クライアント
• WebSocket アプリケーション

threading を使う場合

• 既存の同期ライブラリを使う I/O バウンドなタスク
• シンプルな並列処理が必要な場合
• GIL を解放する C 拡張との連携

multiprocessing を使う場合

• CPU バウンドなタスク（計算、データ処理）
• 真の並列処理が必要な場合（GIL を回避）
• 共有メモリを必要としないタスク

よくある落とし穴

await の付け忘れ

# Wrong - returns coroutine object
result = async_function()

# Correct
result = await async_function()

イベントループのブロッキング

# Wrong - blocks event loop
import time
async def bad():
    time.sleep(1)  # Blocks!

# Correct
async def good():
    await asyncio.sleep(1)  # Non-blocking

キャンセル処理の欠如

async def cancelable_task():
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        # Perform cleanup
        raise  # Re-raise to propagate

同期コードと非同期コードの混在

# Wrong
def sync_function():
    result = await async_function()  # SyntaxError!

# Correct
def sync_function():
    result = asyncio.run(async_function())

非同期コードのテスト

import asyncio
import pytest

@pytest.mark.asyncio
async def test_async_function():
    result = await fetch_data("https://api.example.com")
    assert result is not None

@pytest.mark.asyncio
async def test_with_timeout():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(5), timeout=1.0)

ベストプラクティス

1. エントリーポイントには asyncio.run() を使用する（Python 3.7 以降）
2. コルーチンを実行するには 必ず await する
3. 複数のタスクを並行実行するには gather() を使用する
4. try/except による 適切なエラー処理を実装する
5. 処理が停止したままになるのを防ぐために タイムアウトを使用する
6. パフォーマンス向上のために 接続をプールする
7. 非同期コードでは ブロッキング操作を避ける ― run_in_executor を使用する
8. レート制限には セマフォを使用する
9. タスクのキャンセルを適切に処理する
10. pytest-asyncio で 非同期コードをテストする

参考資料

• references/async-patterns.md - 非同期コンテキストマネージャー、非同期イテレーター／ジェネレーター、プロデューサー・コンシューマーパターン、非同期ロックと同期処理、aiohttp による Web スクレイピング、非同期データベース操作、WebSocket サーバーの実装、接続プール、バッチ処理、エグゼキューターでのブロッキング操作の実行

リソース

• Python asyncio ドキュメント: https://docs.python.org/3/library/asyncio.html
• aiohttp: 非同期 HTTP クライアント／サーバー
• FastAPI: モダンな非同期 Web フレームワーク
• asyncpg: 非同期 PostgreSQL ドライバー
• motor: 非同期 MongoDB ドライバー