Advanced Async & Await Patterns

This lesson dives beyond the basics. You'll learn event loop mechanics, tasks, concurrency patterns, async iterators, async generators, synchronization primitives, and real-world architectures used in production environments.

The Python async system is powered by the event loop, a scheduler that:

Basic structure:

import asyncio

async def main():
    print("Hello Async")

asyncio.run(main())

asyncio.run():

• Creates an event loop
• Starts running the coroutine
• Closes the loop after completion

Coroutines can call other coroutines using await:

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

async def process_data():
    data = await fetch_data()
    print("Processed:", data)

You can chain dozens of async functions without blocking the thread.

This is how you run coroutines in parallel (non-blocking):

async def a():
    await asyncio.sleep(1)
    return "A"

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

results = await asyncio.gather(a(), b())

Time taken: 1 second, not 2.

Used heavily in:

• Web scraping
• Parallel API calls
• Batch data processing

Tasks let coroutines run independently in the background:

async def background():
    while True:
        print("Running in background")
        await asyncio.sleep(3)

async def main():
    task = asyncio.create_task(background())
    await asyncio.sleep(10)
    task.cancel()

Perfect for:

• Polling loops
• Streams
• Websocket heartbeats
• Scheduled jobs

Different from gather(), wait() lets you specify:

done, pending = await asyncio.wait(
    {task1, task2, task3},
    return_when=asyncio.FIRST_COMPLETED
)

Used for:

• Racing multiple sources
• Timeout systems
• Picking fastest available API

async def fetch():
    await asyncio.sleep(5)
    return "done"

try:
    await asyncio.wait_for(fetch(), timeout=2)
except asyncio.TimeoutError:
    print("Timeout!")

Critical for resilient systems that depend on external APIs.

Used for resources that need async setup AND cleanup:

class AsyncResource:
    async def __aenter__(self):
        print("Opening")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        print("Closing")

async with AsyncResource() as r:
    print("Using resource")

Used for:

• Streams
• File IO wrappers
• Real-time data feeds

Example:

class AsyncCounter:
    def __init__(self):
        self.value = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.value >= 5:
            raise StopAsyncIteration
        await asyncio.sleep(1)
        self.value += 1
        return self.value

async for x in AsyncCounter():
    print(x)

These are perfect for streaming data where each item needs async fetching:

async def stream_numbers():
    for i in range(5):
        await asyncio.sleep(1)
        yield i

Consume it with:

async for n in stream_numbers():
    print(n)

Used in:

• Websocket data
• Real-time dashboards
• Server push notifications

Common pattern for decoupling work:

• AI data pipelines: Load data → Process → Save
• Web crawlers: Discover URLs → Fetch pages → Extract data
• Background jobs: Receive requests → Queue → Process

queue = asyncio.Queue()

async def producer():
    for i in range(5):
        await queue.put(i)

async def consumer():
    while True:
        item = await queue.get()
        print("Got:", item)
        queue.task_done()

AsyncIO = Concurrency, NOT CPU parallelism!

Choose the right tool:

• asyncio → IO-bound (network, files, database)
• multiprocessing → CPU-bound (math, ML, image processing)
• concurrent.futures → Mixed workloads

A professional engineer must know when to choose which.

Using aiohttp:

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as r:
            return await r.json()

This lets you send 1000 requests concurrently without blocking.

FastAPI, Starlette, aiohttp, Quart — all rely on:

• async routers
• async database sessions
• async locks for shared state
• async iterators for streaming responses

Example FastAPI stream:

async def event_stream():
    for i in range(100):
        yield f"data: {i}\n\n"
        await asyncio.sleep(0.1)

A real system might include:

• async file readers
• async DB connections
• async external API fetchers
• queues (producer/consumer)
• batch processors
• background tasks
• cancellation handlers

Async lets each component run without blocking any other.

Tasks can be cancelled, but you must handle the cancellation gracefully:

async def worker():
    try:
        while True:
            print("Working...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Worker cancelled safely")

# Cancelling:
task = asyncio.create_task(worker())
await asyncio.sleep(3)
task.cancel()
await task

Why this matters:

• ✔ graceful shutdowns
• ✔ timeouts
• ✔ web servers closing requests
• ✔ stopping background loops
• ✔ cleaning up resources

If you don't handle CancelledError, tasks become "dangling zombies" and corrupt state.

TaskGroups improve error handling and cancellation. If one task inside a group fails, the entire group is safely cancelled.

async with asyncio.TaskGroup() as tg:
    tg.create_task(fetch_user())
    tg.create_task(fetch_orders())

Benefits:

• ✔ predictable cancellation
• ✔ easier debugging
• ✔ no silent orphan tasks
• ✔ safer microservices
• ✔ recommended by Python core devs

TaskGroups will replace asyncio.gather() in most future architectures.

Async code introduces unique failure modes:

await asyncio.gather(a(), b(), c())

results = await asyncio.gather(a(), b(), c(), return_exceptions=True)

async def retry(coro, attempts=3):
    for _ in range(attempts):
        try:
            return await coro()
        except:
            await asyncio.sleep(1)

Deadlocks occur when tasks wait on each other incorrectly.

Example of a deadlock:

await lock.acquire()
await lock.acquire()   # never releases → deadlock

Proper pattern:

async with lock:
    ...

Async deadlocks usually come from:

• ✔ forgetting await
• ✔ acquiring multiple locks
• ✔ circular waiting
• ✔ blocking calls inside async code

Async queues give safe producer/consumer flow.

queue = asyncio.Queue()

async def producer():
    for i in range(100):
        await queue.put(i)

async def consumer():
    while True:
        item = await queue.get()
        print("Processing", item)
        queue.task_done()

Real uses:

• ✔ background job systems
• ✔ AI pipeline batching
• ✔ distributed crawlers
• ✔ video/audio frame processing
• ✔ websocket message routing

Queues stop you from overwhelming your system.

Example — streaming Bitcoin prices, logs, or live chat updates:

async def price_stream():
    while True:
        yield await fetch_price()
        await asyncio.sleep(1)

# Consumption:
async for price in price_stream():
    print(price)

This pattern powers:

• ✔ real-time dashboards
• ✔ monitoring systems
• ✔ live analytics
• ✔ trading bots

AsyncIO ≠ preemptive threading.

Your code must yield control to let other tasks run:

time.sleep(1)

await asyncio.sleep(1)

for i in range(10_000_000): ...

await asyncio.to_thread(cpu_heavy_func)

Sometimes you must call blocking code inside async systems:

result = await asyncio.to_thread(blocking_function)

Used for:

• ✔ image processing
• ✔ file compression
• ✔ pandas computations
• ✔ machine learning prediction (non-async)

to_thread() prevents freezing the event loop.

Running thousands of tasks at once can overload:

• ✔ network
• ✔ memory
• ✔ CPU
• ✔ database

Use semaphores:

sem = asyncio.Semaphore(5)

async def safe_fetch(url):
    async with sem:
        return await fetch(url)

Now only 5 requests run at once.

For CPU-heavy workloads (AI / ML / video processing), async alone is not enough.

Pattern:

Example:

loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, cpu_task)

Used in:

• ✔ video rendering
• ✔ neural network inference
• ✔ compression
• ✔ hashing
• ✔ scientific computing

This hybrid model is extremely powerful.

open("file.txt").read()

import aiofiles

async with aiofiles.open("big.txt") as f:
    async for line in f:
        ...

Used in:

• ✔ huge logs
• ✔ large dataset preprocessing
• ✔ AI training input pipelines

This is how websocket heartbeats, game loops, monitoring tasks work.

async def heartbeat():
    while True:
        send_ping()
        await asyncio.sleep(5)

# Restartable version:
async def resilient_heartbeat():
    while True:
        try:
            await heartbeat()
        except:
            await asyncio.sleep(1)

Used in:

• ✔ Discord bots
• ✔ IoT sensors
• ✔ distributed systems

Backpressure prevents fast producers from exploding memory.

if queue.qsize() > 1000:
    await asyncio.sleep(0.1)

Real-world use:

• ✔ Kafka consumers
• ✔ Redis Streams workers
• ✔ RabbitMQ consumers
• ✔ FastAPI streaming endpoints

async def fetch_with_retry(url, attempts=5):
    delay = 0.5

    for _ in range(attempts):
        try:
            return await fetch(url)
        except:
            await asyncio.sleep(delay)
            delay *= 2

This protects:

• ✔ ML data ingestion jobs
• ✔ microservice calls
• ✔ database queries
• ✔ message queues

Processing thousands of tasks at controlled speed:

async def bulk_process(urls, batch=50):
    for i in range(0, len(urls), batch):
        chunk = urls[i:i+batch]
        await asyncio.gather(*(fetch(u) for u in chunk))

Used in:

• ✔ bulk API ingestion
• ✔ scraping 10,000 pages
• ✔ ML dataset downloading
• ✔ parallel text generation

Caching async functions requires async-safe patterns — you cannot use normal functools.lru_cache on coroutines.

Basic async cache:

cache = {}

async def cached_fetch(key, func):
    if key in cache:
        return cache[key]
    result = await func()
    cache[key] = result
    return result

Time-based expiration:

import time

async_cache = {}

async def timed_cache(key, func, ttl=60):
    if key in async_cache:
        value, timestamp = async_cache[key]
        if time.time() - timestamp < ttl:
            return value

    value = await func()
    async_cache[key] = (value, time.time())
    return value

Real use cases:

• ✔ ML inference caching
• ✔ API rate-limit protection
• ✔ expensive SQL queries
• ✔ metadata caching
• ✔ reusable results in pipelines

Async LRU caching requires manual implementation:

from collections import OrderedDict

class AsyncLRU:
    def __init__(self, size=128):
        self.cache = OrderedDict()
        self.size = size

    async def get(self, key, func):
        if key in self.cache:
            self.cache.move_to_end(key)
            return self.cache[key]

        result = await func()
        self.cache[key] = result

        if len(self.cache) > self.size:
            self.cache.popitem(last=False)

        return result

Used in:

• ✔ ML preprocessing
• ✔ NLP tokenization
• ✔ thumbnail generation
• ✔ API microservices
• ✔ complex dependency graphs

A professional async pipeline processes streaming data in stages.

async def reader():
    async for line in aio_read_stream():
        yield line

async def cleaner(stream):
    async for line in stream:
        yield line.strip()

async def tokenizer(stream):
    async for line in stream:
        yield line.split()

async def run_pipeline():
    async for tokens in tokenizer(cleaner(reader())):
        print(tokens)

This powers:

• ✔ ETL pipelines
• ✔ log processors
• ✔ real-time analytics
• ✔ AI data loaders
• ✔ chat message routers

Each stage is async → memory safe → fully streaming.

Some tasks are IO-heavy, some are CPU-heavy.

Pattern:

Hybrid example:

async def transform_async(stream):
    async for item in stream:
        result = await asyncio.to_thread(cpu_heavy_compute, item)
        yield result

Used in:

• ✔ AI preprocessing
• ✔ video/audio decoding
• ✔ encryption
• ✔ hashing pipelines
• ✔ huge text transformations

import websockets
import asyncio

async def handler(ws):
    async for message in ws:
        await ws.send(f"Echo: {message}")

asyncio.run(websockets.serve(handler, "localhost", 9000))

Why this matters:

• ✔ chat apps
• ✔ live dashboards
• ✔ multiplayer games
• ✔ trading bots
• ✔ IoT communications

WebSockets are the backbone of real-time async systems.

Many real systems run supervised workers that restart when they fail.

async def supervise(coro):
    while True:
        try:
            await coro()
        except Exception as e:
            print("Restarting due to:", e)
            await asyncio.sleep(1)

Used in:

• ✔ monitoring daemons
• ✔ heartbeat services
• ✔ queue workers
• ✔ real-time scrapers
• ✔ Discord bots

This prevents silent crashes.

A robust pattern for distributed workers:

retry_queue = asyncio.Queue()

async def worker():
    while True:
        job = await retry_queue.get()
        try:
            await process(job)
        except:
            await asyncio.sleep(1)
            await retry_queue.put(job)

This ensures:

• ✔ jobs NEVER disappear
• ✔ failed jobs are retried
• ✔ system never overloads
• ✔ stable long-running services

Avoid DDoSing an API by accident.

Example with jitter:

import random

async def backoff_retry(func):
    delay = 1
    for _ in range(5):
        try:
            return await func()
        except:
            await asyncio.sleep(delay + random.random())
            delay *= 2

This is used in:

• ✔ AWS SDKs
• ✔ Google Cloud libraries
• ✔ Stripe API clients
• ✔ Redis cluster drivers

Sometimes you want a task to finish even if outer tasks are cancelled.

async with asyncio.shield(task):
    await task

Useful when:

• ✔ writing logs on shutdown
• ✔ saving ML model state
• ✔ finishing DB transactions
• ✔ flushing message queues

Professional testing:

@pytest.mark.asyncio
async def test_fetch():
    result = await fetch()
    assert result == 123

Mocking async functions:

async def fake_fetch():
    return {"ok": True}

monkeypatch.setattr(module, "fetch", fake_fetch)

Testing is essential for async correctness.

Pooling reduces overhead:

class Pool:
    async def acquire(self): ...
    async def release(self, item): ...

Used in:

• ✔ database connectors
• ✔ HTTP clients
• ✔ GPU memory managers
• ✔ ML model workers

Pools prevent resource exhaustion.

Real distributed systems use async clients:

• ✔ aioredis
• ✔ aiokafka
• ✔ aio_pika

Example — Kafka consumer:

async for msg in consumer:
    process(msg.value)

Used in:

• ✔ event-driven architecture
• ✔ streaming analytics
• ✔ ML log pipelines

A microservice might include:

• ✔ async router
• ✔ async DB layer
• ✔ async background workers
• ✔ async external API fetchers
• ✔ async timeouts & retries
• ✔ async signals
• ✔ async message queues

This is how modern companies scale to millions of users.

You now understand high-level async engineering concepts: