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.

---

2. Concurrency vs Parallelism vs Blocking#

Python
CopyAsk AI
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#

Python
CopyAsk AI
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

Python
CopyAsk AI
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#

Python
CopyAsk AI
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.

Python
CopyAsk AI
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#

Python
CopyAsk AI
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.

Python
CopyAsk AI
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.

Python
CopyAsk AI
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.

Python
CopyAsk AI
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#

Python
CopyAsk AI
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#

Python
CopyAsk AI
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#

Python
CopyAsk AI
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#

Python
CopyAsk AI
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#

Python
CopyAsk AI
async def main():
    result = await fetch()
    print(result)  # data

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

Python
CopyAsk AI
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#

Python
CopyAsk AI
async def main():
    task = asyncio.create_task(fetch_data("A", 2))
    await task

---

8. A Realistic Example: Concurrent API Calls#

Python
CopyAsk AI
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.