Async/Await & Synchronous vs Asynchronous

Asynchronous programming lets Python handle thousands of I/O operations concurrently — without threads, without blocking, and without the GIL headache.

Synchronous vs Asynchronous: The Big Picture

The Problem With Synchronous I/O

import time

def fetch_user(user_id):
    """Synchronous: blocks the entire program while waiting."""
    print(f"Fetching user {user_id}...")
    time.sleep(1)  # Simulates waiting for a database/API
    return {"id": user_id, "name": f"User_{user_id}"}

def fetch_post(post_id):
    print(f"Fetching post {post_id}...")
    time.sleep(1)
    return {"id": post_id, "title": f"Post_{post_id}"}

# Problem: Each call BLOCKS — total time = sum of all waits
start = time.time()

user = fetch_user(1)
post = fetch_post(42)

print(f"\nUser: {user}")
print(f"Post: {post}")
print(f"Total time: {time.time() - start:.2f}s")  # ~2 seconds
print("With 100 requests this would take ~100 seconds!")

Output:

Fetching user 1...
Fetching post 42...

User: {'id': 1, 'name': 'User_1'}
Post: {'id': 42, 'title': 'Post_42'}
Total time: 2.00s
With 100 requests this would take ~100 seconds!

---

The asyncio Event Loop

The event loop is the engine of async Python. It manages coroutines, schedules their execution, and handles I/O callbacks.

Your First Coroutine

import asyncio

# A coroutine is defined with "async def"
async def greet(name, delay):
    """A coroutine that pauses non-blocking."""
    print(f"Hello, {name}! (starting)")
    await asyncio.sleep(delay)  # Non-blocking pause
    print(f"Hello again, {name}! (after {delay}s)")
    return f"Done greeting {name}"

async def main():
    # Call a coroutine — must use await
    result = await greet("Alice", 1)
    print(f"Result: {result}")

# Run the event loop
asyncio.run(main())

Output:

Hello, Alice! (starting)
Hello again, Alice! (after 1s)
Result: Done greeting Alice

async def vs def — Key Differences

import asyncio
import time

# Regular function: blocks while sleeping
def sync_task(name, duration):
    print(f"[sync] {name}: started")
    time.sleep(duration)           # BLOCKS — no other code runs
    print(f"[sync] {name}: done ({duration}s)")

# Coroutine: yields control during await
async def async_task(name, duration):
    print(f"[async] {name}: started")
    await asyncio.sleep(duration)  # YIELDS — event loop runs other tasks
    print(f"[async] {name}: done ({duration}s)")

# Sync: sequential, total = 3s
print("=== Synchronous ===")
start = time.time()
sync_task("A", 1)
sync_task("B", 1)
sync_task("C", 1)
print(f"Sync total: {time.time()-start:.2f}s\n")

# Async: concurrent, total = ~1s
async def async_main():
    print("=== Asynchronous ===")
    start = time.time()
    await asyncio.gather(
        async_task("A", 1),
        async_task("B", 1),
        async_task("C", 1),
    )
    print(f"Async total: {time.time()-start:.2f}s")

asyncio.run(async_main())

Output:

=== Synchronous ===
[sync] A: started
[sync] A: done (1s)
[sync] B: started
[sync] B: done (1s)
[sync] C: started
[sync] C: done (1s)
Sync total: 3.00s

=== Asynchronous ===
[async] A: started
[async] B: started
[async] C: started
[async] A: done (1s)
[async] B: done (1s)
[async] C: done (1s)
Async total: 1.00s

asyncio.gather — Run Tasks Concurrently

gather() is the workhorse of async Python — runs multiple coroutines concurrently and collects all results.

import asyncio
import time

async def fetch_data(source, delay):
    """Simulate fetching data from a source."""
    print(f"  Fetching from {source}...")
    await asyncio.sleep(delay)
    print(f"  Got data from {source}")
    return {"source": source, "data": f"result_{source}"}

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

    # All three run concurrently — total time ~ max(0.5, 1.0, 0.8) = 1s
    results = await asyncio.gather(
        fetch_data("database", 0.5),
        fetch_data("api", 1.0),
        fetch_data("cache", 0.8),
    )

    print(f"\nAll done in {time.time()-start:.2f}s")
    for r in results:
        print(f"  {r}")

    # return_exceptions=True prevents one failure from cancelling all others
    results_safe = await asyncio.gather(
        fetch_data("db", 0.3),
        fetch_data("api", 0.5),
        return_exceptions=True
    )
    print("\nSafe results:", results_safe)

asyncio.run(main())

Output:

  Fetching from database...
  Fetching from api...
  Fetching from cache...
  Got data from database
  Got data from cache
  Got data from api

All done in 1.01s
  {'source': 'database', 'data': 'result_database'}
  {'source': 'api', 'data': 'result_api'}
  {'source': 'cache', 'data': 'result_cache'}

asyncio.create_task — Fire and Forget

import asyncio
import time

async def background_job(job_id, duration):
    """A task that runs independently in the background."""
    print(f"Job {job_id}: starting")
    await asyncio.sleep(duration)
    print(f"Job {job_id}: completed after {duration}s")
    return f"result_{job_id}"

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

    # Create tasks — they start immediately in the background
    task1 = asyncio.create_task(background_job(1, 2), name="job-1")
    task2 = asyncio.create_task(background_job(2, 1), name="job-2")
    task3 = asyncio.create_task(background_job(3, 3), name="job-3")

    print("Tasks created, doing other work now...")
    await asyncio.sleep(0.5)  # Do other work while tasks run
    print("Still doing other things...")

    # Await tasks when you need their results
    result1 = await task1
    result2 = await task2
    result3 = await task3

    print(f"\nResults: {result1}, {result2}, {result3}")
    print(f"Total time: {time.time()-start:.2f}s")

asyncio.run(main())

Output:

Job 1: starting
Job 2: starting
Job 3: starting
Tasks created, doing other work now...
Still doing other things...
Job 2: completed after 1s
Job 1: completed after 2s
Job 3: completed after 3s

Results: result_1, result_2, result_3
Total time: 3.01s

Async Context Managers & Iterators

import asyncio

# Async Context Manager
class AsyncDatabase:
    """Simulates an async database connection."""

    async def __aenter__(self):
        print("Opening DB connection...")
        await asyncio.sleep(0.1)  # Simulate connection time
        print("DB connected!")
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing DB connection...")
        await asyncio.sleep(0.05)
        print("DB disconnected.")

    async def query(self, sql):
        await asyncio.sleep(0.2)  # Simulate query time
        return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

# Async Generator
async def async_range(start, stop, delay=0.1):
    """An async generator that yields values with delays."""
    for i in range(start, stop):
        await asyncio.sleep(delay)  # Non-blocking delay
        yield i

async def main():
    # Using async context manager
    async with AsyncDatabase() as db:
        rows = await db.query("SELECT * FROM users")
        print(f"Query returned: {rows}")

    print()

    # Using async for with async generator
    print("Async iteration:")
    async for value in async_range(0, 5, 0.1):
        print(f"  Got: {value}")

asyncio.run(main())

Output:

Opening DB connection...
DB connected!
Query returned: [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]
Closing DB connection...
DB disconnected.

Async iteration:
  Got: 0
  Got: 1
  Got: 2
  Got: 3
  Got: 4

asyncio.Semaphore — Throttle Concurrency

import asyncio
import time

async def fetch_with_limit(semaphore, url_id):
    """Only N requests can run at the same time."""
    async with semaphore:  # Blocks if semaphore is at capacity
        print(f"  Fetching URL {url_id}...")
        await asyncio.sleep(0.5)  # Simulate API call
        print(f"  Done URL {url_id}")
        return f"response_{url_id}"

async def main():
    # Limit to 3 concurrent requests (e.g., rate-limit an API)
    semaphore = asyncio.Semaphore(3)

    urls = list(range(10))  # 10 URLs to fetch

    start = time.time()
    results = await asyncio.gather(
        *[fetch_with_limit(semaphore, url) for url in urls]
    )

    print(f"\nFetched {len(results)} URLs in {time.time()-start:.2f}s")
    print("(With limit=3: ~4 batches of 3 = ~2s instead of ~5s)")

asyncio.run(main())

Output:

  Fetching URL 0...
  Fetching URL 1...
  Fetching URL 2...
  Done URL 0
  Fetching URL 3...
  Done URL 1
  ...
Fetched 10 URLs in 2.01s
(With limit=3: ~4 batches of 3 = ~2s instead of ~5s)

Exception Handling in Async Code

import asyncio

async def risky_operation(n):
    """Sometimes fails."""
    await asyncio.sleep(0.1)
    if n % 3 == 0:
        raise ValueError(f"Failed for n={n}")
    return f"success_{n}"

async def safe_wrapper(n):
    """Wraps an operation to handle its exceptions."""
    try:
        result = await risky_operation(n)
        return result
    except ValueError as e:
        print(f"  Caught error: {e}")
        return None

async def main():
    print("--- Method 1: Handle in wrapper ---")
    tasks = [safe_wrapper(i) for i in range(9)]
    results = await asyncio.gather(*tasks)
    print("Results:", results)

    print("\n--- Method 2: return_exceptions=True ---")
    raw_results = await asyncio.gather(
        *[risky_operation(i) for i in range(6)],
        return_exceptions=True  # Errors become values, not crashes
    )
    for i, r in enumerate(raw_results):
        if isinstance(r, Exception):
            print(f"  Task {i} FAILED: {r}")
        else:
            print(f"  Task {i} OK: {r}")

asyncio.run(main())

Output:

--- Method 1: Handle in wrapper ---
  Caught error: Failed for n=0
  Caught error: Failed for n=3
  Caught error: Failed for n=6
Results: [None, 'success_1', 'success_2', None, 'success_4', 'success_5', None, 'success_7', 'success_8']

--- Method 2: return_exceptions=True ---
  Task 0 FAILED: Failed for n=0
  Task 1 OK: success_1
  Task 2 OK: success_2
  Task 3 FAILED: Failed for n=3
  Task 4 OK: success_4
  Task 5 OK: success_5

Real-World Pattern: Async HTTP with aiohttp

# NOTE: Install aiohttp in a real project: pip install aiohttp
# This example simulates the aiohttp pattern

import asyncio
import time

# --- Simulated async HTTP client (mirrors real aiohttp API) ---
class MockSession:
    async def __aenter__(self): return self
    async def __aexit__(self, *_): pass

    async def get(self, url):
        return MockResponse(url)

class MockResponse:
    def __init__(self, url):
        self.url = url
        self.status = 200

    async def __aenter__(self): return self
    async def __aexit__(self, *_): pass

    async def json(self):
        await asyncio.sleep(0.3)  # Simulate network latency
        return {"url": self.url, "data": "response_data"}

# --- Real aiohttp usage looks exactly like this ---
async def fetch_url(session, url):
    """Fetch a single URL asynchronously."""
    async with session.get(url) as response:
        if response.status == 200:
            data = await response.json()
            return data
        else:
            raise Exception(f"HTTP {response.status} for {url}")

async def fetch_all_urls(urls):
    """Fetch multiple URLs concurrently with rate limiting."""
    semaphore = asyncio.Semaphore(5)  # Max 5 concurrent requests

    async def bounded_fetch(session, url):
        async with semaphore:
            return await fetch_url(session, url)

    async with MockSession() as session:  # Real: aiohttp.ClientSession()
        tasks = [bounded_fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

async def main():
    urls = [f"https://api.example.com/items/{i}" for i in range(10)]

    start = time.time()
    results = await fetch_all_urls(urls)
    elapsed = time.time() - start

    successful = [r for r in results if not isinstance(r, Exception)]
    print(f"Fetched {len(successful)}/{len(urls)} URLs in {elapsed:.2f}s")
    print(f"Sequential would take ~{len(urls) * 0.3:.1f}s")
    print("First result:", results[0])

asyncio.run(main())

Output:

Fetched 10/10 URLs in 0.62s
Sequential would take ~3.0s
First result: {'url': 'https://api.example.com/items/0', 'data': 'response_data'}

Async Queue — Producer-Consumer Pattern

import asyncio
import time

async def async_producer(queue, n_items):
    """Produces items asynchronously."""
    for i in range(n_items):
        item = f"item-{i}"
        await queue.put(item)
        print(f"Produced: {item}")
        await asyncio.sleep(0.2)

    # Send sentinels to stop each consumer
    for _ in range(3):
        await queue.put(None)

async def async_consumer(queue, consumer_id):
    """Consumes items asynchronously."""
    while True:
        item = await queue.get()
        if item is None:
            print(f"Consumer {consumer_id}: stopping")
            queue.task_done()
            break
        await asyncio.sleep(0.1)  # Simulate async processing
        print(f"Consumer {consumer_id} processed: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)

    start = time.time()

    # Run producer and consumers concurrently
    await asyncio.gather(
        async_producer(queue, 9),
        async_consumer(queue, 1),
        async_consumer(queue, 2),
        async_consumer(queue, 3),
    )

    print(f"\nAll done in {time.time()-start:.2f}s")

asyncio.run(main())

Output:

Produced: item-0
Consumer 1 processed: item-0
Produced: item-1
Consumer 2 processed: item-1
...
Consumer 1: stopping
Consumer 2: stopping
Consumer 3: stopping

All done in 1.81s

asyncio.wait_for — Timeouts

import asyncio

async def slow_operation(name, duration):
    """An operation that may be too slow."""
    print(f"{name}: starting (will take {duration}s)")
    await asyncio.sleep(duration)
    return f"{name}: completed"

async def main():
    # Operation within timeout — succeeds
    try:
        result = await asyncio.wait_for(
            slow_operation("fast-op", 0.5),
            timeout=2.0  # Allow up to 2 seconds
        )
        print(f"Success: {result}")
    except asyncio.TimeoutError:
        print("fast-op timed out!")

    print()

    # Operation exceeds timeout — cancelled
    try:
        result = await asyncio.wait_for(
            slow_operation("slow-op", 3.0),
            timeout=1.0  # Only allow 1 second
        )
        print(f"Success: {result}")
    except asyncio.TimeoutError:
        print("slow-op cancelled — took too long!")

    print()

    # asyncio.wait: more granular control than gather
    tasks = [
        asyncio.create_task(slow_operation("task-A", 0.5)),
        asyncio.create_task(slow_operation("task-B", 1.5)),
        asyncio.create_task(slow_operation("task-C", 0.8)),
    ]
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    print(f"First done: {len(done)} task(s), still pending: {len(pending)}")
    for task in pending:
        task.cancel()  # Cancel remaining tasks

asyncio.run(main())

Output:

fast-op: starting (will take 0.5s)
Success: fast-op: completed

slow-op: starting (will take 3.0s)
slow-op cancelled — took too long!

task-A: starting (will take 0.5s)
task-B: starting (will take 1.5s)
task-C: starting (will take 0.8s)
First done: 1 task(s), still pending: 2

---

Sync vs Async: Side-by-Side Comparison

import asyncio
import time

# ===== SYNCHRONOUS VERSION =====
def sync_pipeline():
    def fetch(item_id): time.sleep(0.3); return f"raw_{item_id}"
    def process(data): time.sleep(0.2); return f"processed_{data}"
    def save(data):    time.sleep(0.1); return f"saved_{data}"

    start = time.time()
    results = []
    for item_id in range(5):
        raw = fetch(item_id)
        processed = process(raw)
        saved = save(processed)
        results.append(saved)
    print(f"Sync pipeline: {len(results)} items in {time.time()-start:.2f}s")

# ===== ASYNCHRONOUS VERSION =====
async def async_pipeline():
    async def fetch(item_id):
        await asyncio.sleep(0.3)
        return f"raw_{item_id}"

    async def process(data):
        await asyncio.sleep(0.2)
        return f"processed_{data}"

    async def save(data):
        await asyncio.sleep(0.1)
        return f"saved_{data}"

    async def pipeline_one(item_id):
        raw = await fetch(item_id)
        processed = await process(raw)
        return await save(processed)

    start = time.time()
    results = await asyncio.gather(*[pipeline_one(i) for i in range(5)])
    print(f"Async pipeline: {len(results)} items in {time.time()-start:.2f}s")

# Run both and compare
sync_pipeline()
asyncio.run(async_pipeline())

Output:

Sync pipeline: 5 items in 3.00s   <- 5 × (0.3+0.2+0.1) = 3s
Async pipeline: 5 items in 0.60s  <- all 5 run concurrently

---

When to Use Async?

Coding Challenge: Async Web Scraper Simulator

import asyncio
import time
import random

async def fetch_page(url, semaphore):
    """Fetch one page with concurrency limit."""
    async with semaphore:
        delay = random.uniform(0.2, 1.0)
        await asyncio.sleep(delay)
        # Simulate occasional failure
        if random.random() < 0.2:
            raise ConnectionError(f"Failed to fetch {url}")
        word_count = random.randint(100, 5000)
        return {"url": url, "words": word_count, "time": round(delay, 2)}

async def scrape_site(urls, max_concurrent=5):
    """Scrape multiple pages concurrently, respecting rate limits."""
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [fetch_page(url, semaphore) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    success = [r for r in results if isinstance(r, dict)]
    failed  = [r for r in results if isinstance(r, Exception)]
    return success, failed

async def main():
    urls = [f"https://example.com/page/{i}" for i in range(20)]

    print(f"Scraping {len(urls)} pages with max 5 concurrent...")
    start = time.time()

    success, failed = await scrape_site(urls, max_concurrent=5)
    elapsed = time.time() - start

    total_words = sum(r["words"] for r in success)
    avg_words   = total_words // len(success) if success else 0

    print(f"\nCompleted in {elapsed:.2f}s")
    print(f"Successful: {len(success)}, Failed: {len(failed)}")
    print(f"Total words scraped: {total_words:,}")
    print(f"Average words per page: {avg_words:,}")
    if failed:
        print(f"Errors: {[str(e) for e in failed[:3]]}")

asyncio.run(main())

Output:

Scraping 20 pages with max 5 concurrent...

Completed in 2.43s
Successful: 16, Failed: 4
Total words scraped: 38,241
Average words per page: 2,390
Errors: ['Failed to fetch https://example.com/page/2', ...]

Quiz

Key Takeaways

✅ async def defines a coroutine — a function that can be paused.
✅ await yields control back to the event loop during I/O waits.
✅ asyncio.gather() runs many coroutines concurrently and collects results.
✅ asyncio.create_task() schedules a coroutine as a background task.
✅ asyncio.Semaphore(n) throttles concurrent access to n at a time.
✅ Use async with and async for for async context managers and iterators.
✅ asyncio is ideal for I/O-bound work — for CPU-bound work use multiprocessing.

What's Next?

You've now mastered Python's three concurrency models:

• threading — shared-memory I/O concurrency
• multiprocessing — true CPU parallelism
• asyncio — single-threaded, high-concurrency I/O

Combine them wisely: use asyncio for your main I/O loop, ProcessPoolExecutor for CPU tasks inside it, and you'll build blazing-fast Python applications.

Keep coding! 🚀