البداية: gather هي الرد الأول الخاطئ

لما بتواجهك مجموعة مهام مستقلة، أول ما يخطر في بالك هو asyncio.gather:

async def main():
    await asyncio.gather(
        fetch_a(), fetch_b(), fetch_c(),
    )

وهي مش غلط في حد ذاتها — بس في حالات كتير هي الخيار الخشبي: بتعمل تزامن بنفسك، في حين أن المكتبة اللي هتتكلم معها واعية بالتزامن أكتر منك. غيّر ترتيب تفكيرك:

1. هل فيه أمر واحد للمكتبة بياخد كله؟ → استخدمه.
2. هل المكتبة بتدعم batch / pipeline؟ → استخدمه.
3. هل فيه عميل متزامن (AsyncClient, aiohttp)؟ → استخدم الأصل مباشرة.
4. وفشل كل الخيارات السابقة → gather والحل اليدوي.

Redis Pipeline: التزامن اللي كان جوه المكتبة

أوضح مثال. عملت ميزة كانت محتاجة تعدّ/تقرأ مجموعة مفاتيح، ورد الفعل الأول كان:

results = await asyncio.gather(
    *(client.get(key) for key in keys)
)

صح تقنيًا، بس غلط هندسيًا: كل أمر طلب، وكل واحد رحلة ذهاب وإياب منفصلة لها ثمن. الحل الأصلي اللي عرفت إنه موجود: pipeline بيعمل نفس الدفعة في رحلة شبكة واحدة:

pipe = client.pipeline()
for key in keys:
    pipe.get(key)
results = await pipe.execute()   # طلب واحد، نتايج عشرات

نفس النتيجة، مهندسة صح، وفارق ملحوظ في الربط الفعلي.

httpx limits و aiohttp Connector

مثال تاني من جمع بيانات: عايز أرمي 100 طلب على مصدر واحد. gather بيبعتهم كلهم في نفس اللحظة → الخادم ينهار. الحل مش إدارة semaphore بإيدك فقط، الحل كمان إن العميل نفسه بيقدم حدود أدق (concurrency bound لكل host):

limits = httpx.Limits(max_connections=10, max_keepalive_connections=10)
async with httpx.AsyncClient(limits=limits) as client:
    ...

لما gather صح فعلًا

قاعدة «اسأل المكتبة الأول» بتنطبق هنا كمان: gather تبقى صح لما الشغل مستقل بلا مكتبة بتدير تزامنها — مثلاً مهام محلية (شغل على صور محفوظة) أو طلبات على خدمات متفرقة، لكل واحدة عميل خاص. في الحالات دي gather مش وسيلة مؤقتة — هي الحل الوحيد.

الخلاصة

ابدأ بسؤال «المكتبة بتفهّم التزامن ولا لأ» مش بسؤال «أرمي gather ولا loop». التزامن الأصلي مش لأنه أسرع بس — لأنه بيضع الحدود والقرارات في المكان الصح، ويحتفظ بكود أقل نحتاج نتحكم فيه بإيدنا.