python-async-ops

Python Async パターン

並行 Python プログラミングのための asyncio パターンです。

Async と Sync の使い分け

Async を使う場合	Sync を使う場合
I/O バウンドの操作（HTTP、DB、ファイル）	CPU バウンドの計算
高い並行性（数百以上の接続）	シンプルなスクリプト、単発のタスク
WebSocket／ストリーミング接続	小規模なデータ処理
ネットワーク呼び出しを伴うマイクロサービス	単一の逐次処理

判断フロー：

1. CPU バウンドですか？ → Sync（または multiprocessing）
2. 高い並行性を伴う I/O バウンドですか？ → Async
3. 接続数の少ないシンプルな I/O ですか？ → Sync で問題ありません

コアコンセプト

import asyncio

# Coroutine (must be awaited)
async def fetch(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

# Entry point
async def main():
    result = await fetch("https://example.com")
    return result

asyncio.run(main())

パターン 1：gather による並行処理

async def fetch_all(urls: list[str]) -> list[str]:
    """Fetch multiple URLs concurrently."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

パターン 2：並行性の制限

async def fetch_with_limit(urls: list[str], limit: int = 10):
    """Limit concurrent requests."""
    semaphore = asyncio.Semaphore(limit)

    async def bounded_fetch(url):
        async with semaphore:
            return await fetch_one(url)

    return await asyncio.gather(*[bounded_fetch(url) for url in urls])

パターン 3：TaskGroup（Python 3.11+）

async def process_items(items):
    """Structured concurrency with automatic cleanup."""
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(process_one(item))
    # All tasks complete here, or exception raised

パターン 4：タイムアウト

async def with_timeout():
    try:
        async with asyncio.timeout(5.0):  # Python 3.11+
            result = await slow_operation()
    except asyncio.TimeoutError:
        result = None
    return result

重要な注意事項

# WRONG - blocks event loop
async def bad():
    time.sleep(5)         # Never use time.sleep!
    requests.get(url)     # Blocking I/O!

# CORRECT
async def good():
    await asyncio.sleep(5)
    async with aiohttp.ClientSession() as s:
        await s.get(url)

# WRONG - orphaned task
async def bad():
    asyncio.create_task(work())  # May be garbage collected!

# CORRECT - keep reference
async def good():
    task = asyncio.create_task(work())
    await task

クイックリファレンス

パターン	用途
gather(*tasks)	複数の独立した操作
Semaphore(n)	レート制限、リソース制約
TaskGroup()	構造化並行処理（3.11+）
Queue()	プロデューサー／コンシューマー
timeout(s)	タイムアウトラッパー（3.11+）
Lock()	共有可変状態

Async コンテキストマネージャー

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_connection():
    conn = await create_connection()
    try:
        yield conn
    finally:
        await conn.close()

その他のリソース

詳細なパターンについては、以下を参照してください。

• ./references/concurrency-patterns.md - Queue、Lock、プロデューサー／コンシューマー
• ./references/aiohttp-patterns.md - HTTP クライアント／サーバーのパターン
• ./references/mixing-sync-async.md - run_in_executor、スレッドプール
• ./references/debugging-async.md - デバッグモード、プロファイリング、問題の特定
• ./references/production-patterns.md - グレースフルシャットダウン、ヘルスチェック、シグナル処理
• ./references/error-handling.md - バックオフ付きリトライ、サーキットブレーカー、部分的な失敗
• ./references/performance.md - uvloop、コネクションプーリング、バッファーサイズの設定

スクリプト

• ./scripts/find-blocking-calls.sh - Async 関数内のブロッキング呼び出しをコードから検索します

アセット

• ./assets/async-project-template.py - 本番環境向けの Async アプリのひな形

---

関連項目

前提条件：

• python-typing-ops - Async 関数の型ヒント

関連スキル：

• python-fastapi-ops - Async Web API
• python-observability-ops - Async ロギングとトレーシング
• python-database-ops - Async データベースアクセス