python-async-patterns

import asyncio
import aiohttp

async def main() -> None:
    # --- your async logic here ---
    pass

if __name__ == "__main__":
    asyncio.run(main())          # Python 3.7+ preferred entry point

import asyncio

async def greet(name: str) -> str:
    await asyncio.sleep(0.1)   # yields control to the event loop
    return f"Hello, {name}"

async def main():
    # Await directly — simplest form
    result = await greet("world")

    # Schedule as a background Task
    task = asyncio.create_task(greet("background"))
    # ... do other work concurrently ...
    result = await task           # get result / propagate exception

async def main():
    # All three run concurrently — total time ≈ max(each), not sum
    a, b, c = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )

# Prevent one failure from killing all — return_exceptions=True
results = await asyncio.gather(
    risky_task(1),
    risky_task(2),
    risky_task(3),
    return_exceptions=True,        # exceptions become values, not raises
)
for r in results:
    if isinstance(r, Exception):
        print(f"Task failed: {r}")
    else:
        process(r)

async def main():
    async with asyncio.TaskGroup() as tg:
        task_a = tg.create_task(fetch_user(1))
        task_b = tg.create_task(fetch_user(2))
        task_c = tg.create_task(fetch_user(3))
    # All tasks guaranteed to finish before this line

    print(task_a.result(), task_b.result(), task_c.result())

# Handle multiple exception types with except*
try:
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(risky(i)) for i in range(10)]
except* ValueError as eg:
    for exc in eg.exceptions:
        print(f"ValueError: {exc}")

import asyncio, aiohttp

async def fetch(session, url: str, sem: asyncio.Semaphore) -> dict:
    async with sem:                    # blocks if limit reached
        async with session.get(url) as resp:
            return await resp.json()

async def fetch_all(urls: list[str], concurrency: int = 10) -> list[dict]:
    sem = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, sem) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

import asyncio

async def producer(queue: asyncio.Queue, items: list):
    for item in items:
        await queue.put(item)
    await queue.put(None)              # sentinel to signal done

async def consumer(queue: asyncio.Queue, worker_id: int):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)      # re-broadcast sentinel
            break
        print(f"Worker {worker_id} processed {item}")
        queue.task_done()

async def main():
    queue: asyncio.Queue = asyncio.Queue(maxsize=50)  # backpressure at 50 items
    NUM_WORKERS = 5

    await asyncio.gather(
        producer(queue, range(100)),
        *[consumer(queue, i) for i in range(NUM_WORKERS)],
    )

import asyncio

# --- asyncio.timeout() — Python 3.11+ (preferred) ---
async def main():
    try:
        async with asyncio.timeout(10.0):
            result = await slow_operation()
    except TimeoutError:
        print("operation timed out")

# --- asyncio.wait_for() — all versions ---
try:
    result = await asyncio.wait_for(slow_operation(), timeout=10.0)
except asyncio.TimeoutError:
    print("timed out")

# --- Graceful cleanup in a coroutine ---
async def worker():
    try:
        while True:
            await do_work()
    except asyncio.CancelledError:
        await cleanup()            # run cleanup BEFORE re-raising
        raise                      # always re-raise CancelledError

# --- Shield: protect critical work from external cancellation ---
async def main():
    task = asyncio.create_task(important_cleanup())
    try:
        await asyncio.shield(task)
    except asyncio.CancelledError:
        await task                 # still wait for cleanup to finish

import aiohttp, asyncio

# ❌ WRONG — new session per call, leaks connections
async def fetch_wrong(url):
    async with aiohttp.ClientSession() as session:
        ...

# ✓ CORRECT — shared session injected into each call
async def fetch(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()

async def main(urls: list[str]):
    timeout = aiohttp.ClientTimeout(total=30, connect=5)
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)

    async with aiohttp.ClientSession(
        timeout=timeout,
        connector=connector,
        raise_for_status=True,
    ) as session:
        results = await asyncio.gather(*[fetch(session, u) for u in urls])

async def fetch_with_retry(
    session: aiohttp.ClientSession,
    url: str,
    retries: int = 3,
    backoff: float = 1.0,
) -> dict:
    for attempt in range(retries):
        try:
            async with session.get(url) as resp:
                resp.raise_for_status()
                return await resp.json()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == retries - 1:
                raise
            wait = backoff * (2 ** attempt)
            await asyncio.sleep(wait)

from aiohttp import web
import time, logging

# --- Request handlers ---
async def handle_get(request: web.Request) -> web.Response:
    name = request.match_info.get("name", "World")
    return web.json_response({"message": f"Hello, {name}"})

async def handle_post(request: web.Request) -> web.Response:
    body = await request.json()
    return web.json_response({"received": body}, status=201)

# --- Middleware: logging ---
@web.middleware
async def logging_middleware(request: web.Request, handler) -> web.Response:
    start = time.monotonic()
    response = await handler(request)
    elapsed = time.monotonic() - start
    logging.info(f"{request.method} {request.path} → {response.status} ({elapsed:.3f}s)")
    return response

# --- App factory ---
def create_app() -> web.Application:
    app = web.Application(middlewares=[logging_middleware])
    app.router.add_get("/hello/{name}", handle_get)
    app.router.add_post("/items", handle_post)
    return app

if __name__ == "__main__":
    web.run_app(create_app(), host="0.0.0.0", port=8080)

from aiohttp import web
import aiohttp

clients: set[web.WebSocketResponse] = set()

async def ws_handler(request: web.Request) -> web.WebSocketResponse:
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    clients.add(ws)
    try:
        async for msg in ws:
            if msg.type == aiohttp.WSMsgType.TEXT:
                # broadcast to all other connected clients
                for client in clients - {ws}:
                    if not client.closed:
                        await client.send_str(msg.data)
    finally:
        clients.discard(ws)
    return ws

import asyncio
from contextlib import asynccontextmanager

# --- Async generator: paginated API ---
async def paginated_fetch(base_url: str):
    page = 1
    while True:
        data = await fetch_page(base_url, page)
        if not data:
            return
        for item in data:
            yield item             # one item at a time
        page += 1

# Consume
async def main():
    async for item in paginated_fetch("https://api.example.com/items"):
        process(item)

    # Async comprehension
    items = [item async for item in paginated_fetch("https://api.example.com/items")]

# --- Async context manager ---
@asynccontextmanager
async def managed_resource(name: str):
    resource = await acquire(name)
    try:
        yield resource
    finally:
        await release(resource)   # guaranteed cleanup

async with managed_resource("db") as db:
    await db.query("SELECT 1")

import asyncio, functools

# Sync → Async: offload blocking function to thread
async def main():
    result = await asyncio.to_thread(blocking_function, arg1, arg2)

    # With keyword args
    fn = functools.partial(blocking_function, key="value")
    result = await asyncio.to_thread(fn)

    # CPU-bound: use ProcessPoolExecutor
    from concurrent.futures import ProcessPoolExecutor
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, heavy_cpu_func, data)

# Async → Sync: call from regular function
def sync_caller():
    result = asyncio.run(my_async_function())   # creates + runs + closes loop
    return result

# Cross-thread: schedule on running loop from another thread
future = asyncio.run_coroutine_threadsafe(my_coro(), existing_loop)
result = future.result(timeout=30)

import asyncio, logging

# Enable debug mode (warns on slow callbacks > 100ms, logs unawaited coroutines)
logging.basicConfig(level=logging.DEBUG)
asyncio.run(main(), debug=True)
# Or via env: PYTHONASYNCIODEBUG=1 python myscript.py

# Inspect all running tasks
async def print_tasks():
    for task in asyncio.all_tasks():
        print(f"  {task.get_name()}: {task.get_coro().__name__}")
        task.print_stack()

# Loop blocking watchdog — warns if event loop blocked > 100ms
def loop_watchdog(loop: asyncio.AbstractEventLoop, threshold: float = 0.1):
    import time
    last_tick = time.monotonic()

    def on_tick():
        nonlocal last_tick
        now = time.monotonic()
        blocked = now - last_tick
        if blocked > threshold:
            print(f"WARNING: event loop blocked for {blocked:.3f}s")
        last_tick = now
        loop.call_later(0.05, on_tick)

    loop.call_soon(on_tick)

pip install pytest-asyncio

# Test
@pytest.mark.asyncio
async def test_fetch():
    async with aiohttp.ClientSession() as session:
        result = await fetch(session, "https://httpbin.org/get")
    assert result["url"] == "https://httpbin.org/get"

# Async fixture
@pytest_asyncio.fixture
async def client_session():
    async with aiohttp.ClientSession() as session:
        yield session