Async/Await — Python المتقدم

البرمجة غير المتزامنة: async و await

كثير من البرامج تقضي معظم وقتها منتظرة: رد من الشبكة، قراءة من قاعدة بيانات. البرنامج العادي ينتظر كل عملية حتى تنتهي قبل أن يبدأ التالية. البرمجة غير المتزامنة تسمح ببدء عدة عمليات انتظار معاً، والانتقال بينها أثناء الانتظار، باستخدام خيط واحد.

المفهوم بمثال المطبخ

طاهٍ واحد يحضّر 3 أطباق: يضع الماء على النار، وبدل أن يقف ينتظر الغليان يقطّع الخضار للطبق الثاني. طاهٍ واحد (خيط واحد) لكن الوقت الكلي أقل بكثير. هذا لا يفيد إذا كان العمل كله تقطيعاً متواصلاً (حسابات): هناك لا يوجد انتظار لاستغلاله.

الصيغة

import asyncio

async def fetch(name, seconds):
    print(f"بدء {name}")
    await asyncio.sleep(seconds)   # انتظار لا يوقف البرنامج كله
    return f"{name} جاهز"

async def main():
    results = await asyncio.gather(fetch("أ", 2), fetch("ب", 1))
    print(results)

asyncio.run(main())   # يستغرق ~2 ثانية وليس 3

- async def تعرّف coroutine: استدعاؤها لا ينفذها بل ينشئ كائناً قابلاً للانتظار.

- await داخل دالة async: "انتظر هذا، واسمح لغيري بالعمل خلال الانتظار".

- asyncio.run(main()) يشغّل حلقة الأحداث (event loop) من الكود العادي، مرة واحدة في بداية البرنامج.

gather و create_task

- asyncio.gather(*coros): تشغيل عدة عمليات معاً وانتظار نتائجها كلها (بنفس الترتيب).

- asyncio.create_task(coro): بدء عملية في الخلفية والمتابعة، ثم await task لاحقاً.

- asyncio.wait_for(coro, timeout=3): مهلة قصوى.

- asyncio.TaskGroup (Python 3.11+): إدارة مجموعة مهام مع إلغاء البقية عند فشل إحداها.

أخطاء شائعة

1. نسيان await: fetch("أ", 1) بدون await لا تنفذ شيئاً (وتظهر تحذير "coroutine was never awaited").

2. استدعاء دالة متزامنة بطيئة داخل async مثل time.sleep(2) أو requests.get: توقف حلقة الأحداث كلها. استخدم asyncio.sleep ومكتبات async مثل httpx أو aiohttp، أو await asyncio.to_thread(func) للدوال المتزامنة.

3. استخدام async للحسابات الثقيلة: لن تتسرع؛ الدرس التالي يشرح البديل.

أين يُستخدم؟

خوادم الويب الحديثة (FastAPI)، عملاء الشبكة الذين يرسلون مئات الطلبات، روبوتات الدردشة، وأي تطبيق يتعامل مع اتصالات كثيرة متزامنة.

تمرين

اكتب دالة async تحاكي تحميل 5 ملفات بأزمنة مختلفة (asyncio.sleep)، وشغّلها مرة بالتتابع ومرة بـ gather، وقِس الفرق في الوقت بـ time.perf_counter(). ثم أضف مهلة ثانيتين لكل تحميل.

import asyncio

# Coroutine بسيطة
async def say_hello():
    print("مرحباً!")
    await asyncio.sleep(1)  # انتظار غير متزامن
    print("وداعاً!")

# تشغيل
asyncio.run(say_hello())

# عمليات متوازية
async def fetch_data(name, delay):
    print(f"بدء جلب {name}...")
    await asyncio.sleep(delay)
    print(f"انتهى جلب {name}")
    return f"بيانات {name}"

async def main():
    # تنفيذ متتالي (بطيء)
    # result1 = await fetch_data("A", 2)
    # result2 = await fetch_data("B", 1)
    
    # تنفيذ متوازي (سريع)
    results = await asyncio.gather(
        fetch_data("A", 2),
        fetch_data("B", 1),
        fetch_data("C", 3)
    )
    print(results)

asyncio.run(main())

# مع timeout
async def slow_operation():
    await asyncio.sleep(10)
    return "انتهى"

async def with_timeout():
    try:
        result = await asyncio.wait_for(
            slow_operation(),
            timeout=2.0
        )
    except asyncio.TimeoutError:
        print("انتهى الوقت!")

asyncio.run(with_timeout())

# Tasks للتحكم أكثر
async def background_task():
    while True:
        print("عمل في الخلفية...")
        await asyncio.sleep(1)

async def main_with_task():
    task = asyncio.create_task(background_task())
    
    await asyncio.sleep(3)
    task.cancel()  # إلغاء المهمة
    
    try:
        await task
    except asyncio.CancelledError:
        print("تم إلغاء المهمة")

asyncio.run(main_with_task())

# Async Context Manager
class AsyncResource:
    async def __aenter__(self):
        print("فتح المورد")
        await asyncio.sleep(0.5)
        return self
    
    async def __aexit__(self, *args):
        print("إغلاق المورد")
        await asyncio.sleep(0.5)
    
    async def do_something(self):
        print("استخدام المورد")

async def use_resource():
    async with AsyncResource() as resource:
        await resource.do_something()

asyncio.run(use_resource())