Why Async?

Async I/O lets a single thread handle many concurrent operations — ideal for web servers, API clients, and database connections where tasks spend time waiting.

Basic asyncio 

  import asyncio

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

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

asyncio.run(main())

Running Multiple Tasks 

  async def main():
    results = await asyncio.gather(
        fetch_data("users", 2),
        fetch_data("orders", 1),
        fetch_data("products", 3),
    )
    print(results)  # all three complete in ~3s, not 6s

asyncio.run(main())

Async Context Managers 

  import aiohttp

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

Async Generators 

  async def stream_items():
    for i in range(5):
        await asyncio.sleep(0.5)
        yield i

async def main():
    async for item in stream_items():
        print(item)

When to Use Async

Good for Not good for
HTTP/API calls CPU-bound computation
Database queries Image/video processing
WebSocket servers Heavy numerical work

For CPU-bound tasks, use multiprocessing or concurrent.futures.ProcessPoolExecutor instead.

FastAPI Example 

  from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await db.fetch_user(user_id)  # non-blocking DB call
    return user

Async Python is essential for modern web backends and high-concurrency services.

asyncio.create_task

Fire-and-forget or concurrent tasks without blocking:

  async def main():
    task1 = asyncio.create_task(fetch_data("users", 2))
    task2 = asyncio.create_task(fetch_data("orders", 1))
    await task1
    await task2

asyncio.wait_for Timeout 

  try:
    result = await asyncio.wait_for(fetch_data("slow-api", 30), timeout=5.0)
except asyncio.TimeoutError:
    print("Request timed out")

Semaphore for Rate Limiting 

  sem = asyncio.Semaphore(10)  # max 10 concurrent requests

async def fetch_limited(url):
    async with sem:
        return await fetch_url(url)

Running Sync Code in Async Context 

  import asyncio

def cpu_heavy():
    return sum(i * i for i in range(10_000_000))

async def main():
    result = await asyncio.to_thread(cpu_heavy)
    print(result)

Use asyncio.to_thread() (Python 3.9+) to avoid blocking the event loop.

Debugging Async Code 

  import asyncio
asyncio.run(main(), debug=True)  # warns about slow callbacks

Enable debug mode during development to catch unawaited coroutines.