asyncio-programming

name	Asyncio Programming
description	Master asynchronous programming with asyncio, async/await, concurrent operations, and async frameworks
version	2.1.0
sasmp_version	1.3.0
bonded_agent	05-async-concurrency
bond_type	PRIMARY_BOND
retry_strategy	exponential_backoff
observability	{"logging":true,"metrics":"task_completion_rate"}

Asyncio Programming

Overview

Master asynchronous programming in Python with asyncio. Learn to write concurrent code that efficiently handles I/O-bound operations, build async web applications, and understand the async/await paradigm.

Learning Objectives

• Understand asynchronous programming concepts
• Write async functions with async/await syntax
• Manage concurrent operations with asyncio
• Build async web applications
• Handle async I/O operations efficiently
• Debug and test async code

Core Topics

1. Async/Await Basics

• Understanding coroutines
• async/await syntax
• Event loop fundamentals
• Running async functions
• Async vs sync execution
• Common pitfalls

Code Example:

import asyncio
import time

# Synchronous version (slow)
def fetch_data_sync(url):
    print(f"Fetching {url}...")
    time.sleep(2)  # Simulating network delay
    return f"Data from {url}"

def main_sync():
    urls = ['url1', 'url2', 'url3']
    results = []
    for url in urls:
        data = fetch_data_sync(url)
        results.append(data)
    return results

# Takes 6 seconds (2 * 3)
start = time.time()
main_sync()
print(f"Sync took: {time.time() - start:.2f}s")

# Asynchronous version (fast)
async def fetch_data_async(url):
    print(f"Fetching {url}...")
    await asyncio.sleep(2)  # Non-blocking sleep
    return f"Data from {url}"

async def main_async():
    urls = ['url1', 'url2', 'url3']
    # Create tasks and run concurrently
    tasks = [fetch_data_async(url) for url in urls]
    results =  asyncio.gather(*tasks)
     results


start = time.time()
asyncio.run(main_async())
()

2. Asyncio Tasks & Coroutines

• Creating and managing tasks
• asyncio.gather() vs asyncio.wait()
• Task cancellation
• Task groups (Python 3.11+)
• Exception handling in tasks
• Timeouts

Code Example:

import asyncio

async def process_item(item_id, delay):
    print(f"Processing item {item_id}")
    await asyncio.sleep(delay)
    if item_id == 3:
        raise ValueError(f"Item {item_id} failed!")
    return f"Result {item_id}"

async def main():
    # Method 1: gather (returns results in order)
    tasks = [
        process_item(1, 1),
        process_item(2, 2),
        process_item(3, 1),
    ]
    try:
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Task {i} failed: {result}")
            else:
                print(f"Task {i} result: {result}")
    except Exception as e:
        print(f"Error: {e}")

    # Method 2: wait (returns done/pending sets)
    tasks = [
        asyncio.create_task(process_item(i, i))
         i  (, )
    ]
    done, pending =  asyncio.wait(tasks, timeout=)

    ()

    
     task  pending:
        task.cancel()

    
      asyncio.TaskGroup()  tg:
         i  (, ):
            tg.create_task(process_item(i, ))
    

asyncio.run(main())

3. Async I/O Operations

• Async file operations (aiofiles)
• Async HTTP requests (aiohttp)
• Async database operations (asyncpg, motor)
• Async messaging (aio-pika)
• Streams and protocols

Code Example:

import asyncio
import aiohttp
import aiofiles
from typing import List

# Async HTTP requests
async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

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

# Async file operations
async def read_file_async(filepath):
    async with aiofiles.open(filepath, 'r') as f:
        content = await f.read()
        return content

async def write_file_async(filepath, content):
    async with aiofiles.open(filepath, 'w') as f:
        await f.write(content)


 asyncpg

  ():
    conn =  asyncpg.connect(
        user=,
        password=,
        database=,
        host=
    )
    :
        rows =  conn.fetch()
         rows
    :
         conn.close()


  ():
    
    urls = [
        ,
        ,
        ,
    ]
    results =  fetch_multiple_urls(urls)

    
    content =  read_file_async()
     write_file_async(, content.upper())

    
    users =  fetch_users()
    ()

asyncio.run(main())

4. Async Web Frameworks

• FastAPI async routes
• aiohttp web server
• WebSocket handling
• Background tasks
• Middleware and dependencies

Code Example:

# FastAPI async example
from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

# Async route
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Async database call
    user = await fetch_user_from_db(user_id)
    return user

# Background task
async def send_notification(email: str, message: str):
    await asyncio.sleep(2)  # Simulate email sending
    print(f"Sent email to {email}: {message}")

@app.post("/orders/")
async def create_order(order_data: dict, background_tasks: BackgroundTasks):
    # Process order synchronously
    order_id = save_order(order_data)

    # Send notification in background
    background_tasks.add_task(
        send_notification,
        order_data['customer_email'],
        f"Order #{order_id} created"
    )

    return {"order_id": order_id}

# aiohttp web server
from aiohttp import web

  ():
    name = request.match_info.get(, )
     asyncio.sleep()  
     web.json_response({: })

app = web.Application()
app.add_routes([web.get(, handle_request)])


  ():
    ws = web.WebSocketResponse()
     ws.prepare(request)

      msg  ws:
         msg. == web.WSMsgType.TEXT:
             ws.send_str()
         msg. == web.WSMsgType.ERROR:
            ()

     ws

app.add_routes([web.get(, websocket_handler)])

Hands-On Practice

Project 1: Async Web Scraper

Build a concurrent web scraper with rate limiting.

Requirements:

• Scrape multiple websites concurrently
• Implement rate limiting
• Handle errors gracefully
• Save results to async database
• Progress tracking
• Retry failed requests

Key Skills: aiohttp, async I/O, error handling

Project 2: Real-time Chat Server

Create a WebSocket-based chat application.

Requirements:

• WebSocket server with aiohttp
• Multiple chat rooms
• User authentication
• Message broadcasting
• Connection management
• Message history persistence

Key Skills: WebSockets, async server, state management

Project 3: Async Task Queue

Build a distributed task processing system.

Requirements:

• Task queue with Redis/RabbitMQ
• Worker pool management
• Task prioritization
• Result caching
• Progress monitoring
• Graceful shutdown

Key Skills: Message queues, concurrent workers, cleanup

Assessment Criteria

• Understand async/await semantics
• Write efficient concurrent code
• Handle async exceptions properly
• Use asyncio tasks effectively
• Build async web applications
• Debug async code
• Manage async resources (cleanup)

Resources

Official Documentation

• asyncio Docs - Official documentation
• aiohttp - Async HTTP client/server
• FastAPI - Modern async web framework

Learning Platforms

• Using Asyncio in Python - Real Python guide
• Python Concurrency - O'Reilly book
• Asyncio Recipes - Practical examples

Tools

• aiohttp - Async HTTP
• aiofiles - Async file I/O
• asyncpg - Async PostgreSQL
• aiodebug - Async debugging

Next Steps

After mastering asyncio, explore:

• Multiprocessing - CPU-bound parallelism
• Celery - Distributed task queue
• gRPC - Async RPC framework
• Kafka - Async event streaming