Async/Await و asyncio

Async/Await و asyncio انقلابی در برنامه‌نویسی پایتون ایجاد کردند. به‌جای threading سنتی، با concurrency تک‌نخی و event loop می‌توانیم هزاران اتصال شبکه را همزمان مدیریت کنیم. در این فصل با coroutines، tasks، gather، aiohttp و الگوهای async آشنا می‌شویم.

چرا async؟

تصور کنید برنامه‌ای که 100 درخواست HTTP می‌فرستد:

# سنکرون - 100 ثانیه (هر کدام 1 ثانیه)
import requests
import time

start = time.time()
for i in range(100):
    requests.get("https://api.example.com")
print(f"زمان: {time.time() - start:.1f}s")  # ~100s

# آسنکرون - حدود 1 ثانیه!
import asyncio
import aiohttp

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get("https://api.example.com")
            for _ in range(100)
        ]
        await asyncio.gather(*tasks)

asyncio.run(main())  # ~1s

Coroutines – تابع‌های آسنکرون

import asyncio

async def hello():
    print("شروع")
    await asyncio.sleep(1)  # غیر بلوکی
    print("پایان")
    return "نتیجه"

# اجرا
result = asyncio.run(hello())
print(result)
# شروع
# (1 ثانیه صبر)
# پایان
# نتیجه

await – منتظر coroutine

import asyncio

async def fetch_data(url):
    print(f"درخواست {url}")
    await asyncio.sleep(2)  # شبیه‌سازی I/O
    return f"داده {url}"

async def main():
    # ترتیبی - یکی پس از دیگری
    data1 = await fetch_data("url1")
    data2 = await fetch_data("url2")
    print(data1, data2)

asyncio.run(main())  # 4 ثانیه

asyncio.gather – اجرای موازی

async def main():
    # موازی - همزمان
    data1, data2 = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2")
    )
    print(data1, data2)

asyncio.run(main())  # 2 ثانیه (نه 4)

gather با تعداد متغیر

urls = ["url1", "url2", "url3", "url4", "url5"]

async def main():
    results = await asyncio.gather(
        *[fetch_data(url) for url in urls]
    )
    for url, data in zip(urls, results):
        print(url, "→", data)

Tasks – اجرای پس‌زمینه

async def background_job():
    while True:
        print("کار پس‌زمینه...")
        await asyncio.sleep(1)

async def main():
    # شروع task در پس‌زمینه
    task = asyncio.create_task(background_job())
    
    # کارهای دیگر
    await asyncio.sleep(3)
    
    # لغو task
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("لغو شد")

asyncio.run(main())

TaskGroup (Python 3.11+)

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("url1"))
        task2 = tg.create_task(fetch_data("url2"))
        task3 = tg.create_task(fetch_data("url3"))
    # همه به‌طور خودکار await شدند
    print(task1.result())
    print(task2.result())

aiohttp – HTTP Client آسنکرون

pip install aiohttp

import aiohttp
import asyncio

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

async def fetch_json(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    urls = [
        "https://api.github.com/users/torvalds",
        "https://api.github.com/users/gvanrossum",
        "https://api.github.com/users/raymondh",
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_json(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    for r in results:
        print(r["name"])

asyncio.run(main())

POST و headers

async def post_data(session, url, data):
    headers = {"Authorization": "Bearer token123"}
    async with session.post(url, json=data, headers=headers) as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        result = await post_data(
            session,
            "https://api.example.com/users",
            {"name": "علی", "email": "ali@example.com"}
        )
        print(result)

Timeout و خطا

import asyncio

async def main():
    try:
        # timeout 5 ثانیه
        await asyncio.wait_for(
            slow_operation(),
            timeout=5.0
        )
    except asyncio.TimeoutError:
        print("زمان تمام شد!")

# Cancel و cleanup
async def safe_operation():
    try:
        await some_async_work()
    except asyncio.CancelledError:
        print("در حال cleanup...")
        await cleanup()
        raise  # propagate لازم است

محدود کردن همزمانی – Semaphore

async def fetch_with_limit(session, url, semaphore):
    async with semaphore:  # حداکثر N همزمان
        async with session.get(url) as response:
            return await response.text()

async def main():
    # حداکثر 10 درخواست همزمان
    semaphore = asyncio.Semaphore(10)
    
    urls = [f"https://api.example.com/{i}" for i in range(1000)]
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_with_limit(session, url, semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
    
    print(f"تکمیل: {len(results)} درخواست")

Async Iterators و Generators

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.current = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.limit:
            raise StopAsyncIteration
        self.current += 1
        await asyncio.sleep(0.5)
        return self.current

async def main():
    async for num in AsyncCounter(5):
        print(num)

# Async generator (راه ساده‌تر)
async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def main():
    async for i in async_range(5):
        print(i)
    
    # یا با list comprehension
    nums = [i async for i in async_range(5)]

asyncio.Queue – تولیدکننده/مصرف‌کننده

async def producer(queue):
    for i in range(5):
        await queue.put(f"آیتم {i}")
        await asyncio.sleep(0.5)
        print(f"تولید: آیتم {i}")
    await queue.put(None)  # سیگنال پایان

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"مصرف: {item}")
        await asyncio.sleep(1)

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

asyncio.run(main())

Async Context Manager

from contextlib import asynccontextmanager

@asynccontextmanager
async def db_connection(url):
    print("اتصال...")
    conn = await async_connect(url)
    try:
        yield conn
    finally:
        print("قطع اتصال")
        await conn.close()

async def main():
    async with db_connection("postgresql://...") as conn:
        result = await conn.execute("SELECT * FROM users")

ترکیب کد سنکرون و آسنکرون

import asyncio
from concurrent.futures import ThreadPoolExecutor

# اجرای کد سنکرون در thread pool
async def run_in_thread(blocking_func, *args):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, blocking_func, *args)

# مثال
import requests  # کتابخانه سنکرون

def slow_blocking_call(url):
    response = requests.get(url)
    return response.text

async def main():
    # اجرای کد سنکرون بدون بلاک کردن
    text = await run_in_thread(slow_blocking_call, "https://example.com")
    print(text[:100])

asyncio.run(main())

مثال جامع: Web Scraper

import aiohttp
import asyncio
from bs4 import BeautifulSoup

async def fetch_page(session, url):
    async with session.get(url) as response:
        return await response.text()

async def parse_page(session, url, semaphore):
    async with semaphore:
        try:
            html = await fetch_page(session, url)
            soup = BeautifulSoup(html, "html.parser")
            return {
                "url": url,
                "title": soup.title.string if soup.title else "",
                "links": len(soup.find_all("a"))
            }
        except Exception as e:
            return {"url": url, "error": str(e)}

async def main(urls):
    semaphore = asyncio.Semaphore(20)
    timeout = aiohttp.ClientTimeout(total=30)
    
    async with aiohttp.ClientSession(timeout=timeout) as session:
        tasks = [parse_page(session, url, semaphore) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    for r in results:
        if isinstance(r, Exception):
            print(f"خطا: {r}")
        else:
            print(r)

urls = ["https://example.com", "https://python.org", ...]
asyncio.run(main(urls))

بهترین شیوه‌ها

• هرگز time.sleep() در کد async استفاده نکنید – await asyncio.sleep()
• هرگز کد سنکرون بلوکی در coroutine اجرا نکنید (CPU-bound) – از run_in_executor
• برای محدود کردن همزمانی، Semaphore استفاده کنید
• از asyncio.gather با return_exceptions=True برای fail-tolerance
• همیشه async with برای session‌ها
• TaskGroup (3.11+) بهتر از gather برای error handling است

چه زمانی async استفاده نکنیم؟

• کار CPU-bound: محاسبات سنگین، image processing – از multiprocessing استفاده کنید
• اسکریپت‌های ساده: اگر فقط 2-3 کار همزمان دارید، threading کافی است
• کتابخانه‌های سنکرون: اگر کتابخانه async نسخه ندارد

جمع‌بندی

• async/await برای I/O-bound concurrency بسیار قدرتمند است
• asyncio.run() برای ورود به دنیای async
• gather برای اجرای موازی coroutines
• create_task برای اجرای پس‌زمینه
• aiohttp برای درخواست‌های HTTP غیرهمزمان
• Semaphore برای محدود کردن همزمانی
• TaskGroup در Python 3.11+ بهتر از gather