طلب aiohttp واحد عبر وكيل HTTP

لطلب غير متزامن واحد عبر وكيل HTTP دون مصادقة، أبقِ المسار على session.get(). أضف aiohttp إلى المشروع عند الحاجة:

uv add aiohttp

استبدل عنوان التوثيق أدناه بنقطة استماع HTTP لديك أو مرشح من قائمة وكلاء HTTP. يجب أن يسمح بـCONNECT إلى وجهة HTTPS؛ أما نقطة SOCKS فقط فتحتاج إلى موصل مناسب بدل وسيطة proxy هذه.

import asyncio
import aiohttp

async def main():
    timeout = aiohttp.ClientTimeout(total=15, connect=5)
    async with aiohttp.ClientSession(
        timeout=timeout,
        trust_env=False,
    ) as session:
        async with session.get(
            "https://example.com/",
            proxy="http://203.0.113.10:8080",
        ) as response:
            print(response.status)
            print((await response.text())[:200])

asyncio.run(main())

يطبع الطلب حالة HTTP وأول 200 محرف من نص الاستجابة. تبقى 403 أو 429 هنا استجابة قابلة للفحص؛ استدعِ response.raise_for_status() إذا أردت أن يحول التطبيق الحالات غير الناجحة إلى استثناءات. يغلق مديرو السياق غير المتزامنين الجلسة والاستجابة عند انتهاء كتلهم، بما في ذلك عند وقوع استثناء.

تغطي ميزانية total البالغة 15 ثانية عملية الطلب. وتشمل ميزانية connect البالغة 5 ثوانٍ انتظار اتصال متاح من المجمّع، لذا قد يستهلكها مجمّع مشغول قبل فتح مقبس جديد. يهم هذا التمييز عند تحويل الاستدعاء الواحد إلى فحوص متزامنة التنفيذ.

يبقي trust_env=False المسار المعروض صريحًا. لا يكتشف aiohttp متغيرات بيئة الوكيل افتراضيًا؛ وتصدير أحدها في طرفية أخرى لن يهيئ هذا الاستدعاء. اختر trust_env=True فقط عندما تكون البيئة التي ورثتها العملية هي سياسة التوجيه المقصودة.