async-python-patterns

非同期 Python パターン

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

このスキルを使う場面

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

基本パターン

1. 基本的な Async/Await

すべての非同期処理の基礎：

import asyncio

async def fetch_data(url: str) -> dict:
    """Fetch data asynchronously."""
    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())

主な概念：

• async def はコルーチン（一時停止できる関数）を定義します
• await はイベントループに制御を戻します
• asyncio.run() はエントリーポイントです（Python 3.7 以降）
• 単一スレッドの協調的マルチタスク

2. 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]:
    """Fetch multiple users concurrently."""
    tasks = [fetch_user(uid) for uid in user_ids]
    results = await asyncio.gather(*tasks)
    return results

# Speed: Sequential = 5s, Concurrent = 0.5s for 10 users

使用する場面：

• 並列実行できる独立した処理
• I/O バウンドのタスク（API 呼び出し、データベースクエリ）
• 次に進む前にすべての結果が必要な場合
• 部分的な失敗を処理するには return_exceptions=True を使用します

3. タスクの作成と管理

独立して実行されるバックグラウンドタスク：

import asyncio

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

async def main():
    # Create tasks (starts execution immediately)
    task1 = asyncio.create_task(background_task("Task 1", 2))
    task2 = asyncio.create_task(background_task("Task 2", 1))

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

    # Wait for results when needed
    result1, result2 = await task1, await task2

違い：

• await coroutine() - すぐに待機します
• asyncio.create_task() - バックグラウンド実行を開始します
• task.cancel() でタスクをキャンセルできます

4. エラー処理

並行処理の堅牢なエラー処理：

import asyncio
from typing import List, Optional

async def safe_operation(item_id: int) -> Optional[dict]:
    try:
        await asyncio.sleep(0.1)
        if item_id % 3 == 0:
            raise ValueError(f"Item {item_id} failed")
        return {"id": item_id, "status": "success"}
    except ValueError as e:
        print(f"Error: {e}")
        return None

async def process_items(item_ids: List[int]):
    # gather with return_exceptions=True continues on errors
    results = await asyncio.gather(
        *[safe_operation(iid) for iid in item_ids],
        return_exceptions=True
    )

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

    print(f"Success: {len(successful)}, Failed: {len(failed)}")
    return successful

5. タイムアウト処理

処理が無期限に停止するのを防ぎます：

import asyncio

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

高度なパターン

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

非同期処理で適切にリソースを管理します：

import asyncio

class AsyncDatabaseConnection:
    """Async database connection with automatic cleanup."""

    def __init__(self, dsn: str):
        self.dsn = dsn
        self.connection = None

    async def __aenter__(self):
        print("Opening connection")
        await asyncio.sleep(0.1)  # Simulate connection
        self.connection = {"dsn": self.dsn, "connected": True}
        return self.connection

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing connection")
        await asyncio.sleep(0.1)  # Simulate cleanup
        self.connection = None

async def query_database():
    async with AsyncDatabaseConnection("postgresql://localhost") as conn:
        # Connection automatically closed on exit
        return await perform_query(conn)

使用例：

• データベース接続（asyncpg、motor）
• HTTP セッション（aiohttp.ClientSession）
• ファイル I/O（aiofiles）
• ロックとセマフォ

7. 非同期イテレーターとジェネレーター

データを非同期にストリーム処理します：

import asyncio
from typing import AsyncIterator

async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]:
    """Fetch paginated data lazily."""
    for page in range(1, max_pages + 1):
        await asyncio.sleep(0.2)  # API call
        yield {
            "page": page,
            "url": f"{url}?page={page}",
            "data": [f"item_{page}_{i}" for i in range(5)]
        }

async def process_stream():
    async for page_data in fetch_pages("https://api.example.com", 10):
        # Process each page as it arrives (memory efficient)
        print(f"Processing page {page_data['page']}")

利点：

• 大規模なデータセットでもメモリー効率が高い
• すべてのデータが届く前に処理を開始できます
• バックプレッシャーを自然に処理できます

8. キューを使ったプロデューサー・コンシューマー

プロデューサーとコンシューマー間で作業を調整します：

import asyncio
from asyncio import Queue

async def producer(queue: Queue, producer_id: int, num_items: int):
    for i in range(num_items):
        item = f"Item-{producer_id}-{i}"
        await queue.put(item)
        await asyncio.sleep(0.1)
    await queue.put(None)  # Signal completion

async def consumer(queue: Queue, consumer_id: int):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break

        print(f"Consumer {consumer_id} processing: {item}")
        await asyncio.sleep(0.2)
        queue.task_done()

async def run_pipeline():
    queue = Queue(maxsize=10)

    # 2 producers, 3 consumers
    producers = [asyncio.create_task(producer(queue, i, 5)) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, i)) for i in range(3)]

    await asyncio.gather(*producers)
    await queue.join()  # Wait for all items processed

    for c in consumers:
        c.cancel()

9. セマフォによるレート制限

並行処理を制御します：

import asyncio
from typing import List

async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:  # Only N operations at once
        print(f"Calling {url}")
        await asyncio.sleep(0.5)
        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)

# Limits to 5 concurrent requests regardless of total URLs

使用例：

• API のレート制限（API のクォータを遵守）
• データベース接続数の制限
• ファイルディスクリプター数の制限
• メモリーに制約のある処理

10. 共有状態のための非同期ロック

非同期コンテキストでのスレッドセーフな処理：

import asyncio

class AsyncCounter:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()

    async def increment(self):
        async with self.lock:
            current = self.value
            await asyncio.sleep(0.01)  # Simulate work
            self.value = current + 1

    async def get_value(self) -> int:
        async with self.lock:
            return self.value

同期プリミティブ：

• Lock：相互排他
• Event：タスク間のシグナル
• Condition：条件を待機
• Semaphore：同時アクセスを制限

パフォーマンスのベストプラクティス

1. 接続プールを使用する

効率化のために接続を再利用します：

import aiohttp

async def with_connection_pool():
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [session.get(f"https://api.example.com/item/{i}")
                 for i in range(50)]
        return await asyncio.gather(*tasks)

2. イベントループをブロックしない

CPU 負荷の高い処理はエグゼキューターで実行します：

import asyncio
import concurrent.futures

def blocking_operation(data):
    """CPU-intensive blocking operation."""
    import time
    time.sleep(1)
    return data * 2

async def run_in_executor(data):
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_operation, data)
        return result

避けるべき一般的なブロッキング処理：

• time.sleep() - asyncio.sleep() を使用します
• 同期ファイル I/O - aiofiles を使用します
• 同期 HTTP - aiohttp または httpx を使用します
• 重い計算 - loop.run_in_executor() を使用します

3. 処理をバッチ化する

メモリーを制御するため、チャンク単位で処理します：

async def batch_process(items: List[str], batch_size: int = 10):
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        results = await asyncio.gather(*[process_item(item) for item in batch])
        print(f"Processed batch {i // batch_size + 1}")

よくある落とし穴

1. await の付け忘れ

# Wrong - returns coroutine, doesn't execute
result = async_function()

# Correct
result = await async_function()

2. イベントループのブロック

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

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

3. キャンセル処理の未対応

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

4. 同期処理と非同期処理の混在

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

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

非同期コードのテスト

テストには pytest-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)

リソース

• Python asyncio のドキュメント：https://docs.python.org/3/library/asyncio.html
• aiohttp：非同期 HTTP クライアント／サーバーフレームワーク
• FastAPI：OpenAPI を自動生成するモダンな非同期 Web フレームワーク
• asyncpg：高性能な非同期 PostgreSQL ドライバー
• motor：Python 用の非同期 MongoDB ドライバー
• pytest-asyncio：非同期コードのテストフレームワーク

ベストプラクティスのまとめ

1. エントリーポイントにはasyncio.run()を使用します（Python 3.7以降）。
2. コルーチンを実行するには、必ずawaitします。
3. 独立したタスクを並行実行するには、gather()を使用します。
4. try/exceptとreturn_exceptions=Trueを使って、適切なエラー処理を実装します。
5. 処理がハングするのを防ぐため、タイムアウトを設定します。
6. パフォーマンスとリソース管理を改善するには、接続をプールします。
7. 非同期コードではブロッキング処理を避けます（必要に応じてエグゼキューターを使用します）。
8. レート制限とリソース制御には、セマフォを使用します。
9. CancelledErrorを使って、タスクのキャンセルを適切に処理します。
10. pytest-asyncioを使って、非同期コードを十分にテストします。