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

import asyncio

class AsyncConnection:
    def __init__(self, host):
        self.host = host
    
    async def __aenter__(self):
        print(f"{self.host} に接続中...")
        await asyncio.sleep(1)  # 接続をシミュレート
        print("接続完了")
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("切断中...")
        await asyncio.sleep(0.5)  # 切断をシミュレート
        print("切断完了")
    
    async def fetch(self, path):
        await asyncio.sleep(0.5)
        return f"Data from {self.host}{path}"

async def main():
    async with AsyncConnection("example.com") as conn:
        data = await conn.fetch("/api/data")
        print(data)

asyncio.run(main())

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_timer(label):
    start = asyncio.get_event_loop().time()
    print(f"{label} 開始")
    yield
    elapsed = asyncio.get_event_loop().time() - start
    print(f"{label} 完了: {elapsed:.2f}秒")

async def main():
    async with async_timer("データ取得"):
        await asyncio.sleep(1)

asyncio.run(main())

import asyncio
import aiofiles

async def write_file():
    async with aiofiles.open("output.txt", "w") as f:
        await f.write("非同期で書き込み")

async def read_file():
    async with aiofiles.open("output.txt", "r") as f:
        content = await f.read()
        print(content)

async def main():
    await write_file()
    await read_file()

asyncio.run(main())

import asyncio
import aiohttp

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

async def main():
    html = await fetch_url("https://example.com")
    print(f"取得: {len(html)} 文字")

asyncio.run(main())

import asyncio

lock = asyncio.Lock()

async def critical_section(name):
    async with lock:
        print(f"{name} がロック取得")
        await asyncio.sleep(1)
        print(f"{name} がロック解放")

async def main():
    await asyncio.gather(
        critical_section("タスクA"),
        critical_section("タスクB"),
    )

asyncio.run(main())

import asyncio
from contextlib import AsyncExitStack

@asynccontextmanager
async def resource(name):
    print(f"{name} 確保")
    yield name
    print(f"{name} 解放")

async def main():
    async with AsyncExitStack() as stack:
        r1 = await stack.enter_async_context(resource("A"))
        r2 = await stack.enter_async_context(resource("B"))
        print(f"使用中: {r1}, {r2}")

asyncio.run(main())