python-asyncio

クイックリファレンス

関数	目的	コード
asyncio.run()	エントリーポイント	asyncio.run(main())
asyncio.gather()	並行タスク	await asyncio.gather(*tasks)
asyncio.create_task()	起動して待機	task = asyncio.create_task(coro)
asyncio.TaskGroup()	構造化並行性	async with asyncio.TaskGroup() as tg:
asyncio.Semaphore()	レート制限	async with semaphore:
asyncio.timeout()	タイムアウト（3.11以降）	async with asyncio.timeout(5.0):

パターン	順次実行	並行実行
実行	await a(); await b()	await gather(a(), b())
時間	所要時間の合計	所要時間の最大値

ライブラリ	用途
aiohttp	HTTP クライアント（最も人気）
httpx	HTTP（同期＋非同期）
asyncpg	PostgreSQL
uvloop	イベントループを2～4倍高速化

このスキルを使用する場面

非同期／並行プログラミングで使用します：

• ネットワークリクエスト（HTTP、WebSocket）
• データベースクエリ
• ファイル I/O 操作
• 複数の I/O 操作の並行処理
• FastAPI／Starlette の非同期エンドポイント

関連スキル：

• FastAPI については、python-fastapiを参照してください
• 型ヒントについては、python-type-hintsを参照してください
• 注意点については、python-gotchasを参照してください

---

Python Asyncio 完全ガイド

概要

Asyncio は、async／await 構文を使って並行コードを記述するための Python 組み込みフレームワークです。ネットワークリクエスト、ファイル I/O、データベースクエリなど、I/O バウンドの操作に最適です。

Asyncio を使用する場面

適した用途

• ネットワークリクエスト（HTTP クライアント、WebSocket）
• データベースクエリ
• ファイル I/O 操作
• Web サーバー（FastAPI、Starlette）
• メッセージキュー
• 複数の I/O 操作の並行処理

使用しない場面

• CPU バウンドのタスク（代わりに multiprocessing を使用してください）
• 単純な逐次実行スクリプト
• 非同期処理に対応していないレガシーコードベース

コアコンセプト

基本的な Async/Await

import asyncio

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

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

# Entry point
asyncio.run(main())

gather() による並行実行

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return {"url": url, "status": response.status}

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        # Run all requests concurrently
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# Usage
urls = [
    "https://api.example.com/1",
    "https://api.example.com/2",
    "https://api.example.com/3",
]
results = asyncio.run(fetch_all(urls))

TaskGroup（Python 3.11以降）

import asyncio

async def process_item(item: str) -> str:
    await asyncio.sleep(1)
    return f"Processed: {item}"

async def process_all(items: list[str]) -> list[str]:
    results = []

    # TaskGroup provides structured concurrency
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(process_item(item)) for item in items]

    # All tasks complete when exiting the context
    return [task.result() for task in tasks]

# Exception handling with TaskGroup
async def process_with_errors(items: list[str]):
    try:
        async with asyncio.TaskGroup() as tg:
            for item in items:
                tg.create_task(process_item(item))
    except* ValueError as eg:
        # Handle ValueError exceptions
        for exc in eg.exceptions:
            print(f"ValueError: {exc}")
    except* TypeError as eg:
        # Handle TypeError exceptions
        for exc in eg.exceptions:
            print(f"TypeError: {exc}")

create_task() と await の比較

import asyncio

async def task_a():
    await asyncio.sleep(2)
    return "A done"

async def task_b():
    await asyncio.sleep(1)
    return "B done"

# Sequential (slower - 3 seconds total)
async def sequential():
    result_a = await task_a()  # Wait 2 seconds
    result_b = await task_b()  # Wait 1 second
    return result_a, result_b

# Concurrent (faster - 2 seconds total)
async def concurrent():
    task1 = asyncio.create_task(task_a())  # Start immediately
    task2 = asyncio.create_task(task_b())  # Start immediately
    result_a = await task1
    result_b = await task2
    return result_a, result_b

# Using gather (recommended for multiple tasks)
async def concurrent_gather():
    result_a, result_b = await asyncio.gather(task_a(), task_b())
    return result_a, result_b

高度なパターン

レート制限のためのセマフォ

import asyncio
import aiohttp

async def fetch_with_limit(
    session: aiohttp.ClientSession,
    url: str,
    semaphore: asyncio.Semaphore
) -> dict:
    async with semaphore:  # Limits concurrent requests
        async with session.get(url) as response:
            return await response.json()

async def fetch_many(urls: list[str], max_concurrent: int = 10) -> list[dict]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url, semaphore) for url in urls]
        return await asyncio.gather(*tasks)

タイムアウト

import asyncio

async def slow_operation():
    await asyncio.sleep(10)
    return "done"

async def with_timeout():
    try:
        # Wait at most 5 seconds
        result = await asyncio.wait_for(slow_operation(), timeout=5.0)
        return result
    except asyncio.TimeoutError:
        print("Operation timed out")
        return None

# Using timeout context manager (Python 3.11+)
async def with_timeout_context():
    async with asyncio.timeout(5.0):
        result = await slow_operation()
        return result

プロデューサー／コンシューマー用キュー

import asyncio
from typing import Any

async def producer(queue: asyncio.Queue, items: list[Any]):
    for item in items:
        await queue.put(item)
        print(f"Produced: {item}")
    # Signal completion
    await queue.put(None)

async def consumer(queue: asyncio.Queue, consumer_id: int):
    while True:
        item = await queue.get()
        if item is None:
            # Put sentinel back for other consumers
            await queue.put(None)
            break
        print(f"Consumer {consumer_id} processing: {item}")
        await asyncio.sleep(0.5)  # Simulate work
        queue.task_done()

async def main():
    queue: asyncio.Queue = asyncio.Queue(maxsize=10)

    items = list(range(20))

    # Start producer and consumers
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue, items))
        for i in range(3):
            tg.create_task(consumer(queue, i))

asyncio.run(main())

非同期ジェネレーター

import asyncio
from typing import AsyncIterator

async def fetch_pages(url: str, max_pages: int = 10) -> AsyncIterator[dict]:
    """Async generator for paginated API."""
    page = 1
    while page <= max_pages:
        # Simulate API call
        await asyncio.sleep(0.1)
        data = {"page": page, "items": [f"item_{i}" for i in range(10)]}

        if not data["items"]:
            break

        yield data
        page += 1

async def process_pages():
    async for page_data in fetch_pages("https://api.example.com"):
        print(f"Processing page {page_data['page']}")
        for item in page_data["items"]:
            process(item)

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

from contextlib import asynccontextmanager
from typing import AsyncIterator

class AsyncDatabaseConnection:
    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.disconnect()
        return False

    async def connect(self):
        print("Connecting...")
        await asyncio.sleep(0.1)

    async def disconnect(self):
        print("Disconnecting...")
        await asyncio.sleep(0.1)

# Using decorator
@asynccontextmanager
async def async_session() -> AsyncIterator[dict]:
    session = {"connected": True}
    try:
        yield session
    finally:
        session["connected"] = False
        await asyncio.sleep(0.1)  # Cleanup

# Usage
async def main():
    async with AsyncDatabaseConnection() as db:
        await db.query("SELECT * FROM users")

    async with async_session() as session:
        print(session)

パフォーマンスの最適化

Eager Task Factory（Python 3.12以降）

import asyncio

async def cached_operation(key: str) -> str:
    cache = {"a": "value_a", "b": "value_b"}
    if key in cache:
        return cache[key]  # Returns synchronously
    await asyncio.sleep(1)  # Only if cache miss
    return f"fetched_{key}"

async def main():
    loop = asyncio.get_event_loop()
    # Enable eager task execution
    loop.set_task_factory(asyncio.eager_task_factory)

    # Cached operations complete synchronously without event loop overhead
    result = await cached_operation("a")

uvloop の使用

# Install: pip install uvloop

import asyncio

try:
    import uvloop
    # 2-4x performance improvement
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except ImportError:
    pass  # Fall back to default event loop

async def main():
    # Your async code here
    pass

asyncio.run(main())

フリースレッド asyncio（Python 3.14以降）

# Python 3.14 improvements for free-threaded builds:
# - Thread-safe asyncio with lock-free data structures
# - Linear scaling with number of threads
# - 10-20% single-threaded performance improvement
# - Reduced memory usage

import asyncio
import threading

async def per_thread_loop():
    """Each thread can run its own event loop."""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        await asyncio.sleep(1)
    finally:
        loop.close()

# Multiple event loops in parallel (free-threaded build)
threads = [
    threading.Thread(target=lambda: asyncio.run(per_thread_loop()))
    for _ in range(4)
]
for t in threads:
    t.start()
for t in threads:
    t.join()

よくある注意点

イベントループをブロックする

import asyncio
import time

# BAD: Blocks the event loop
async def bad_example():
    time.sleep(5)  # Blocks everything!
    return "done"

# GOOD: Use async sleep
async def good_example():
    await asyncio.sleep(5)
    return "done"

# GOOD: Run blocking code in executor
async def blocking_in_executor():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, time.sleep, 5)
    return result

# For CPU-bound work, use ProcessPoolExecutor
from concurrent.futures import ProcessPoolExecutor

async def cpu_bound_work(data: list) -> list:
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, heavy_computation, data)
    return result

誤ったコンテキストでタスクを作成する

import asyncio

# BAD: Task created outside async context
# task = asyncio.create_task(some_coroutine())  # RuntimeError!

# GOOD: Create tasks inside async function
async def main():
    task = asyncio.create_task(some_coroutine())
    await task

asyncio.run(main())

await の付け忘れ

import asyncio

async def fetch():
    await asyncio.sleep(1)
    return "data"

# BAD: Coroutine never executed
async def bad():
    result = fetch()  # Just creates coroutine object!
    print(result)     # Prints coroutine object, not "data"

# GOOD: Always await coroutines
async def good():
    result = await fetch()
    print(result)  # Prints "data"

タスク内の例外処理

import asyncio

async def failing_task():
    await asyncio.sleep(1)
    raise ValueError("Task failed!")

async def main():
    # BAD: Exception silently lost
    task = asyncio.create_task(failing_task())
    await asyncio.sleep(2)  # Task exception ignored

    # GOOD: Always await tasks or use TaskGroup
    task = asyncio.create_task(failing_task())
    try:
        await task
    except ValueError as e:
        print(f"Caught: {e}")

    # BEST: Use TaskGroup (Python 3.11+)
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(failing_task())
    except* ValueError as eg:
        for exc in eg.exceptions:
            print(f"Caught: {exc}")

非同期ライブラリ

HTTP クライアント

# aiohttp - Most popular
import aiohttp

async def fetch_aiohttp(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

# httpx - Supports both sync and async
import httpx

async def fetch_httpx(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

データベース

# asyncpg - PostgreSQL
import asyncpg

async def query_postgres():
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    rows = await conn.fetch("SELECT * FROM users")
    await conn.close()
    return rows

# aiosqlite - SQLite
import aiosqlite

async def query_sqlite():
    async with aiosqlite.connect("database.db") as db:
        async with db.execute("SELECT * FROM users") as cursor:
            return await cursor.fetchall()

Web フレームワーク

# FastAPI (built on Starlette)
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# Starlette
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    return JSONResponse({"hello": "world"})

app = Starlette(routes=[Route("/", homepage)])

追加リファレンス

このガイドの範囲を超える、本番環境向けのパターンについては、以下を参照してください。

• 非同期パターンライブラリ - トークンバケット型レートリミッター、指数バックオフを使ったリトライ、コネクションプール、バッチプロセッサー、イベントバス、トランザクションコンテキストマネージャー、TTL 付き非同期キャッシュ、グレースフルシャットダウンハンドラー