Async Python: async/await ও Event Loop

import asyncio
import time


async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(1)          # database এর অপেক্ষার নকল
    return {"id": user_id, "name": f"User {user_id}"}


async def main() -> None:
    start = time.perf_counter()

    # একের পর এক: ৩ সেকেন্ড
    a = await fetch_user(1)
    b = await fetch_user(2)
    c = await fetch_user(3)
    print(f"একের পর এক: {time.perf_counter() - start:.1f}s")

    # একসাথে: ১ সেকেন্ড
    start = time.perf_counter()
    a, b, c = await asyncio.gather(fetch_user(1), fetch_user(2), fetch_user(3))
    print(f"একসাথে: {time.perf_counter() - start:.1f}s")


asyncio.run(main())

import asyncio
import httpx

CITIES = {
    "ঢাকা": (23.81, 90.41),
    "চট্টগ্রাম": (22.36, 91.78),
    "সিলেট": (24.89, 91.87),
    "রাজশাহী": (24.37, 88.60),
    "খুলনা": (22.85, 89.54),
}


async def get_temp(client: httpx.AsyncClient, city: str, lat: float, lng: float) -> tuple[str, float]:
    r = await client.get(
        "https://api.open-meteo.com/v1/forecast",
        params={"latitude": lat, "longitude": lng, "current": "temperature_2m"},
    )
    r.raise_for_status()
    return city, r.json()["current"]["temperature_2m"]


async def main() -> None:
    async with httpx.AsyncClient(timeout=10) as client:
        tasks = [get_temp(client, c, *loc) for c, loc in CITIES.items()]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for res in results:
        if isinstance(res, Exception):
            print("ব্যর্থ:", res)
        else:
            city, temp = res
            print(f"{city}: {temp}°C")


asyncio.run(main())

sem = asyncio.Semaphore(10)        # একসাথে সর্বোচ্চ ১০টা

async def limited_fetch(client, url):
    async with sem:
        r = await client.get(url)
        return r.status_code

async def main():
    try:
        async with asyncio.timeout(2):         # Python 3.11+
            await slow_operation()
    except TimeoutError:
        print("২ সেকেন্ডে শেষ হয়নি")

import time
import requests            # sync library

async def bad_handler():
    time.sleep(2)                          # ❌ পুরো event loop ২ সেকেন্ড আটকে
    r = requests.get("https://...")        # ❌ sync HTTP, event loop আটকে
    return r.json()

async def good_handler():
    await asyncio.sleep(2)                 # ✓
    async with httpx.AsyncClient() as c:   # ✓ async HTTP
        r = await c.get("https://...")
    return r.json()

import asyncio

def generate_pdf_sync(data) -> bytes:      # ধীর sync কাজ
    ...

async def handler(data):
    pdf = await asyncio.to_thread(generate_pdf_sync, data)   # event loop আটকায় না
    return pdf