أولا: مين صاحب الـ loop؟

أكبر مصدر مشاكل الـ async عندي كان سؤال «مين يدير الحلقة؟». القاعدة الثابتة في النهاية: حلقة واحدة في العملية، ومين يبدأها تحتويها (عشان asyncio.run) — وإلا بتحصل أخطاء غبية في العميل اللي مولود في حلقة تانية.

async def main():
    async with httpx.AsyncClient() as c:   # العميل اتولد داخل الـ loop الشغال
        ...
asyncio.run(main())

لا تفتح AsyncClient في دالة عادية برّا الـ loop. لو محتاج عميلًا عالميًا، صمّمه بإنشائه داخل الـ event loop الحقيقي (أو بعميل لكل loop مع own_session=True زي ما عملنا في ayyad-apis).

المهام الخلفية: أمسك مرجعها ولا تدعها تضيع

لركن مهمة تعمل في الخلفية وأنت مستني، asyncio.create_task مش كافي لوحده — الـ garbage collector ممكن يجمع المهمة لو ما فيش مرجع ليها. المرجع الصحيح، والأهم الحماية من إلغائها عند shutdown:

background = set()
def spawn(task):
    background.add(task)
    task.add_done_callback(background.discard)
    return task

ولحظة الـ CancelledError: لو المهمة بتعمل شغل مهم قبل الإلغاء، استخدم asyncio.shield على الجزء الحساس:

await asyncio.shield(important_step())   # بيمتص الإلغاء ويكمل الشغل المهم

السياق المشترك: أغلِق عليه

عندك loop والـ contextvars بتتوارث عبر المهام. أكبر مصدر مشاكل كان متغيرًا عموميًا بيتتهزّ بين الدوال دي. الحل من تجربتي: عرّف السياق (مثلاً معرّف الطلب) في ContextVar وامسكه بشكل صريح مش global. بأسلوب مشابه لما في loglens، القيمة بتتوارث تلقائيًا لأي مهمة ابنة، من غير ما تعدّيها يدويًا عبر كل دالة.

كشف الـ blocking قبل ما يسدّ الشغل

أي دالة متزامنة (بتعمل requests مباشرة، time.sleep) في وسط async بتجمّد الـ loop بالكامل. الوسيلة العملية اللي أثبتت جودتها:

• الأداة: loop.slow_callback_duration عشان تحذّر من الفانكشن اللي بتعدّي المدة.
• العادة: اكتب await asyncio.sleep(0) بين الجمل في الأماكن اللي ممكن يحصل فيها block مؤقت جوه مكتبة طرف ثالث.

خلاصة

الـ async مش بيشتغل سِحْرًا — بيشتغل بقواعد مِلكية واضحة: مالِك واحد للـ loop، مرجع صحيح للمهام الخلفية، سياق مغلق، ومراقبة مستمرة للـ blocking. لما دي بتتبّع، الـ asyncio بيبقى إصدار ثابت من غير مفاجآت.