Asyncio foundations

asyncio is Python's standard framework for concurrent I/O using async/await. An event loop runs tasks on one thread. A task continues until it finishes, raises, or reaches an operation that suspends it.

Coroutines are not results

import asyncio

async def answer() -> int:
    await asyncio.sleep(0.1)
    return 42

async def main() -> None:
    value = await answer()
    print(value)

if __name__ == "__main__":
    asyncio.run(main())

Calling answer() creates a coroutine object; it does not execute the body. Await it or schedule it as a task. asyncio.run() owns the event loop and should normally appear once at the program boundary. It cannot be called while another loop is running, as in many notebook environments.

Sequential versus concurrent awaits

async def sequential() -> tuple[int, int]:
    first = await answer()
    second = await answer()
    return first, second

async def concurrent() -> tuple[int, int]:
    first_task = asyncio.create_task(answer())
    second_task = asyncio.create_task(answer())
    return await first_task, await second_task

Creating a task schedules independent progress. Keep a strong reference and eventually await it. For related tasks, prefer TaskGroup, covered next, because it owns their lifetime and failure behavior.

Cooperative scheduling

Async code is concurrent only at suspension points. A CPU-heavy loop or blocking call freezes the event loop:

import asyncio
from pathlib import Path

def blocking_read(path: Path) -> str:
    return path.read_text(encoding="utf-8")

async def read_without_blocking_loop(path: Path) -> str:
    return await asyncio.to_thread(blocking_read, path)

to_thread() is intended mainly for blocking I/O. It does not make CPU-bound Python execute efficiently in parallel. Use a process pool for substantial pure-Python CPU work.

Async context managers and iterators

Resources can require asynchronous setup and cleanup:

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

@asynccontextmanager
async def connection() -> AsyncIterator[str]:
    resource = "connected"
    try:
        yield resource
    finally:
        await asyncio.sleep(0)

Use async with for async context managers and async for for async iterables. Cleanup may itself await, which is why a synchronous with cannot replace it.

Streams and backpressure

asyncio.open_connection() returns a StreamReader and StreamWriter:

import asyncio

async def fetch_status(host: str, port: int) -> bytes:
    reader, writer = await asyncio.open_connection(host, port)
    try:
        writer.write(b"STATUS\n")
        await writer.drain()
        return await reader.readline()
    finally:
        writer.close()
        await writer.wait_closed()

drain() applies flow control when the transport buffer grows. Set application-level timeouts and message-size limits; a peer may connect but never finish a message.

Debugging tasks

Name tasks for diagnostics:

task = asyncio.create_task(answer(), name="compute-answer")
print(task.get_name())

Enable asyncio debug mode during development with asyncio.run(main(), debug=True) or PYTHONASYNCIODEBUG=1. Debug mode can reveal slow callbacks and forgotten awaits, but tests must still exercise cancellation and failure paths.

Common mistakes

• Calling a coroutine without awaiting it.
• Using time.sleep() inside async code instead of await asyncio.sleep().
• Performing synchronous DNS, HTTP, file, or database work on the loop thread.
• Creating unowned background tasks whose exceptions are never observed.
• Assuming async makes CPU work faster.
• Forgetting that cancellation can occur at almost any await.

Practice

Create an async TCP client that sends three requests concurrently, names each task, limits every response to one line, and closes every writer. First implement it sequentially and compare elapsed time with a server that delays each response.