Event Loop and Coroutine Execution

Asyncio provides a complete solution for asynchronous I/O programming in Python. Here's an example executing 10 concurrent simulated HTTP requests:

import asyncio
import time

async def fetch_resource(url):
    print(f"Starting fetch for {url}")
    await asyncio.sleep(2)
    print(f"Completed fetch for {url}")

async def main():
    start = time.time()
    tasks = [fetch_resource(f"https://api.example.com/data/{i}") for i in range(10)]
    await asyncio.gather(*tasks)
    print(f"Total execution time: {time.time() - start:.2f} seconds")

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

Retrieving Coroutine Return Values

To obtain return values from coroutines, create tasks and access their results:

import asyncio

async def data_fetcher(url):
    print(f"Fetching data from {url}")
    await asyncio.sleep(2)
    return {"status": "success", "data": "sample_content"}

async def main():
    task = asyncio.create_task(data_fetcher("https://api.example.com"))
    await task
    print(f"Result: {task.result()}")

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

Callback Handling with Partial Functions

Use partial functions to pass parameters to completion callbacks:

import asyncio
from functools import partial

async def process_data(url):
    print(f"Processing {url}")
    await asyncio.sleep(2)
    return f"processed_{url}"

def completion_handler(base_url, future):
    print(f"Completed processing for {base_url}")
    print("Notification sent")

async def main():
    url = "https://api.example.com/items"
    task = asyncio.create_task(process_data(url))
    task.add_done_callback(partial(completion_handler, url))
    await task
    print(f"Final result: {task.result()}")

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

Task Grouping with gather()

The gather() function allows grouping and cancellation of tasks:

import asyncio

async def api_call(endpoint):
    print(f"Calling {endpoint}")
    await asyncio.sleep(2)
    print(f"Received response from {endpoint}")

async def main():
    group_a = [api_call("https://api.a.com/resource") for _ in range(3)]
    group_b = [api_call("https://api.b.com/data") for _ in range(3)]
    
    batch_a = asyncio.gather(*group_a)
    batch_b = asyncio.gather(*group_b)
    
    await asyncio.gather(batch_a, batch_b)

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

Task Cancellation and Control

Handle task cancellation through keyboard interrupts:

import asyncio

async def long_running_operation(duration):
    print(f"Operation running for {duration}s")
    try:
        await asyncio.sleep(duration)
        print(f"Operation completed after {duration}s")
    except asyncio.CancelledError:
        print(f"Operation cancelled after {duration}s")
        raise

async def main():
    operations = [
        long_running_operation(2),
        long_running_operation(3),
        long_running_operation(4)
    ]
    
    try:
        await asyncio.gather(*operations)
    except KeyboardInterrupt:
        print("Cancelling all tasks")
        for task in asyncio.all_tasks():
            if task is not asyncio.current_task():
                task.cancel()
        await asyncio.gather(*operations, return_exceptions=True)

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

Scheduled Callback Execution

Various methods for scheduling callbacks within the event loop:

import asyncio

def event_handler(message, loop):
    print(f"Event: {message} at loop time {loop.time()}")

async def main():
    loop = asyncio.get_event_loop()
    current_time = loop.time()
    
    loop.call_soon(event_handler, "Immediate execution", loop)
    loop.call_at(current_time + 1, event_handler, "Scheduled at +1s", loop)
    loop.call_at(current_time + 2, event_handler, "Scheduled at +2s", loop)
    loop.call_later(3, event_handler, "Delayed by 3s", loop)
    
    await asyncio.sleep(4)

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

Integrating ThreadPool with Asyncio

Combine thread pools with asyncio for blocking I/O operations:

import asyncio
from concurrent.futures import ThreadPoolExecutor
import requests

def blocking_http_call(url):
    response = requests.get(url)
    return response.text[:100]

async def main():
    executor = ThreadPoolExecutor(max_workers=5)
    loop = asyncio.get_event_loop()
    
    urls = [f"https://httpbin.org/delay/{i}" for i in range(5)]
    tasks = [
        loop.run_in_executor(executor, blocking_http_call, url)
        for url in urls
    ]
    
    results = await asyncio.gather(*tasks)
    for url, result in zip(urls, results):
        print(f"URL: {url}, Result: {result}")

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

Asynchronous HTTP Client Implementation

Custom HTTP client using asyncio's socket itnerface:

import asyncio
from urllib.parse import urlparse

async def http_get(url):
    parsed = urlparse(url)
    host, path = parsed.netloc, parsed.path or "/"
    
    reader, writer = await asyncio.open_connection(host, 80)
    request = f"GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n"
    writer.write(request.encode())
    await writer.drain()
    
    response_lines = []
    async for line in reader:
        response_lines.append(line.decode())
    
    writer.close()
    await writer.wait_closed()
    return "\n".join(response_lines)

async def main():
    urls = [f"http://httpbin.org/get?request={i}" for i in range(3)]
    tasks = [http_get(url) for url in urls]
    
    for completed in asyncio.as_completed(tasks):
        result = await completed
        print(f"Response length: {len(result)}")

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

Resource Synchronization in Coroutines

Using locks for shared resource access in asynchronous code:

import asyncio
from asyncio import Lock

shared_cache = {}
access_lock = Lock()

async def cached_resource_fetcher(key):
    async with access_lock:
        if key in shared_cache:
            return shared_cache[key]
        
        print(f"Fetching fresh data for {key}")
        await asyncio.sleep(1)  # Simulate network call
        shared_cache[key] = f"data_for_{key}"
        return shared_cache[key]

async def main():
    keys = ["user_profile", "user_profile", "system_config"]
    tasks = [cached_resource_fetcher(key) for key in keys]
    results = await asyncio.gather(*tasks)
    
    for key, result in zip(keys, results):
        print(f"Key: {key}, Result: {result}")

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

Asynchronous Queue Communication

Imlpementing producer-consumer pattern with asyncio queues:

import asyncio
from asyncio import Queue

async def producer(queue, items):
    for item in items:
        await queue.put(item)
        print(f"Produced: {item}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # Sentinel value

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        await asyncio.sleep(0.2)

async def main():
    queue = Queue(maxsize=3)
    items = [f"item_{i}" for i in range(10)]
    
    await asyncio.gather(
        producer(queue, items),
        consumer(queue)
    )

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