Asynchronous Programming

Asynchronous programming allows you to perform multiple I/O-bound tasks concurrently without blocking execution. Python’s asyncio library provides the foundation for async code using the async and await keywords.

Why Asynchronous Programming?

Synchronous (Blocking) - Sequential Execution

sync_example.py

import time

def fetch_data(data_id: str, delay: float):
    print(f"Fetching {data_id}...")
    time.sleep(delay)  # Blocks execution
    print(f"Fetched {data_id}")
    return data_id

# Sequential execution
start = time.time()
fetch_data("Data 1", 2)
fetch_data("Data 2", 2)
print(f"Total time: {time.time() - start:.2f} seconds")
# Total time: ~4 seconds

Asynchronous (Non-Blocking) - Concurrent Execution

async_example.py

import asyncio
import time

async def fetch_data(data_id: str, delay: float):
    print(f"Fetching {data_id}...")
    await asyncio.sleep(delay)  # Non-blocking
    print(f"Fetched {data_id}")
    return data_id

async def main():
    # Run tasks concurrently
    start = time.time()
    results = await asyncio.gather(
        fetch_data("Data 1", 2),
        fetch_data("Data 2", 2)
    )
    print(f"Total time: {time.time() - start:.2f} seconds")
    print("Results:", results)

asyncio.run(main())
# Total time: ~2 seconds (both tasks run in parallel)

Basic Async Concepts

Defining Async Functions

async_functions.py

import asyncio

async def greet(name: str):
    """Async function - returns a coroutine"""
    await asyncio.sleep(1)  # Simulate I/O operation
    return f"Hello, {name}!"

# Call async function
async def main():
    result = await greet("Alice")
    print(result)

asyncio.run(main())

Running Multiple Tasks

multiple_tasks.py

import asyncio

async def task(name: str, delay: float):
    print(f"Starting {name}...")
    await asyncio.sleep(delay)
    print(f"Finished {name}")
    return f"Result from {name}"

async def main():
    # Run tasks concurrently
    results = await asyncio.gather(
        task("Task 1", 2),
        task("Task 2", 1),
        task("Task 3", 3)
    )
    print("All tasks completed:", results)

asyncio.run(main())
# Tasks run concurrently, not sequentially

Real-World Example: Concurrent API Requests

async_api.py

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession, url: str):
    """Fetch a single URL"""
    async with session.get(url) as response:
        return await response.text()

async def fetch_multiple_urls(urls: list[str]):
    """Fetch multiple URLs concurrently"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

# Usage
async def main():
    urls = [
        "https://api.example.com/data1",
        "https://api.example.com/data2",
        "https://api.example.com/data3"
    ]
    results = await fetch_multiple_urls(urls)
    print(f"Fetched {len(results)} URLs")

asyncio.run(main())

Async Context Managers

You can create async context managers:

async_context.py

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_database_connection(connection_string: str):
    """Async context manager"""
    print("Connecting to database...")
    connection = await connect_to_database(connection_string)
    try:
        yield connection
    finally:
        print("Closing database connection...")
        await connection.close()

async def connect_to_database(connection_string: str):
    # Simulate async connection
    await asyncio.sleep(0.1)
    return {"status": "connected"}

# Usage
async def main():
    async with async_database_connection("postgresql://localhost/db") as db:
        print("Using database:", db)

asyncio.run(main())

Async Classes

You can create classes with async methods:

async_classes.py

import asyncio

class AsyncDataFetcher:
    """Class with async methods"""
    def __init__(self, base_url: str):
        self.base_url = base_url

    async def fetch(self, endpoint: str):
        """Async method"""
        await asyncio.sleep(1)  # Simulate network request
        return f"Data from {self.base_url}/{endpoint}"

    async def fetch_multiple(self, endpoints: list[str]):
        """Fetch multiple endpoints concurrently"""
        tasks = [self.fetch(ep) for ep in endpoints]
        return await asyncio.gather(*tasks)

# Usage
async def main():
    fetcher = AsyncDataFetcher("https://api.example.com")
    results = await fetcher.fetch_multiple(["users", "posts", "comments"])
    print(results)

asyncio.run(main())

When to Use Async

Use async for:

• I/O-bound operations (network requests, file I/O, database queries)
• Multiple concurrent operations
• Web servers and APIs
• Real-time applications

When NOT to Use Async

Don’t use async for:

• CPU-bound operations (use multiprocessing instead)
• Simple sequential operations
• Operations that don’t benefit from concurrency

Remember: Async is for I/O-bound operations. For CPU-bound work, use multiprocessing!

Key Takeaways

Remember

• async def defines an async function (coroutine)
• await waits for an async operation to complete
• asyncio.gather() runs multiple tasks concurrently
• asyncio.run() runs the main async function
• Async is for I/O-bound operations, not CPU-bound
• Use async context managers (@asynccontextmanager) for async resources
• Async enables concurrent execution without threads

Final Thought

Asynchronous programming is powerful for I/O-bound applications where you need to handle many operations concurrently without blocking.