Advanced Async & Await

Reviewed & published by Brayan K

Asynchronous programming is at the heart of modern Python applications. Whether you're building high-performance web APIs, data pipelines, websocket apps, scrapers, or microservices — mastering advanced async/await patterns gives you the ability to build systems that scale effortlessly.

Part of the free Python course at LearnCodingFast — hands-on lessons with examples you run in your browser, plus practice exercises and a quick quiz.

What You'll Learn

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.

🔥 1. Understanding the Event Loop Deeply

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

import asyncio

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

asyncio.run(main())

# ✅ Expected output:
# Hello Async

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

⚡ 2. Coroutine Chaining & Composition

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.

⚙️ 3. Running Tasks Concurrently With asyncio.gather

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.

• Web scraping
• Parallel API calls
• Batch data processing

🧨 4. Creating Background Tasks With asyncio.create_task

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()

• Polling loops
• Websocket heartbeats
• Scheduled jobs

📖 Worked Example: await vs create_task

The snippets above show the pieces. This is the first complete, runnable program, and it settles the question people get wrong most often: does await make things concurrent? No. await waits. What makes things concurrent is starting the work before you wait for it — with create_task or gather.

The two halves do identical work. Watch the totals: 0.3 seconds versus 0.2 seconds, from moving two lines.

import asyncio
import time

async def step(name, seconds):
    """One unit of waiting work. await hands control back to the event loop."""
    await asyncio.sleep(seconds)          # NOT time.sleep — that would block everything
    print(f"  {name} finished after {seconds}s")
    return name.upper()

async def one_at_a_time():
    """Plain await runs them in order: each one starts only after the last ends."""
    start = time.perf_counter()
    a = await step("alpha", 0.2)          # nothing else can start until this returns
    b = await step("beta", 0.1)
    print("one_at_a_time:", [a, b], "took", round(time.perf_counter() - start, 1), "s")

async def overlapping():
    """create_task starts the coroutine straight away; await only collects it."""
    start = time.perf_counter()
    t1 = asyncio.create_task(step("alpha", 0.2))   # running from this line onwards
    t2 = asyncio.create_task(step("beta", 0.1))    # also running now
    a = await t1                                   # wait for whatever is left of t1
    b = await t2                                   # t2 already finished — returns instantly
    print("overlapping:  ", [a, b], "took", round(time.perf_counter() - start, 1), "s")

async def main():
    print("Sequential:")
    await one_at_a_time()
    print("Concurrent:")
    await overlapping()

asyncio.run(main())      # creates the loop, runs main(), closes the loop

# ✅ Expected output (timings may vary by a hundredth of a second):
# Sequential:
#   alpha finished after 0.2s
#   beta finished after 0.1s
# one_at_a_time: ['ALPHA', 'BETA'] took 0.3 s
# Concurrent:
#   beta finished after 0.1s
#   alpha finished after 0.2s
# overlapping:   ['ALPHA', 'BETA'] took 0.2 s

🎯 Your Turn: Price Three Suppliers at Once

Three suppliers, three different response times. Asking them one at a time would take 0.6 seconds; asking all three at once takes as long as the slowest one. Everything is written for you except the three async keywords. Replace each ___ and run it.

import asyncio
import time

# 🎯 YOUR TURN — replace the three ___ blanks

PRICES = {"apples": 1.20, "bread": 2.50, "milk": 0.99}

async def get_price(item, delay):
    """Pretend to ask a slow supplier API for one price."""
    ___ asyncio.sleep(delay)      # 👉 the keyword that pauses WITHOUT blocking others
    return PRICES[item]

async def main():
    start = time.perf_counter()

    # Ask all three suppliers at the same time instead of one after another.
    prices = await asyncio.___(   # 👉 which function runs several coroutines together?
        get_price("apples", 0.3),
        get_price("bread", 0.2),
        get_price("milk", 0.1),
    )

    print("Prices:", prices)
    print("Basket total:", round(sum(prices), 2))
    print("Elapsed:", round(time.perf_counter() - start, 1), "seconds")

asyncio.___(main())               # 👉 which function starts the event loop?

# ✅ Expected output:
# Prices: [1.2, 2.5, 0.99]
# Basket total: 4.69
# Elapsed: 0.3 seconds

1) await — await asyncio.sleep(delay).

2) gather — asyncio.gather(...) returns the results as a list, in the order you passed the coroutines, not the order they finished.

If your elapsed time is 0.6 seconds, you awaited each call separately instead of gathering them. If you see RuntimeWarning: coroutine 'get_price' was never awaited, you called it but never awaited or gathered the result.

🏆 Mini-Challenge: The Polite Crawler

Outline only — no filled-in logic. You have five pages to fetch, but the API you are hitting will ban you if you send more than two requests at a time. Unlimited concurrency is not the goal here; bounded concurrency is. An asyncio.Semaphore is the tool: it holds a fixed number of permits, and async with makes a coroutine queue up until one is free.

Five jobs, two at a time, means three rounds. Work out the expected total time before you run it — then check.

import asyncio
import time

# 🎯 MINI-CHALLENGE: fetch 5 pages, never more than 2 at once
# 1. Make a module-level sem = asyncio.Semaphore(2)
# 2. async def fetch(n):
#        async with sem:            <- waits here until a permit is free
#            await asyncio.sleep(0.2)
#            return n * n
# 3. In async def main(): record start = time.perf_counter()
# 4. results = await asyncio.gather(*(fetch(n) for n in range(1, 6)))
#    (the * unpacks the generator into five separate arguments)
# 5. Print "Results:" then the list, and the elapsed time rounded to 1 decimal
# 6. Start it with asyncio.run(main())
#
# ✅ Expected output:
# Results: [1, 4, 9, 16, 25]
# Elapsed: 0.6 seconds
#
# Why 0.6 and not 0.2? Five jobs, two permits: rounds of 2, 2 and 1.

# your code here

import asyncio
import time

sem = asyncio.Semaphore(2)          # two permits, so two fetches at a time

async def fetch(n):
    async with sem:                 # queue here until a permit frees up
        await asyncio.sleep(0.2)
        return n * n

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(*(fetch(n) for n in range(1, 6)))
    print("Results:", results)
    print("Elapsed:", round(time.perf_counter() - start, 1), "seconds")

asyncio.run(main())

Change the semaphore to Semaphore(5) and the elapsed time drops to about 0.2 seconds; change it to Semaphore(1) and you are back to 1.0 second — fully sequential. The semaphore is the dial between "polite" and "fast".

🧵 5. Using asyncio.wait() for Advanced Control

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

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

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

🌀 6. Timeouts With asyncio.wait_for

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.

🧱 7. Async Context Managers (async with)

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")

🌀 8. Async Iterators (async for)

• File IO wrappers
• Real-time data feeds

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)

🌊 9. Async Generators (async def … yield)

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

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

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

🔒 10. Async Locks, Semaphores & Synchronization

Prevents race conditions when multiple tasks access shared data:

lock = asyncio.Lock()

async with lock:
    # only one task enters here at a time
    ...

sem = asyncio.Semaphore(10)

async with sem:
    await fetch_api()

event = asyncio.Event()
event.set()  # resume waiting tasks

⚡ 11. Producer–Consumer Pattern (Async Pipeline)

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()

🧠 12. Parallelism vs Concurrency

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.

🌐 13. Combining Async With APIs (Practical)

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.

🔍 14. Async Patterns Used in Real Web Frameworks

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

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

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

🧩 15. Full Architecture Demo (Advanced)

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.

🔥 16. Understanding Task Cancellation in Depth

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

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

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

🧠 17. Task Groups (Python 3.11+) — Structured Concurrency

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())

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

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

⚙️ 18. Async Error Handling (Fail-Fast, Fail-Safe, Recovering)

Async code introduces unique failure modes:

Case 1 — fail fast (stop everything):

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

Case 2 — fail safe (continue, collect errors):

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

Case 3 — retry pattern:

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

Case 4 — circuit breaker (advanced):

Used in microservices to avoid hammering broken APIs.

🧵 19. Avoiding Deadlocks in Async Code

Deadlocks occur when tasks wait on each other incorrectly.

Example of a deadlock:

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

async with lock:
    ...

Async deadlocks usually come from:

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

🔒 20. Using Queues for Safe Concurrency

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()

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

Queues stop you from overwhelming your system.

📡 21. Building Real-Time Streams With Async Generators

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)

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

🧩 22. Understanding Cooperative Multitasking

Your code must yield control to let other tasks run:

❌ Bad (blocks event loop):

time.sleep(1)

await asyncio.sleep(1)

❌ Bad (CPU work blocks):

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

✅ Fix: push CPU work to executor:

await asyncio.to_thread(cpu_heavy_func)

⚗️ 23. Mixing AsyncIO With Sync Code (The Right Way)

Sometimes you must call blocking code inside async systems:

result = await asyncio.to_thread(blocking_function)

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

to_thread() prevents freezing the event loop.

📦 24. Handling Bounded Parallelism (Prevent Overload)

Running thousands of tasks at once can overload:

sem = asyncio.Semaphore(5)

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

Now only 5 requests run at once.

🧬 25. Combining AsyncIO With Multiprocessing

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

AsyncIO → handles network + coordination

ProcessPoolExecutor → handles CPU-heavy tasks

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

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

This hybrid model is extremely powerful.

🌍 26. Async File IO (aiofiles)

❌ Normal file IO blocks:

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

import aiofiles

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

• ✔ large dataset preprocessing
• ✔ AI training input pipelines

🔁 27. Restartable Background Loops

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)

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

🎛 28. Backpressure Techniques (Critical for Stability)

Backpressure prevents fast producers from exploding memory.

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

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

🧨 29. Async Retry Strategies With Exponential Backoff

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

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

🌐 30. Async Bulk Execution Pattern (High Throughput)

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))

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

🔥 31. Async Caching (In-Memory, Time-Based, and Function-Scoped)

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

cache = {}

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

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

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

⚡ 32. Async LRU Cache (Custom Implementation)

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

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

🌊 33. Async Stream Pipelines (End-to-End Processing)

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)

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

Each stage is async → memory safe → fully streaming.

🧩 34. Async + CPU Hybrid Pipelines

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

process pool → heavy CPU

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

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

📡 35. Async WebSockets — Real-Time Bi-Directional Streams

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))

• ✔ live dashboards
• ✔ multiplayer games
• ✔ IoT communications

WebSockets are the backbone of real-time async systems.

🧵 36. Async Task Supervision (Supervisor Pattern)

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)

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

This prevents silent crashes.

🔄 37. Async Retry Queues (Automatic Failure Recovery)

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)

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

🧠 38. Backoff, Jitter & Fail-Fast Patterns (Industry Standard)

Avoid DDoSing an API by accident.

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:

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

🧬 39. Async Cancellation Shields (protect tasks)

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

async with asyncio.shield(task):
    await task

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

🧪 40. Async Testing Patterns (pytest + asyncio)

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

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

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

Testing is essential for async correctness.

🧩 41. Async Resource Pools (DB, HTTP, GPU)

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

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

Pools prevent resource exhaustion.

🛠 42. Combining Async with Redis, Kafka, RabbitMQ

Real distributed systems use async clients:

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

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

🌐 43. Async Microservice Mesh (Framework-Level Example)

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.

🧠 44. Avoid These async/await Mistakes (Every Beginner Makes Them)

🎉 Conclusion

You now understand high-level async engineering concepts:

📋 Quick Reference — Async & Await

🎉 Great work! You've completed this lesson.

You now understand async/await patterns, how to structure concurrent code, and when to use asyncio vs threads.

Practice quiz

Continue this course

• Previous: Generators & Iterators Mastery
• Next: AsyncIO: Event Loop, Tasks & Futures — Deep dive into the AsyncIO event loop, tasks, and concurrent I/O
• Quick reference: Python cheat sheet