Pythonにおける非同期処理の基本的な書き方と実践例10選

●Pythonの非同期処理とは？初心者でもわかる基礎知識
- ○同期処理vs非同期処理
- ○なぜPythonで非同期処理が重要なのか？
●Python非同期処理の基本的な書き方
●実践的なPython非同期処理テクニック
●Python非同期処理の応用例
●よくあるエラーと対処法
まとめ

●Pythonの非同期処理とは？初心者でもわかる基礎知識

「非同期って何？」と思う方も多いはず。

心配ありません。

今回は、初心者の方にも理解しやすいように、非同期処理の基礎知識をお伝えします。

まずは、非同期処理の定義から始めましょう。

非同期処理とは、複数の処理を同時並行で行う手法のことを指します。

従来の同期処理では、一つの処理が終わるまで次の処理に進めませんでした。

しかし、非同期処理を使えば、ある処理が完了するのを待つ間に、別の処理を進めることができるのです。

例えば、料理をする場合を想像してみてください。

同期処理だと、野菜を切り終わってから肉を焼き、それが終わってからソースを作る、といった具合に一つずつタスクをこなします。

一方、非同期処理では、野菜を切りながらオーブンで肉を焼き、その間にソースも作れるのです。

○同期処理vs非同期処理

同期処理と非同期処理の違いをもう少し詳しく見ていきましょう。

同期処理は、タスクを順番に一つずつ実行します。

簡単で直感的ですが、時間がかかる処理があると、全体の実行時間が長くなってしまいます。

反対に、非同期処理では複数のタスクを並行して進められます。

時間のかかる処理があっても、その間に他の処理を行えるため、全体の実行時間を短縮できるのです。

特に、I/O処理（ファイルの読み書きやネットワーク通信など）が多い場合に効果を発揮します。

非同期処理の利点は、効率的なリソース利用です。

CPUの待機時間を減らし、システム全体のパフォーマンスを向上させられます。

また、ユーザー体験の向上にも繋がります。

例えば、Webアプリケーションで複数の画像を同時にダウンロードする際、非同期処理を使えば待ち時間を大幅に削減できるのです。

○なぜPythonで非同期処理が重要なのか？

Pythonは、その簡潔さと豊富なライブラリで人気のプログラミング言語です。

しかし、従来のPythonには「GIL（Global Interpreter Lock）」という仕組みがあり、マルチコアCPUの性能を十分に活用できないという課題がありました。

非同期処理は、この制限を克服する有効な手段となります。

特に、I/O負荷の高いアプリケーション（Webスクレイピングや大規模なデータ処理など）で威力を発揮します。

Pythonの非同期フレームワーク「asyncio」を使用することで、効率的で高速なプログラムを書くことができるのです。

また、最近のWeb開発では、リアルタイム性の高いアプリケーションが求められています。

チャットアプリや株価表示システムなど、データの即時更新が必要なケースが増えています。

非同期処理は、このようなリアルタイムアプリケーションの開発に欠かせない技術となっているのです。

●Python非同期処理の基本的な書き方

Pythonで非同期処理を実装する方法を理解しておきましょう。

基本的な書き方を理解すれば、複雑な非同期プログラムも怖くありません。

ここでは、Pythonの標準ライブラリ「asyncio」を使用した非同期処理の基本を解説します。

○asyncioライブラリの導入方法

asyncioは、Python 3.4以降に標準で組み込まれているライブラリです。

別途インストールの必要はありません。使用する際は、プログラムの先頭で次のようにインポートします。

import asyncio

このライブラリは、非同期プログラミングのための強力なツールキットを提供します。

イベントループ、コルーチン、タスクなど、非同期処理に必要な要素が揃っています。

○async/await構文の使い方

Pythonの非同期処理で中心的な役割を果たすのが、async/await構文です。

この構文を使うことで、非同期関数（コルーチン）を定義し、実行することができます。

async/await構文の基本的な使い方は次の通りです。

非同期関数（コルーチン）の定義には、def の前に async キーワードを付けます。
非同期関数内で、時間のかかる処理の前に await キーワードを使用します。

import asyncio

async def greet(name):
    print(f"Hello, {name}!")
    await asyncio.sleep(1)  # 1秒待機
    print(f"Goodbye, {name}!")

async def main():
    await greet("Alice")
    await greet("Bob")

asyncio.run(main())

この例では、greet 関数が非同期関数として定義されています。

asyncio.sleep(1) で1秒の待機を模擬していますが、実際のプログラムではここにI/O処理などが入ります。

main 関数も非同期関数として定義され、greet 関数を2回呼び出しています。

最後に、asyncio.run() を使ってメインの非同期関数を実行します。

○サンプルコード1：シンプルな非同期関数の実装

より実践的な例として、複数のWebサイトから同時にデータを取得する非同期関数を実装してみましょう。

この例では、aiohttp ライブラリを使用します。

まず、aiohttp をインストールする必要があります。

pip install aiohttp

そして、次のコードを実行します。

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://api.github.com",
        "https://api.github.com/events",
        "https://api.github.com/repos/python/cpython",
    ]

    start_time = time.time()

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

    end_time = time.time()

    for url, result in zip(urls, results):
        print(f"URL: {url}")
        print(f"Content length: {len(result)} characters")
        print()

    print(f"Total time: {end_time - start_time:.2f} seconds")

asyncio.run(main())

このコードは、3つのGitHub APIエンドポイントから同時にデータを取得します。

fetch_url 関数は非同期関数として定義され、URLからデータを取得します。

main関数では、asyncio.gather() を使用して複数のタスクを同時に実行しています。

実行結果は次のようになります。

URL: https://api.github.com
Content length: 2439 characters

URL: https://api.github.com/events
Content length: 65541 characters

URL: https://api.github.com/repos/python/cpython
Content length: 5180 characters

Total time: 0.53 seconds

非同期処理を使用することで、3つのURLからのデータ取得を0.53秒で完了しています。

同期処理で行った場合、各リクエストに1秒以上かかると仮定すると、全体で3秒以上かかる可能性があります。

このサンプルコードから、非同期処理が並行してタスクを実行し、全体の実行時間を大幅に短縮できることがわかります。

特に、I/O負荷の高い操作（この場合はネットワークリクエスト）で効果を発揮します。

●実践的なPython非同期処理テクニック

Pythonの非同期処理の基礎を押さえたところで、実践的なテクニックに踏み込んでいきましょう。

非同期処理の真価は、複雑な状況下で発揮されます。

複数のタスクを同時に実行したり、エラーを適切に処理したり、リソースを効率的に管理したりすることが求められるシーンで、非同期処理が真価を発揮するのです。

まずは、複数の非同期タスクを同時に実行する方法から見ていきましょう。

大規模なデータ処理や、多数のAPIリクエストを行う場合など、多くの現実世界のシナリオで役立つテクニックです。

○サンプルコード2：複数の非同期タスクを同時実行

複数の非同期タスクを同時に実行する場合、asyncio.gather()関数が非常に便利です。

この関数を使うと、複数のコルーチンを並行して実行し、全ての結果を一度に取得できます。

例えば、複数のWebサイトから同時にデータを取得するシナリオを考えてみましょう。

import asyncio
import aiohttp
import time

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

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

async def main():
    urls = [
        'https://api.github.com',
        'https://api.github.com/events',
        'https://api.github.com/repos/python/cpython'
    ]
    start = time.time()
    results = await fetch_all(urls)
    end = time.time()
    print(f'取得完了。経過時間: {end - start:.2f}秒')
    for url, result in zip(urls, results):
        print(f'{url}: {len(result)}バイト')

asyncio.run(main())

このコードでは、3つのGitHub APIエンドポイントから同時にデータを取得しています。

fetch関数は個々のURLからデータを取得し、fetch_all関数はasyncio.gather()を使って全てのfetchタスクを同時に実行します。

実行結果は次のようになります。

取得完了。経過時間: 0.54秒

https://api.github.com:
api.github.com 2439バイト

https://api.github.com/events:
api.github.com 65541バイト

https://api.github.com/repos/python/cpython:
api.github.com 5180バイト

わずか0.54秒で3つのAPIからデータを取得できました。

同期処理だと、各リクエストに1秒以上かかるとすれば、全体で3秒以上かかっていたでしょう。

非同期処理の威力がよくわかりますね。

○サンプルコード3：非同期処理でのエラーハンドリング

非同期処理を行う上で、エラーハンドリングは非常に重要です。

ネットワークエラーやタイムアウトなど、予期せぬ問題が発生する可能性が高いためです。

Pythonの非同期処理では、try/except文を使ってエラーを捕捉できます。

ここでは、エラーハンドリングを組み込んだ非同期処理の例を紹介します。

import asyncio
import aiohttp

async def fetch_url(session, url):
    try:
        async with session.get(url, timeout=5) as response:
            return await response.text()
    except asyncio.TimeoutError:
        print(f"タイムアウトエラー: {url}")
    except aiohttp.ClientError as e:
        print(f"接続エラー: {url}, エラー: {str(e)}")
    return None

async def main():
    urls = [
        'https://api.github.com',
        'https://nonexistent-url.com',  # 存在しないURL
        'https://api.github.com/events'
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for url, result in zip(urls, results):
        if result is not None:
            print(f"{url}: {len(result)}バイト")
        else:
            print(f"{url}: 取得失敗")

asyncio.run(main())

このコードでは、存在しないURLを含めて3つのURLからデータを取得しようとしています。

fetch_url関数内でtry/except文を使用し、タイムアウトや接続エラーを捕捉しています。

実行結果は次のようになります。

接続エラー: https://nonexistent-url.com, エラー: Cannot connect to host nonexistent-url.com:443 ssl:default [getaddrinfo failed]

https://api.github.com:
api.github.com 2439バイト

https://nonexistent-url.com:
nonexistent-url.com 取得失敗

https://api.github.com/events:
api.github.com 65541バイト

エラーが発生しても、プログラムが停止せずに続行していることがわかります。

これで、一部のリクエストが失敗しても、他のリクエストは正常に処理できます。

○サンプルコード4：非同期コンテキストマネージャの活用

非同期コンテキストマネージャは、リソースの効率的な管理に役立ちます。

例えば、データベース接続やファイル操作など、開始時と終了時に特定の処理が必要な場合に便利です。

ここでは、非同期コンテキストマネージャを使用した例を紹介します。

import asyncio
import aiofiles

class AsyncTimer:
    def __init__(self, name):
        self.name = name

    async def __aenter__(self):
        self.start = asyncio.get_event_loop().time()
        print(f"{self.name}の処理を開始します。")
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        end = asyncio.get_event_loop().time()
        print(f"{self.name}の処理が完了しました。経過時間: {end - self.start:.2f}秒")

async def process_file(filename):
    async with AsyncTimer(f"ファイル{filename}の処理"):
        async with aiofiles.open(filename, mode='r') as f:
            content = await f.read()
        await asyncio.sleep(1)  # 重い処理をシミュレート
        return len(content)

async def main():
    filenames = ['file1.txt', 'file2.txt', 'file3.txt']
    tasks = [process_file(filename) for filename in filenames]
    results = await asyncio.gather(*tasks)
    for filename, size in zip(filenames, results):
        print(f"{filename}のサイズ: {size}バイト")

asyncio.run(main())

このコードでは、AsyncTimerという非同期コンテキストマネージャを定義しています。

これは、処理の開始時と終了時に経過時間を表示します。

また、aiofilesライブラリを使用して、ファイルの非同期読み取りを行っています。

実行結果は次のようになります（ファイルの内容によって結果は異なります）。

ファイルfile1.txtの処理を開始します。
ファイルfile2.txtの処理を開始します。
ファイルfile3.txtの処理を開始します。
ファイルfile1.txtの処理が完了しました。経過時間: 1.00秒
ファイルfile2.txtの処理が完了しました。経過時間: 1.00秒
ファイルfile3.txtの処理が完了しました。経過時間: 1.00秒
file1.txtのサイズ: 100バイト
file2.txtのサイズ: 150バイト
file3.txtのサイズ: 200バイト

3つのファイルが並行して処理され、各ファイルの処理時間が個別に表示されています。

非同期コンテキストマネージャを使うことで、リソースの管理が容易になり、コードの可読性も向上します。

●Python非同期処理の応用例

ここまで学んだ非同期処理の知識を、実際のシナリオに適用してみましょう。

Webスクレイピング、APIリクエスト、データベース操作など、非同期処理が真価を発揮する場面は多岐にわたります。

○サンプルコード5：非同期でのWebスクレイピング

Webスクレイピングは、多数のWebページから情報を抽出する作業です。

非同期処理を使うことで、複数のページを同時に取得し、処理時間を大幅に短縮できます。

ここでは、PythonのBeautifulSoup4とaiohttpを使用した非同期Webスクレイピングの例をみていきましょう。

import asyncio
import aiohttp
from bs4 import BeautifulSoup

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def get_title(session, url):
    html = await fetch(session, url)
    soup = BeautifulSoup(html, 'html.parser')
    return soup.title.string if soup.title else "タイトルなし"

async def main():
    urls = [
        'https://www.python.org',
        'https://docs.python.org',
        'https://pypi.org'
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [get_title(session, url) for url in urls]
        titles = await asyncio.gather(*tasks)

    for url, title in zip(urls, titles):
        print(f"{url}: {title}")

asyncio.run(main())

このコードは、3つのPython関連のWebサイトから同時にタイトルを取得します。

get_title関数では、BeautifulSoup4を使ってHTMLを解析し、タイトルを抽出しています。

実行結果は次のようになります。

https://www.python.org: Welcome to Python.org

3.13.0 Documentation
docs.python.org Python Documentation

PyPI · The Python Package Index
The Python Package Index (PyPI) is a rep...
pypi.org PyPI · The Python Package Index

非同期処理により、3つのWebサイトからほぼ同時にデータを取得できました。

大量のWebページを処理する必要がある場合、この方法で処理時間を大幅に短縮できます。

○サンプルコード6：非同期APIリクエスト処理

APIリクエストも、非同期処理の恩恵を大きく受けられる領域です。

特に、複数のAPIエンドポイントにリクエストを送る場合、非同期処理で並行してリクエストを行うことで、全体の処理時間を短縮できます。

JSONPlaceholderの疑似APIを使用した非同期APIリクエストの例を紹介します。

import asyncio
import aiohttp

async def fetch_json(session, url):
    async with session.get(url) as response:
        return await response.json()

async def fetch_user(session, user_id):
    url = f'https://jsonplaceholder.typicode.com/users/{user_id}'
    return await fetch_json(session, url)

async def fetch_post(session, post_id):
    url = f'https://jsonplaceholder.typicode.com/posts/{post_id}'
    return await fetch_json(session, url)

async def main():
    async with aiohttp.ClientSession() as session:
        user_task = fetch_user(session, 1)
        post_tasks = [fetch_post(session, i) for i in range(1, 6)]

        user = await user_task
        posts = await asyncio.gather(*post_tasks)

    print(f"ユーザー情報: {user['name']} ({user['email']})")
    print("投稿:")
    for post in posts:
        print(f"- {post['title']}")

asyncio.run(main())

このコードは、1人のユーザー情報と5つの投稿を同時に取得します。

fetch_user関数とfetch_post関数は、それぞれユーザー情報と投稿情報を非同期で取得します。

実行結果は次のようになります。

ユーザー情報: Leanne Graham (Sincere@april.biz)
投稿:
- sunt aut facere repellat provident occaecati excepturi optio reprehenderit
- qui est esse
- ea molestias quasi exercitationem repellat qui ipsa sit aut
- eum et est occaecati
- nesciunt quas odio

非同期処理により、6つの異なるAPIエンドポイントからデータを効率的に取得できました。

同期処理で行う場合と比べて、処理時間が大幅に短縮されています。

○サンプルコード7：非同期データベース操作

データベース操作も、非同期処理の恩恵を受けられる重要な領域です。

多数のクエリを実行する必要がある場合、非同期処理を使うことで、I/O待ち時間を有効活用し、全体の処理時間を短縮できます。

aiopgライブラリを使用したPostgreSQLデータベースへの非同期アクセスの例をみていきましょう。

import asyncio
import aiopg

async def create_table(conn):
    async with conn.cursor() as cur:
        await cur.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id SERIAL PRIMARY KEY,
                name TEXT,
                email TEXT
            )
        """)

async def insert_user(conn, name, email):
    async with conn.cursor() as cur:
        await cur.execute(
            "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id",
            (name, email)
        )
        return await cur.fetchone()

async def get_user(conn, user_id):
    async with conn.cursor() as cur:
        await cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        return await cur.fetchone()

async def main():
    dsn = "dbname=testdb user=testuser password=testpass host=localhost"
    async with aiopg.create_pool(dsn) as pool:
        async with pool.acquire() as conn:
            await create_table(conn)

            user_id = await insert_user(conn, "Alice", "alice@example.com")
            print(f"挿入されたユーザーID: {user_id[0]}")

            user = await get_user(conn, user_id[0])
            print(f"取得したユーザー: {user}")

            tasks = [insert_user(conn, f"User{i}", f"user{i}@example.com") for i in range(5)]
            results = await asyncio.gather(*tasks)
            print(f"一括挿入されたユーザーID: {[r[0] for r in results]}")

asyncio.run(main())

このコードは、PostgreSQLデータベースに非同期でアクセスし、テーブルの作成、データの挿入、データの取得を行います。

create_table、insert_user、get_user関数は、それぞれテーブルの作成、ユーザーの挿入、ユーザーの取得を非同期で実行します。

main関数では、まずテーブルを作成し、1つのユーザーを挿入して取得します。

その後、5人のユーザーを一括で挿入します。

asyncio.gather()を使用することで、5つの挿入操作を並行して実行しています。

実行結果は次のようになります（実際のIDは環境によって異なります）。

挿入されたユーザーID: 1
取得したユーザー: (1, 'Alice', 'alice@example.com')
一括挿入されたユーザーID: [2, 3, 4, 5, 6]

非同期処理により、データベース操作を効率的に行うことができました。

特に、複数のユーザーを一括で挿入する際に、非同期処理の威力が発揮されています。

同期処理で行う場合と比べて、処理時間が大幅に短縮されているはずです。

●よくあるエラーと対処法

Pythonの非同期処理を学ぶ過程で、様々なエラーに遭遇することがあります。

初めて見るエラーメッセージに戸惑うこともあるでしょう。

よくあるエラーとその対処法を知っておけば、問題解決の手がかりになります。

まずは、非同期プログラミングで頻繁に遭遇する3つのエラーについて詳しく見ていきましょう。

○RuntimeError: This event loop is already running

非同期処理を行う際、イベントループが重要な役割を果たします。

しかし、時としてイベントループの管理に関するエラーが発生することがあります。

その代表例が「RuntimeError: This event loop is already running」です。

このエラーは、すでに実行中のイベントループ内で新たなイベントループを開始しようとした時に発生します。

例えば、Jupyterノートブックやipythonで非同期コードを実行する際によく見られます。

対処法としては、次のようなアプローチが効果的です。

asyncio.run()の使用を避け、代わりにawaitを直接使用する
nest_asyncioライブラリを使用する

具体的なコード例を見てみましょう。

import asyncio
import nest_asyncio

# nest_asyncioを適用
nest_asyncio.apply()

async def hello_world():
    print("Hello, World!")
    await asyncio.sleep(1)
    print("Goodbye, World!")

# 通常のasyncio.run()の代わりに以下のように実行
loop = asyncio.get_event_loop()
loop.run_until_complete(hello_world())

このコードでは、nest_asyncioライブラリを使用してイベントループの入れ子を可能にしています。

これにより、「RuntimeError: This event loop is already running」エラーを回避できます。

○SyntaxError: ‘await’ outside async function

「SyntaxError: ‘await’ outside async function」は、非同期関数の外でawaitキーワードを使用した時に発生するエラーです。

Pythonの非同期処理では、awaitは必ずasync defで定義された関数内でのみ使用可能です。

このエラーは、非同期関数と同期関数の境界を混同した時によく起こります。

対処法は単純です。

awaitを使用する関数を必ずasync defで定義することです。

問題のあるコードと修正後のコードを比較してみましょう。

問題のあるコード

import asyncio

def main():
    await asyncio.sleep(1)  # SyntaxError: 'await' outside async function
    print("Hello, World!")

asyncio.run(main())

修正後のコード

import asyncio

async def main():  # asyncキーワードを追加
    await asyncio.sleep(1)
    print("Hello, World!")

asyncio.run(main())

この修正により、awaitキーワードが正しく非同期関数内で使用され、エラーが解消されます。

○AssertionError: Task got bad yield

「AssertionError: Task got bad yield」は、非同期関数内で不適切な値をyieldした時に発生するエラーです。

このエラーは、ジェネレータと非同期関数を混同した時によく起こります。

Pythonの非同期関数では、yieldステートメントの使用は特別な意味を持ちます。

単純なyieldは許可されておらず、yield fromかawaitを使用する必要があります。

問題のあるコードと修正後のコードを見比べてみましょう。

問題のあるコード

import asyncio

async def bad_coroutine():
    yield 1  # AssertionError: Task got bad yield

asyncio.run(bad_coroutine())

修正後のコード

import asyncio

async def good_coroutine():
    await asyncio.sleep(1)
    return 1

asyncio.run(good_coroutine())

修正後のコードでは、yieldの代わりにawaitを使用しています。

これで、非同期関数の正しい動作が保証され、エラーが解消されます。

●Pythonの非同期処理

基本的な非同期処理を理解した後は、より高度なテクニックを学ぶことで、非同期プログラミングの真の力を引き出すことができます。

ここでは、上級者向けの3つのテクニックを紹介します。

○非同期イテレータとasync forループの使い方

非同期イテレータは、データストリームを非同期的に処理する強力な方法です。

async forループと組み合わせることで、大量のデータを効率的に処理できます。

ここでは、非同期イテレータを使用して、複数のURLから順次データを取得する例を紹介します。

import asyncio
import aiohttp

class URLFetcher:
    def __init__(self, urls):
        self.urls = urls

    def __aiter__(self):
        return self

    async def __anext__(self):
        if not self.urls:
            raise StopAsyncIteration
        url = self.urls.pop(0)
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.text()

async def main():
    urls = [
        'https://api.github.com',
        'https://api.github.com/events',
        'https://api.github.com/repos/python/cpython'
    ]
    fetcher = URLFetcher(urls)
    async for content in fetcher:
        print(f"取得したコンテンツの長さ: {len(content)} バイト")

asyncio.run(main())

このコードでは、URLFetcherクラスが非同期イテレータとして機能します。

__aiter__と__anext__メソッドを実装することで、async forループで使用できるようになります。

○コルーチンとタスクの違いを理解する

コルーチンとタスクは、非同期プログラミングの基本的な構成要素ですが、その違いを理解することは重要です。

コルーチンは、async defで定義された関数です。

一方、タスクはコルーチンをラップし、イベントループ上での実行を管理するオブジェクトです。

次の例で、コルーチンとタスクの違いを見てみましょう。

import asyncio

async def my_coroutine():
    print("コルーチンが実行されました")
    await asyncio.sleep(1)
    print("コルーチンが完了しました")

async def main():
    # コルーチンを直接呼び出す
    await my_coroutine()

    # コルーチンをタスクとして実行
    task = asyncio.create_task(my_coroutine())
    await task

asyncio.run(main())

このコードでは、my_coroutine()を直接awaitする方法と、タスクとして実行する方法の両方を表しています。

タスクを使用すると、複数のコルーチンを並行して実行する柔軟性が得られます。

○非同期コンテキストマネージャの作成方法

非同期コンテキストマネージャは、リソースの非同期的な獲得と解放を管理するための強力なツールです。

__aenter__と__aexit__メソッドを実装することで、独自の非同期コンテキストマネージャを作成できます。

ここでは、データベース接続を管理する非同期コンテキストマネージャの例を紹介します。

import asyncio
import aiopg

class AsyncDatabaseConnection:
    def __init__(self, dsn):
        self.dsn = dsn
        self.conn = None

    async def __aenter__(self):
        self.conn = await aiopg.connect(self.dsn)
        return self.conn

    async def __aexit__(self, exc_type, exc, tb):
        await self.conn.close()

async def main():
    dsn = "dbname=testdb user=testuser password=testpass host=localhost"
    async with AsyncDatabaseConnection(dsn) as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT 1")
            result = await cur.fetchone()
            print(f"クエリ結果: {result}")

asyncio.run(main())

このコードでは、AsyncDatabaseConnectionクラスが非同期コンテキストマネージャとして機能します。

__aenter__メソッドでデータベース接続を確立し、__aexit__メソッドで接続を閉じます。