Python Coroutines: async def and await Expressions

Key Insights

Coroutines enable concurrent I/O operations without threading overhead, making them ideal for network requests, file operations, and database queries where your code spends most of its time waiting.
The async def keyword creates a coroutine function that returns a coroutine object, not the actual result—you must use await or asyncio.run() to execute it and retrieve values.
Sequential await calls run one after another, defeating the purpose of async code—use asyncio.gather() or asyncio.create_task() to achieve true concurrency and performance gains.

Introduction to Coroutines and Asynchronous Programming

Python’s async/await syntax transforms how we handle I/O-bound operations. Traditional synchronous code blocks execution while waiting for external resources—network responses, file reads, database queries. Asynchronous programming lets your code start multiple operations and switch between them while waiting, maximizing efficiency.

Consider fetching data from three APIs. Synchronous code waits for each request to complete before starting the next one. If each takes 2 seconds, you’re looking at 6 seconds total. Asynchronous code fires off all three requests simultaneously and processes responses as they arrive, completing in roughly 2 seconds.

Here’s the syntax difference:

# Synchronous approach
import requests
import time

def fetch_data():
    start = time.time()
    response1 = requests.get('https://api.example.com/data1')
    response2 = requests.get('https://api.example.com/data2')
    response3 = requests.get('https://api.example.com/data3')
    print(f"Completed in {time.time() - start:.2f}s")

# Asynchronous approach
import asyncio
import aiohttp

async def fetch_data_async():
    start = time.time()
    async with aiohttp.ClientSession() as session:
        response1 = await session.get('https://api.example.com/data1')
        response2 = await session.get('https://api.example.com/data2')
        response3 = await session.get('https://api.example.com/data3')
    print(f"Completed in {time.time() - start:.2f}s")

asyncio.run(fetch_data_async())

The async version introduces two keywords: async def declares a coroutine function, and await suspends execution until an asynchronous operation completes.

Defining Coroutines with async def

When you prefix a function with async def, you’re not creating a regular function. You’re defining a coroutine function that returns a coroutine object when called. This object represents the execution of that async function but doesn’t run immediately.

import asyncio

async def greet(name):
    await asyncio.sleep(1)  # Simulate async operation
    return f"Hello, {name}!"

# Calling the function returns a coroutine object
coro = greet("Alice")
print(coro)  # <coroutine object greet at 0x...>
print(type(coro))  # <class 'coroutine'>

# To actually execute it and get the result:
result = asyncio.run(coro)
print(result)  # Hello, Alice!

This distinction is critical. Calling greet("Alice") doesn’t execute the function body or pause for that asyncio.sleep(1). It creates a coroutine object that you must either await inside another async function or run with asyncio.run() at the top level.

Inside async functions, you can use await with any “awaitable” object—coroutines, Tasks, or Futures. Regular function calls work normally, but you can’t call other coroutines without await.

Using await Expressions

The await keyword does two things: it suspends the current coroutine’s execution and yields control back to the event loop, which can then run other coroutines. When the awaited operation completes, execution resumes right after the await expression.

import asyncio

async def fetch_user(user_id):
    print(f"Fetching user {user_id}...")
    await asyncio.sleep(2)  # Simulate database query
    return {"id": user_id, "name": f"User{user_id}"}

async def fetch_posts(user_id):
    print(f"Fetching posts for user {user_id}...")
    await asyncio.sleep(1.5)  # Simulate database query
    return [{"id": 1, "title": "Post 1"}, {"id": 2, "title": "Post 2"}]

async def get_user_data(user_id):
    user = await fetch_user(user_id)  # Suspend here for 2s
    print(f"Got user: {user['name']}")
    
    posts = await fetch_posts(user_id)  # Suspend here for 1.5s
    print(f"Got {len(posts)} posts")
    
    return {"user": user, "posts": posts}

asyncio.run(get_user_data(123))
# Total time: ~3.5 seconds (sequential)

Common mistakes include forgetting await, which leaves you with a coroutine object instead of the result, and trying to use await outside an async function, which raises a SyntaxError.

# Wrong - missing await
async def wrong_example():
    result = fetch_user(123)  # Returns coroutine object, not user data
    print(result['name'])  # TypeError: 'coroutine' object is not subscriptable

# Wrong - await outside async function
def also_wrong():
    result = await fetch_user(123)  # SyntaxError

Running Concurrent Coroutines

Sequential await calls miss the point of async programming. To achieve actual concurrency, use asyncio.gather() or asyncio.create_task().

import asyncio
import time

async def fetch_data(source, delay):
    print(f"Fetching from {source}...")
    await asyncio.sleep(delay)
    return f"Data from {source}"

async def sequential_execution():
    start = time.time()
    result1 = await fetch_data("API-1", 2)
    result2 = await fetch_data("API-2", 2)
    result3 = await fetch_data("API-3", 2)
    elapsed = time.time() - start
    print(f"Sequential: {elapsed:.2f}s")
    return [result1, result2, result3]

async def concurrent_execution():
    start = time.time()
    results = await asyncio.gather(
        fetch_data("API-1", 2),
        fetch_data("API-2", 2),
        fetch_data("API-3", 2)
    )
    elapsed = time.time() - start
    print(f"Concurrent: {elapsed:.2f}s")
    return results

# Sequential takes ~6 seconds
asyncio.run(sequential_execution())

# Concurrent takes ~2 seconds
asyncio.run(concurrent_execution())

asyncio.gather() runs multiple coroutines concurrently and returns their results as a list. For fire-and-forget tasks or when you need more control, use asyncio.create_task():

async def background_task():
    await asyncio.sleep(3)
    print("Background task completed")

async def main():
    # Start task in background
    task = asyncio.create_task(background_task())
    
    # Do other work
    await asyncio.sleep(1)
    print("Main work done")
    
    # Wait for background task if needed
    await task

asyncio.run(main())

Practical Patterns and Best Practices

Error handling in async code works like synchronous code, but you need to consider where exceptions might occur:

async def fetch_with_error_handling(url):
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.json()
    except aiohttp.ClientError as e:
        print(f"Request failed: {e}")
        return None
    except asyncio.TimeoutError:
        print("Request timed out")
        return None

Async context managers (async with) ensure proper resource cleanup for async operations:

class AsyncDatabaseConnection:
    async def __aenter__(self):
        print("Opening database connection...")
        await asyncio.sleep(0.5)  # Simulate connection setup
        self.connection = "DB Connection"
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing database connection...")
        await asyncio.sleep(0.2)  # Simulate cleanup
        self.connection = None
    
    async def query(self, sql):
        await asyncio.sleep(0.1)
        return f"Results for: {sql}"

async def use_database():
    async with AsyncDatabaseConnection() as db:
        result = await db.query("SELECT * FROM users")
        print(result)
    # Connection automatically closed here

asyncio.run(use_database())

Async comprehensions and iterators provide clean syntax for async operations:

async def fetch_page(page_num):
    await asyncio.sleep(0.1)
    return f"Page {page_num}"

async def main():
    # Async list comprehension
    pages = [await fetch_page(i) for i in range(5)]
    
    # Or with async for (for async iterators)
    async def page_generator():
        for i in range(5):
            yield await fetch_page(i)
    
    pages = []
    async for page in page_generator():
        pages.append(page)
    
    print(pages)

asyncio.run(main())

Common Pitfalls and Debugging

The most critical mistake is blocking the event loop with synchronous operations. This defeats async’s purpose:

import time

# WRONG - blocks the event loop
async def bad_example():
    time.sleep(2)  # Blocks everything!
    return "Done"

# RIGHT - uses async sleep
async def good_example():
    await asyncio.sleep(2)  # Yields control
    return "Done"

For CPU-bound work, don’t use async—use asyncio.to_thread() or concurrent.futures:

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_intensive_task(n):
    # Heavy computation
    return sum(i * i for i in range(n))

async def handle_cpu_work():
    # Run in separate process to avoid blocking
    with ProcessPoolExecutor() as executor:
        result = await asyncio.get_event_loop().run_in_executor(
            executor, cpu_intensive_task, 10_000_000
        )
    return result

The “coroutine was never awaited” warning means you created a coroutine but never executed it:

async def forgotten():
    await asyncio.sleep(1)
    return "Result"

async def main():
    forgotten()  # RuntimeWarning: coroutine 'forgotten' was never awaited
    # Should be: await forgotten()

asyncio.run(main())

Conclusion and Next Steps

Coroutines with async def and await excel at I/O-bound concurrency. Use them for network requests, file operations, and database queries. Don’t use them for CPU-bound tasks—reach for multiprocessing instead.

Key principles: define coroutines with async def, execute them with await or asyncio.run(), and achieve concurrency with asyncio.gather() or asyncio.create_task(). Avoid blocking the event loop with synchronous operations.

Master these fundamentals, then explore advanced topics: async generators, custom awaitables, asyncio protocols, and libraries like aiohttp, asyncpg, and trio. The asyncio documentation provides comprehensive coverage of the standard library’s capabilities.