Python Async Programming — asyncio, async/await Explained

import asyncio

async def greet(name: str) -> str:
    await asyncio.sleep(1)  # Simulate I/O delay
    return f"Hello, {name}!"

# Run the coroutine
result = asyncio.run(greet("Alice"))
print(result)  # Hello, Alice! (after ~1 second)

async def task(name: str, delay: float):
    print(f"{name}: start")
    await asyncio.sleep(delay)
    print(f"{name}: end after {delay}s")
    return delay

async def main():
    # Run tasks concurrently
    results = await asyncio.gather(
        task("A", 2),
        task("B", 1),
        task("C", 0.5),
    )
    print(f"All done: {results}")

asyncio.run(main())

async def slow_operation(n: int) -> int:
    await asyncio.sleep(n)
    return n * 2

async def main():
    # Create tasks — they start running immediately
    task1 = asyncio.create_task(slow_operation(2))
    task2 = asyncio.create_task(slow_operation(1))

    # Await results — may return in any order
    print(await task2)  # 2 (finishes first)
    print(await task1)  # 4

asyncio.run(main())

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return {"url": url, "status": response.status, "size": len(await response.text())}

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

def main():
    urls = [
        "https://httpbin.org/html" for _ in range(5)
    ]
    results = asyncio.run(fetch_all(urls))
    for r in results:
        print(f'{r["url"]}: status={r["status"]}, size={r["size"]}')

if __name__ == "__main__":
    main()

import asyncio
import aiohttp
from typing import Optional

class AsyncScraper:
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch(self, session: aiohttp.ClientSession, url: str) -> Optional[str]:
        async with self.semaphore:  # Limit concurrent connections
            try:
                async with session.get(url, timeout=aiohttp.ClientTimeout(10)) as resp:
                    return await resp.text()
            except Exception as e:
                print(f"Failed {url}: {e}")
                return None

    async def scrape_many(self, urls: list[str]) -> list[Optional[str]]:
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch(session, url) for url in urls]
            return await asyncio.gather(*tasks)

    def run(self, urls: list[str]) -> list[Optional[str]]:
        return asyncio.run(self.scrape_many(urls))

scraper = AsyncScraper(max_concurrent=5)
results = scraper.run([f"https://httpbin.org/delay/{i}" for i in range(1, 4)])
print(f"Fetched {len([r for r in results if r])} pages successfully")

async def bad():
    time.sleep(1)  # Blocks the entire event loop!

result = my_coroutine()  # Returns a coroutine object, not the result!

import asyncio
import time

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, time.sleep, 2)
    print("Done")

# Sequential — wrong
result1 = await fetch(url1)
result2 = await fetch(url2)

# Concurrent — correct
result1, result2 = await asyncio.gather(fetch(url1), fetch(url2))

import asyncio
import aiohttp
from pathlib import Path

async def download_file(session: aiohttp.ClientSession, url: str, dest: Path) -> bool:
    try:
        async with session.get(url) as response:
            content = await response.read()
            dest.write_bytes(content)
            print(f"Downloaded {url} -> {dest.name}")
            return True
    except Exception as e:
        print(f"Failed {url}: {e}")
        return False

async def download_all(urls: list[str], dest_dir: str = "/tmp/downloads"):
    Path(dest_dir).mkdir(parents=True, exist_ok=True)
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i, url in enumerate(urls):
            dest = Path(dest_dir) / f"file_{i}.html"
            tasks.append(download_file(session, url, dest))
        return await asyncio.gather(*tasks)

urls = [
    f"https://httpbin.org/html" for _ in range(3)
]
successes = asyncio.run(download_all(urls))
print(f"Successfully downloaded: {sum(successes)}/{len(urls)}")