Python Async/Await

Mentor's Note: Async programming is like cooking multiple dishes at once — while the pasta water is boiling, you chop vegetables instead of staring at the pot. 🍳

---

The Scenario: The Restaurant Kitchen

Imagine running a kitchen with one chef.

• Synchronous Chef: Starts boiling water. Stands and watches until it boils. Then starts chopping. Then watches that finish. Everything is done one at a time.
• Async Chef: Starts boiling water. While waiting, chops vegetables. While the sauce simmers, plates the previous order. Everything is done concurrently — waiting time is productive time.
• The Result: More orders completed per hour with the same chef. ✅

---

Concept Explanation

1. What Is Async Programming?

Async programming lets a program wait for slow operations (network requests, file I/O, database queries) without blocking the entire program. Instead of sitting idle, Python can work on other tasks.

Approach	Behavior	Analogy
Synchronous	One task at a time, blocking	Single checkout lane
Multithreading	Multiple tasks, shared memory	Multiple checkout lanes
Async/Await	Single-threaded concurrency	Self-checkout — scan next item while bagging the last

2. The Core Syntax

import asyncio

# Define an async function (coroutine)
async def say_hello():
    print("Hello")
    await asyncio.sleep(1)  # Pause here — other tasks can run
    print("World")

# Run the coroutine
asyncio.run(say_hello())

---

async def and await

async def

Declares a function as a coroutine — it returns a coroutine object, not the result directly.

async def fetch_data():
    return {"id": 1, "name": "Alice"}

# This does NOT run the function:
result = fetch_data()       # Returns a coroutine object
print(result)               # <coroutine object fetch_data at ...>

# You must await it:
result = await fetch_data() # Runs and returns the dict

await

Pauses the current coroutine until the awaited operation completes. During that pause, Python runs other coroutines.

async def step1():
    print("Step 1: Starting")
    await asyncio.sleep(2)   # Pause 2 seconds
    print("Step 1: Done")
    return "Result A"

async def step2():
    print("Step 2: Starting")
    await asyncio.sleep(1)   # Pause 1 second
    print("Step 2: Done")
    return "Result B"

---

Running the Event Loop

import asyncio

async def main():
    print("Main started")
    await asyncio.sleep(1)
    print("Main finished")

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

asyncio.run() creates a new event loop, runs the coroutine, and closes the loop — it handles all the boilerplate for you.

---

Creating Tasks

Tasks let you run multiple coroutines concurrently:

import asyncio

async def fetch_data(delay: int, name: str) -> str:
    print(f"Fetching {name}...")
    await asyncio.sleep(delay)
    print(f"{name} done!")
    return f"Data from {name}"

async def main():
    # Create tasks — they start running immediately
    task1 = asyncio.create_task(fetch_data(3, "API A"))
    task2 = asyncio.create_task(fetch_data(1, "API B"))
    task3 = asyncio.create_task(fetch_data(2, "API C"))

    # Wait for all tasks to complete
    result1 = await task1
    result2 = await task2
    result3 = await task3

    print(f"Results: {result1}, {result2}, {result3}")

asyncio.run(main())

Output:

Fetching API A...
Fetching API B...
Fetching API C...
API B done!
API C done!
API A done!
Results: Data from API A, Data from API B, Data from API C

Notice that all three fetches started immediately, and the shorter ones finished first — total time is ~3 seconds, not 6.

---

asyncio.gather()

A cleaner way to run multiple tasks and collect their results:

import asyncio

async def fetch_data(delay: int, name: str) -> str:
    await asyncio.sleep(delay)
    return f"Data from {name}"

async def main():
    results = await asyncio.gather(
        fetch_data(3, "API A"),
        fetch_data(1, "API B"),
        fetch_data(2, "API C"),
    )
    print(results)  # ['Data from API A', 'Data from API B', 'Data from API C']

asyncio.run(main())

---

Real-World Example: Fetching Multiple URLs

import asyncio
import time

# Use httpx or aiohttp for real async HTTP
import httpx

async def fetch_url(url: str, client: httpx.AsyncClient) -> dict:
    print(f"Fetching: {url}")
    response = await client.get(url)
    return {"url": url, "status": response.status_code, "length": len(response.text)}

async def main():
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2",
        "https://jsonplaceholder.typicode.com/posts/3",
        "https://jsonplaceholder.typicode.com/posts/4",
        "https://jsonplaceholder.typicode.com/posts/5",
    ]

    async with httpx.AsyncClient() as client:
        tasks = [fetch_url(url, client) for url in urls]
        results = await asyncio.gather(*tasks)

    for r in results:
        print(f"{r['url']}: {r['status']} ({r['length']} bytes)")

start = time.time()
asyncio.run(main())
print(f"Total time: {time.time() - start:.2f}s")

---

Sync vs Async Comparison

Synchronous Version (Sequential)

import time
import requests

def fetch_url(url: str) -> dict:
    response = requests.get(url)
    return {"url": url, "length": len(response.text)}

def main():
    urls = ["https://jsonplaceholder.typicode.com/posts/1",
            "https://jsonplaceholder.typicode.com/posts/2",
            "https://jsonplaceholder.typicode.com/posts/3"]

    for url in urls:
        result = fetch_url(url)
        print(f"{result['url']}: {result['length']} bytes")

start = time.time()
main()
print(f"Sync time: {time.time() - start:.2f}s")  # ~2-3 seconds

Async Version (Concurrent)

import asyncio
import time
import httpx

async def fetch_url(url: str, client: httpx.AsyncClient) -> dict:
    response = await client.get(url)
    return {"url": url, "length": len(response.text)}

async def main():
    urls = ["https://jsonplaceholder.typicode.com/posts/1",
            "https://jsonplaceholder.typicode.com/posts/2",
            "https://jsonplaceholder.typicode.com/posts/3"]

    async with httpx.AsyncClient() as client:
        tasks = [fetch_url(url, client) for url in urls]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(f"{r['url']}: {r['length']} bytes")

start = time.time()
asyncio.run(main())
print(f"Async time: {time.time() - start:.2f}s")  # ~1 second

Result: Async completes all 3 requests in roughly the time of the slowest single request.

---

Visual Logic: Execution Flow

---

Common Pitfalls

Pitfall 1: Forgetting await

async def fetch():
    return "data"

async def main():
    result = fetch()           # ❌ Returns coroutine, not data!
    result = await fetch()     # ✅ Correct

asyncio.run(main())

Pitfall 2: Blocking Calls in Async Code

import asyncio
import time

async def bad():
    time.sleep(5)          # ❌ Blocks the entire event loop!
    # Use await asyncio.sleep(5) instead

async def good():
    await asyncio.sleep(5) # ✅ Yields control, others can run

Pitfall 3: Running Without asyncio.run()

async def hello():
    print("Hello")

# hello()                 # ❌ Warning: coroutine was never awaited
# asyncio.run(hello())    # ✅ Correct

---

Sample Dry Run

Scenario: Fetching 3 API endpoints with delays 3s, 1s, 2s

Time	Task 1 (3s)	Task 2 (1s)	Task 3 (2s)
0s	Started	Started	Started
1s	Waiting	Done ✅	Waiting
2s	Waiting	Done	Done ✅
3s	Done ✅	Done	Done

Total time: 3 seconds (not 6) 🏆

---

Pro Tips

• Use async for I/O-bound tasks (network, files, databases), not CPU-bound work.
• Pair asyncio.gather() with return_exceptions=True to prevent one failure from cancelling all tasks.
• Install aiohttp or httpx for async HTTP; aiosqlite for async databases.
• Python 3.11+ has significantly faster asyncio performance.

---

Interview Tip

"Interviewers love: 'What's the difference between concurrency and parallelism?' Concurrency is about dealing with many things at once (structure); parallelism is about doing many things at once (execution). Async provides concurrency on a single thread — context switching at await points."

---

← Back: Type Hints | Next: HTTP Requests →