Python asyncio Event Loop

import asyncio

async def hello_world():
    print("Hello")
    await asyncio.sleep(1)  # Pause here, let other tasks run
    print("World")

# Coroutines must be awaited or run by the event loop
asyncio.run(hello_world())

import asyncio

async def task(name, delay):
    print(f"Task {name} starting")
    await asyncio.sleep(delay)
    print(f"Task {name} completed after {delay}s")
    return f"Result-{name}"

async def main():
    # The event loop runs multiple coroutines concurrently
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 3)
    )
    print(f"All results: {results}")

# asyncio.run() creates an event loop, runs the coroutine, and closes the loop
asyncio.run(main())

import asyncio
import time

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

async def main():
    start = time.time()

    # Create tasks to run concurrently
    task1 = asyncio.create_task(fetch_data(1, 2))
    task2 = asyncio.create_task(fetch_data(2, 3))
    task3 = asyncio.create_task(fetch_data(3, 1))

    # Wait for all tasks
    results = await asyncio.gather(task1, task2, task3)

    print(f"Results: {results}")
    print(f"Total time: {time.time() - start:.2f}s")  # ~3s, not 6s!

asyncio.run(main())

async def fetch_user(user_id):
    # Simulate API call
    await asyncio.sleep(1)
    return {"id": user_id, "name": f"User-{user_id}"}

async def fetch_posts(user_id):
    # Simulate database query
    await asyncio.sleep(0.5)
    return [f"Post-{i}" for i in range(3)]

async def get_user_data(user_id):
    # Concurrent execution using gather
    user, posts = await asyncio.gather(
        fetch_user(user_id),
        fetch_posts(user_id)
    )
    return {"user": user, "posts": posts}

# Run the async function
result = asyncio.run(get_user_data(123))

async def background_task(name):
    await asyncio.sleep(2)
    print(f"Background task {name} completed")

async def main():
    # Create task but don't await it immediately
    asyncio.create_task(background_task("cleanup"))

    # Do other work
    print("Main work...")
    await asyncio.sleep(1)
    print("Main work done")

    # Give background tasks time to complete
    await asyncio.sleep(2)

asyncio.run(main())

async def slow_operation():
    await asyncio.sleep(10)
    return "Complete"

async def main():
    try:
        # Wait maximum 3 seconds
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
        print(result)
    except asyncio.TimeoutError:
        print("Operation timed out!")

asyncio.run(main())

import asyncio
import random

async def producer(queue, producer_id):
    for i in range(5):
        item = f"Item-{producer_id}-{i}"
        await queue.put(item)
        print(f"Producer {producer_id} added {item}")
        await asyncio.sleep(random.uniform(0.5, 1.5))

async def consumer(queue, consumer_id):
    while True:
        item = await queue.get()
        if item is None:  # Poison pill
            break
        print(f"Consumer {consumer_id} processed {item}")
        await asyncio.sleep(random.uniform(0.2, 0.8))
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=10)

    # Create producers and consumers
    producers = [asyncio.create_task(producer(queue, i)) for i in range(2)]
    consumers = [asyncio.create_task(consumer(queue, i)) for i in range(3)]

    # Wait for producers to finish
    await asyncio.gather(*producers)

    # Wait for queue to be processed
    await queue.join()

    # Stop consumers
    for _ in consumers:
        await queue.put(None)

    await asyncio.gather(*consumers)

asyncio.run(main())

async def rate_limited_request(semaphore, url):
    async with semaphore:  # Acquire semaphore
        print(f"Requesting {url}")
        await asyncio.sleep(1)  # Simulate API call
        return f"Response from {url}"

async def main():
    # Limit to 3 concurrent requests
    semaphore = asyncio.Semaphore(3)

    urls = [f"http://api.example.com/endpoint/{i}" for i in range(10)]

    tasks = [rate_limited_request(semaphore, url) for url in urls]
    results = await asyncio.gather(*tasks)

    print(f"Completed {len(results)} requests")

asyncio.run(main())

import asyncio
import aiohttp
from typing import List, Dict

async def fetch_page(session: aiohttp.ClientSession, url: str) -> Dict:
    """Fetch a single page"""
    try:
        async with session.get(url, timeout=5) as response:
            return {
                "url": url,
                "status": response.status,
                "content": await response.text(),
                "headers": dict(response.headers)
            }
    except asyncio.TimeoutError:
        return {"url": url, "error": "Timeout"}
    except Exception as e:
        return {"url": url, "error": str(e)}

async def fetch_all_pages(urls: List[str]) -> List[Dict]:
    """Fetch multiple pages concurrently"""
    connector = aiohttp.TCPConnector(limit=10)  # Limit connections
    timeout = aiohttp.ClientTimeout(total=30)

    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout
    ) as session:
        tasks = [fetch_page(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def process_pages(urls: List[str]):
    """Process pages with progress reporting"""
    print(f"Fetching {len(urls)} pages...")

    results = await fetch_all_pages(urls)

    successful = [r for r in results if "error" not in r]
    failed = [r for r in results if "error" in r]

    print(f"✅ Success: {len(successful)}")
    print(f"❌ Failed: {len(failed)}")

    for failure in failed:
        print(f"  - {failure['url']}: {failure['error']}")

    return results

# Usage
urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/2",
    "https://httpbin.org/status/200",
    "https://httpbin.org/status/404",
]

results = asyncio.run(process_pages(urls))

class AsyncDatabase:
    async def __aenter__(self):
        print("Connecting to database...")
        await asyncio.sleep(0.5)
        self.connection = "Connected"
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Closing database connection...")
        await asyncio.sleep(0.2)
        self.connection = None

    async def query(self, sql):
        await asyncio.sleep(0.1)
        return f"Results for: {sql}"

async def main():
    async with AsyncDatabase() as db:
        result = await db.query("SELECT * FROM users")
        print(result)

asyncio.run(main())

class AsyncCounter:
    def __init__(self, stop):
        self.current = 0
        self.stop = stop

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current < self.stop:
            await asyncio.sleep(0.1)  # Simulate async work
            self.current += 1
            return self.current
        raise StopAsyncIteration

async def main():
    async for num in AsyncCounter(5):
        print(f"Count: {num}")

asyncio.run(main())

async def fetch_paginated_data(pages):
    """Async generator for paginated API calls"""
    for page in range(1, pages + 1):
        await asyncio.sleep(0.5)  # Simulate API call
        yield {
            "page": page,
            "data": [f"item-{i}" for i in range(5)]
        }

async def process_data():
    async for page_data in fetch_paginated_data(3):
        print(f"Processing page {page_data['page']}")
        # Process data as it arrives
        for item in page_data['data']:
            print(f"  - {item}")

asyncio.run(process_data())