Python میں async proxy rotation: httpx اور aiohttp (2026 گائیڈ)

شائع کردہ 12 جون 2026 · ≈10 منٹ کا مطالعہ

Async scraping وہ جگہ ہے جہاں proxy سیٹ اپس خاموشی سے بکھر جاتے ہیں۔ requests کے ساتھ آپ ہر call پر ایک IP گھماتے ہیں اور آگے بڑھ جاتے ہیں؛ asyncio کے ساتھ آپ کے پاس اچانک 200 requests در راہ ہوتی ہیں، سب کو اپنا اپنا exit، اپنی اپنی session affinity، اور اپنی اپنی retry چاہیے جب کوئی ہدف ایک 403 پھینکتا ہے۔ اسے غلط کریں اور آپ یا تو ایک IP کو ban تک پیٹ دیتے ہیں یا ہر login session کو session کے دوران گھما کر تباہ کر دیتے ہیں۔

یہ گائیڈ وہ patterns دکھاتی ہے جو واقعی concurrency پر ٹکے رہتے ہیں: httpx اور aiohttp میں per-request rotation، ایک محدود worker pool، stateful flows کے لیے sticky sessions، اور block-aware retries — کام کرنے والے کوڈ کے ساتھ جسے آپ paste کر سکتے ہیں۔

Async Proxy کے مسئلے کو کیوں بدل دیتا ہے

ایک synchronous scraper ایک وقت میں ایک proxy کو چھوتا ہے، لہٰذا "ہر request پر گھمائیں" معمولی طور پر درست ہے۔ Async ایک ساتھ تین مفروضے توڑ دیتا ہے:

httpx: Per-Request Proxy Rotation

httpx.AsyncClient ہر client کے لیے ایک proxy باندھتا ہے، لہٰذا گھمانے کے لیے آپ ہر request پر proxy منتخب کرتے ہیں اور exit کے لحاظ سے keyed clients کے ایک چھوٹے pool کے ذریعے route کرتے ہیں:

import asyncio, itertools, httpx

PROXIES = [
    "socks5h://USERNAME:[email protected]:913",
    "socks5h://USERNAME:[email protected]:913",
    "socks5h://USERNAME:[email protected]:913",
]
pool = itertools.cycle(PROXIES)

# One reusable client per exit (connection pooling stays intact)
clients = {p: httpx.AsyncClient(proxy=p, timeout=30) for p in PROXIES}

async def fetch(url):
    proxy = next(pool)
    r = await clients[proxy].get(url)
    return r.status_code, r.text

async def main(urls):
    results = await asyncio.gather(*(fetch(u) for u in urls))
    for client in clients.values():
        await client.aclose()
    return results

ہر exit کے لیے ایک client کو دوبارہ استعمال کرنا HTTP/2 اور connection pooling کو زندہ رکھتا ہے بجائے ہر request پر ایک تازہ handshake کی قیمت ادا کرنے کے۔ نوٹ کریں httpx>=0.28 نے client پر proxies= کا نام بدل کر proxy= کر دیا۔

aiohttp: Request کے ذریعے Rotation

aiohttp یہاں سادہ ہے — ایک واحد ClientSession ہر request پر ایک proxy= دلیل لیتا ہے، لہٰذا آپ inline گھماتے ہیں:

import asyncio, itertools, aiohttp

pool = itertools.cycle(PROXIES)  # http:// or socks5h:// (needs aiohttp-socks for SOCKS)

async def fetch(session, url):
    proxy = next(pool)
    async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=30)) as r:
        return r.status, await r.text()

async def main(urls):
    async with aiohttp.ClientSession() as session:
        return await asyncio.gather(*(fetch(session, u) for u in urls))

aiohttp HTTP proxies کو مقامی طور پر بولتا ہے؛ socks5h:// کے لیے aiohttp-socks شامل کریں اور ایک ProxyConnector پاس کریں۔

ایک Semaphore کے ساتھ Concurrency کو محدود کریں

غیر محدود gather آپ کے pool میں ہر IP کو ایک ساتھ flag کرانے کا تیز ترین طریقہ ہے۔ اسے محدود کریں تاکہ ہر exit انسانی طور پر قابلِ یقین تعداد میں متوازی requests اٹھائے:

sem = asyncio.Semaphore(10)   # at most 10 in flight

async def fetch_capped(session, url):
    async with sem:
        return await fetch(session, url)

اصول: concurrency کو آپ کے پاس موجود الگ exits کی تعداد پر یا اس سے کم رکھیں، تاکہ آپ ایک واحد residential IP پر بہت سی متوازی requests کا ڈھیر نہ لگائیں۔

Stateful Flows کے لیے Sticky Sessions

Rotation اندر جانے کے لیے ہے؛ sticky sessions اندر رہنے کے لیے ہیں۔ ایک login، ایک cart، کوئی بھی cookie-bound چیز کو پوری ترتیب کے لیے ایک exit رکھنا چاہیے۔ proxy کو per request کے بجائے per task پن کریں:

async def run_account(account, proxy):
    # Same exit IP for every step of this account's flow
    async with httpx.AsyncClient(proxy=proxy, timeout=30) as client:
        await client.post("https://target.site/login", data=account.creds)
        await client.get("https://target.site/dashboard")
        await client.post("https://target.site/cart", json=account.order)

async def main(accounts):
    # One sticky exit per account, accounts run concurrently
    await asyncio.gather(*(run_account(a, p)
                           for a, p in zip(accounts, itertools.cycle(PROXIES))))

ایسا provider استعمال کریں جو sticky residential sessions کو سپورٹ کرتا ہو تاکہ task کی پوری مدت کے لیے وہی exit رکھا جائے، آپ کے نیچے خاموشی سے گھمایا نہ جائے۔

Block-Aware Retries

Concurrency پر آپ کسی 200 پر بھروسہ نہیں کر سکتے — anti-bot سسٹمز ایک challenge body کے ساتھ ایک 200 واپس کرتے ہیں۔ blocks کا پتہ لگائیں اور واحد ناکام task کو ایک تازہ exit پر retry کریں:

BLOCK_MARKERS = ("just a moment", "/cdn-cgi/challenge-platform",
                 "datadome", "px-captcha", "access denied")

def looks_blocked(status, text):
    return status in (403, 429, 503) or any(m in text[:4000].lower() for m in BLOCK_MARKERS)

async def fetch_retry(url, attempts=4):
    for _ in range(attempts):
        proxy = next(pool)
        async with httpx.AsyncClient(proxy=proxy, timeout=30) as c:
            r = await c.get(url)
            if not looks_blocked(r.status_code, r.text):
                return r
        await asyncio.sleep(0.5)
    raise RuntimeError(f"blocked after {attempts} exits: {url}")

Fingerprint کو پھر بھی میچ کرنا ہوگا

ایک پھنسا ہوا معاملہ جسے async حل نہیں کرتا: httpx اور aiohttp دونوں Python کے TLS stack پر بیٹھتے ہیں، لہٰذا وہ ایک ایسا JA3 بھیجتے ہیں جو کوئی حقیقی براؤزر پیدا نہیں کرتا۔ کسی صاف residential IP پر ایک بہترین rotation pool بھی handshake پر automation کے طور پر fingerprint ہو جاتا ہے۔ async rotation کو TLS impersonation کے ساتھ جوڑیں — دیکھیں curl_cffi کے ساتھ TLS Fingerprinting بائی پاس کریں، اور www.jibaoproxy.com/tools/fingerprint.html پر اپنے exit کے JA3 + ASN کی تصدیق کریں۔

چیک لسٹ

تمام IP پروڈکٹس کے لیے · ہر وقت دستیاب نوڈز کا وسیع پول

ابھی رجسٹر کریں اور ری چارج پر 100% تک کیش بیک حاصل کریں

نئے صارفین کو سائن اپ پر 500MB ملتے ہیں، اس کے علاوہ پہلے ری چارج پر بونس۔ پیشکش محدود وقت کے لیے ہے۔

اکاؤنٹ بنائیں