Async Python

1. Overview#

Async Python lets a single thread juggle many tasks that spend time waiting — for network responses, file I/O, or database queries — without blocking the whole program. It does not give you true parallelism (that's multiprocessing); it gives you highly efficient concurrency for I/O-bound work.

Rule of thumb: use asyncio for I/O-bound work (network calls, APIs, DB queries). Use multiprocessing for CPU-bound work (heavy computation). Async code doesn't make math run faster — it makes waiting cheaper.

---

2. Concurrency vs Parallelism vs Blocking#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import time

def blocking_fetch(name, delay):
 print(f"Fetching {name}...")
 time.sleep(delay) # blocks the entire program
 print(f"Done: {name}")

blocking_fetch("A", 2)
blocking_fetch("B", 2)
# Total time: ~4 seconds — one after another

---

3. Coroutines: async def and await#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio

async def fetch_data(name, delay):
 print(f"Fetching {name}...")
 await asyncio.sleep(delay) # non-blocking "wait" — hands control back to the event loop
 print(f"Done: {name}")
 return f"{name} result"

async def main():
 result = await fetch_data("A", 2)
 print(result)

asyncio.run(main()) # entry point that starts the event loop

Calling fetch_data("A", 2) on its own does not run the function — it creates a coroutine object. The function body only executes once it's await-ed or scheduled on the event loop.

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
async def greet():
 return "hello"

coro = greet() # nothing has run yet
print(coro) # <coroutine object greet at 0x...>
result = asyncio.run(coro) # now it actually runs

---

4. Running Things Concurrently#

4.1 asyncio.gather — Run Many Coroutines at Once#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio

async def fetch_data(name, delay):
 print(f"Fetching {name}...")
 await asyncio.sleep(delay)
 print(f"Done: {name}")
 return f"{name} result"

async def main():
 results = await asyncio.gather(
 fetch_data("A", 2),
 fetch_data("B", 2),
 fetch_data("C", 2),
 )
 print(results)

asyncio.run(main())
# Total time: ~2 seconds, not 6 — all three run concurrently

4.2 asyncio.create_task — Schedule Work in the Background#

create_task starts a coroutine running immediately (scheduled on the event loop) without waiting for it — useful when you want to kick off work and do something else before collecting the result.

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio

async def fetch_data(name, delay):
 await asyncio.sleep(delay)
 return f"{name} result"

async def main():
 task_a = asyncio.create_task(fetch_data("A", 2))
 task_b = asyncio.create_task(fetch_data("B", 2))

 print("Tasks started, doing other work...")
 result_a = await task_a
 result_b = await task_b
 print(result_a, result_b)

asyncio.run(main())

4.3 asyncio.wait_for — Timeouts#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio

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

async def main():
 try:
 result = await asyncio.wait_for(slow_operation(), timeout=2)
 except asyncio.TimeoutError:
 print("Operation timed out")

asyncio.run(main())

---

5. Async Context Managers & Iterators#

5.1 async with#

Used for resources that need asynchronous setup/teardown — like a network connection pool.

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
class AsyncConnection:
 async def __aenter__(self):
 print("Opening connection...")
 await asyncio.sleep(1) # simulate async connect
 return self

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

async def main():
 async with AsyncConnection() as conn:
 print("Using connection")

asyncio.run(main())

5.2 async for#

Iterates over an async generator — one item produced at a time, each possibly involving an await.

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio

async def fetch_pages():
 for page in range(1, 4):
 await asyncio.sleep(1) # simulate an API call per page
 yield f"Page {page} data"

async def main():
 async for page_data in fetch_pages():
 print(page_data)

asyncio.run(main())

---

6. Concurrency Control: Semaphores and Locks#

6.1 asyncio.Semaphore — Limit Concurrent Operations#

Useful when calling an API that rate-limits how many requests can run at once.

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio

async def fetch_with_limit(semaphore, name):
 async with semaphore:
 print(f"Fetching {name}...")
 await asyncio.sleep(2)
 print(f"Done: {name}")

async def main():
 semaphore = asyncio.Semaphore(2) # max 2 concurrent fetches
 await asyncio.gather(*(fetch_with_limit(semaphore, f"item-{i}") for i in range(5)))

asyncio.run(main())

6.2 asyncio.Lock — Protect Shared State#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio

counter = 0
lock = asyncio.Lock()

async def increment():
 global counter
 async with lock:
 current = counter
 await asyncio.sleep(0.01) # simulate work between read and write
 counter = current + 1

async def main():
 await asyncio.gather(*(increment() for _ in range(10)))
 print(counter) # 10 — safe, thanks to the lock

asyncio.run(main())

---

7. Common Pitfalls#

INCORRECT: Mixing Blocking Calls Into Async Code#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio
import time

async def bad_fetch():
 time.sleep(3) # BLOCKS the entire event loop — defeats the purpose of async
 return "data"

CORRECT: Fix — Use the Async Equivalent, or Offload to a Thread#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio

async def good_fetch():
 await asyncio.sleep(3) # yields control back to the event loop
 return "data"

# For unavoidable blocking/CPU-bound calls, offload to a thread pool:
async def wraps_blocking_call():
 result = await asyncio.to_thread(time.sleep, 3)
 return result

INCORRECT: Forgetting to await a Coroutine#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
async def fetch():
 return "data"

async def main():
 result = fetch() # BUG: this is a coroutine object, not the result!
 print(result) # <coroutine object fetch at 0x...>

CORRECT: Fix#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
async def main():
 result = await fetch()
 print(result) # data

INCORRECT: Creating a Task and Never Awaiting or Storing It#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
async def main():
 asyncio.create_task(fetch_data("A", 2)) # fire-and-forget — may be garbage collected mid-run

CORRECT: Fix — Keep a Reference and Await It#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
async def main():
 task = asyncio.create_task(fetch_data("A", 2))
 await task

---

8. A Realistic Example: Concurrent API Calls#

🐍 PythonInteractive WebAssembly
EditCopyAsk AIRun Code
import asyncio
import random

async def call_api(endpoint):
 print(f"Calling {endpoint}...")
 await asyncio.sleep(random.uniform(1, 3)) # simulate variable network latency
 return {"endpoint": endpoint, "status": 200}

async def fetch_all(endpoints):
 semaphore = asyncio.Semaphore(3) # limit to 3 concurrent calls

 async def bounded_call(endpoint):
 async with semaphore:
 return await call_api(endpoint)

 tasks = [bounded_call(ep) for ep in endpoints]
 return await asyncio.gather(*tasks)

async def main():
 endpoints = [f"/api/resource/{i}" for i in range(8)]
 results = await fetch_all(endpoints)
 for r in results:
 print(r)

asyncio.run(main())

---

9. Summary & Best Practices Checklist#

• Use asyncio for I/O-bound work; use multiprocessing for CPU-bound work.
• Always await a coroutine — calling it alone only creates the coroutine object.
• Use asyncio.gather to run independent coroutines concurrently.
• Use asyncio.create_task when you want work to start now but collect the result later.
• Never call blocking functions (time.sleep, blocking I/O) directly inside async def — use asyncio.sleep or asyncio.to_thread.
• Use a Semaphore to cap concurrency against rate-limited APIs.
• Use a Lock when multiple coroutines mutate shared state.
• Always keep a reference to tasks created with create_task until they're awaited.