German Yamil

Posted on May 28

Python asyncio: async/await Without the Confusion

#python #beginners #tutorial #codenewbie

Why async/await Exists

Regular Python code runs one line at a time. While one line waits for a network request, everything else pauses.

# Synchronous — each request waits for the previous
import requests

urls = ["https://api.example.com/1", "https://api.example.com/2", "https://api.example.com/3"]
for url in urls:
    r = requests.get(url)       # Waits here
    print(r.status_code)        # Then continues

Three requests, three full wait cycles. If each takes 1 second, total time: 3 seconds.

With async:

import asyncio, aiohttp

async def fetch(session, url):
    async with session.get(url) as r:
        return r.status

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

Same three requests. Total time: ~1 second (limited by the slowest request, not the sum).

The Core Concepts

async def — defining a coroutine

async def greet(name: str) -> str:
    return f"Hello, {name}"

# Must be awaited or run with asyncio.run()
result = asyncio.run(greet("world"))
print(result)  # Hello, world

async def creates a coroutine — a function that can pause and resume. Calling it doesn't run it. You need await or asyncio.run().

await — pause until done

import asyncio

async def slow_task():
    await asyncio.sleep(1)  # Pauses here, lets other tasks run
    return "done"

async def main():
    result = await slow_task()
    print(result)

asyncio.run(main())

await says: "pause here and let the event loop do other work while waiting."

asyncio.gather() — run multiple tasks concurrently

async def task(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"

async def main():
    # These run concurrently, not sequentially
    results = await asyncio.gather(
        task("A", 1.0),
        task("B", 0.5),
        task("C", 1.5),
    )
    print(results)
    # ['A done', 'B done', 'C done']
    # Total time: ~1.5s, not 3s

asyncio.run(main())

When to Use async

Use async when:

Making multiple HTTP requests
Reading/writing files concurrently
Handling WebSocket connections
Any I/O-bound work where waiting is the bottleneck

Don't use async when:

Doing CPU-intensive computation (use multiprocessing instead)
Only making one request (adds complexity for no gain)
Working with libraries that don't support async

Practical Example: Fetch Multiple APIs

import asyncio
import aiohttp
from typing import Any

async def fetch_json(session: aiohttp.ClientSession, url: str) -> dict[str, Any]:
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as r:
            r.raise_for_status()
            return await r.json()
    except Exception as e:
        return {"error": str(e), "url": url}

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_json(session, url) for url in urls]
        return await asyncio.gather(*tasks)

# Usage
urls = [
    "https://jsonplaceholder.typicode.com/posts/1",
    "https://jsonplaceholder.typicode.com/posts/2",
    "https://jsonplaceholder.typicode.com/posts/3",
]

results = asyncio.run(fetch_all(urls))
for r in results:
    print(r.get("title", r.get("error")))

Install aiohttp: pip install aiohttp

Common Mistakes

Mistake 1: Calling async functions without await

async def get_data():
    return "data"

# Wrong — returns a coroutine object, not the value
result = get_data()  # <coroutine object get_data at 0x...>

# Correct
result = asyncio.run(get_data())  # "data"

Mistake 2: Using blocking calls inside async functions

import time, requests

async def bad_example():
    time.sleep(1)           # Blocks the entire event loop!
    r = requests.get(url)   # Also blocking!
    return r.text

# Use asyncio.sleep() and aiohttp.ClientSession() instead

Mistake 3: Mixing sync and async without a bridge

# If you're in a sync context and need to call async:
import asyncio

def sync_function():
    result = asyncio.run(async_function())  # Correct bridge
    return result

async for and async with

Some objects support the async iteration and context manager protocols:

async def process_stream():
    # async with — for async context managers
    async with aiohttp.ClientSession() as session:

        # async for — for async iterables
        async for line in some_async_reader():
            process(line)

Quick Reference

Concept	Use	Notes
`async def`	Define coroutine	Must be awaited
`await`	Pause and yield	Only inside `async def`
`asyncio.run()`	Entry point	Top-level call
`asyncio.gather()`	Concurrent tasks	Returns list of results
`asyncio.sleep()`	Non-blocking wait	Use instead of `time.sleep()`
`asyncio.create_task()`	Background task	Doesn't wait by default

Get the Full Pipeline

This article is part of the Python AI Publishing Pipeline series — a complete system to write, validate, and publish technical ebooks with Python and Claude.

📋 Free checklist: 7 steps to ship a Python ebook — PDF, no email required.

🚀 Full pipeline + source code: germy5.gumroad.com/l/xhxkzz — $9.99, 30-day money-back guarantee.

If this was useful, the ❤️ button helps other developers find it.

Building a Python content pipeline? I sell the complete automation system as a one-time download — Dev.to API, Claude API, launchd, Gumroad. Check it out ($9.99)

DEV Community