How to Build Your Own Proxy Rotator in Python + aiohttp

Why Build a Proxy Rotator?

When scraping at scale, websites detect and block requests from a single IP. A proxy rotator distributes requests across a pool of proxies, making your traffic appear to come from many different users.

Building your own gives you full control over rotation strategy, error handling, and proxy quality filtering.

The Architecture

Request Queue → Proxy Rotator → Random Proxy Selection → aiohttp Request ↓ Success? → Return Response ↓ Fail? → Mark Proxy → Remove from Pool

Basic Setup

import asyncio
import aiohttp
import random
from dataclasses import dataclass
from typing import Optional

@dataclass
class Proxy:
    url: str
    protocol: str  # http, https, socks5
    failures: int = 0
    max_failures: int = 3

Proxy Pool Manager

class ProxyPool:
    def __init__(self, proxies: list[Proxy]):
        self._proxies = proxies.copy()
        self._alive = proxies.copy()

    def get_random(self) -> Optional[str]:
        if not self._alive:
            self._refill()
        if not self._alive:
            return None
        proxy = random.choice(self._alive)
        return f"{proxy.protocol}://{proxy.url}"

    def mark_failed(self, proxy_url: str):
        for p in self._alive:
            if f"{p.protocol}://{p.url}" == proxy_url:
                p.failures += 1
                if p.failures >= p.max_failures:
                    self._alive.remove(p)
                break

    def mark_success(self, proxy_url: str):
        for p in self._alive:
            if f"{p.protocol}://{p.url}" == proxy_url:
                p.failures = 0
                break

    def _refill(self):
        self._alive = [
            p for p in self._proxies
            if p.failures < p.max_failures
        ]

Async Request Handler

class ProxyRotator:
    def __init__(self, pool: ProxyPool, max_retries: int = 3):
        self.pool = pool
        self.max_retries = max_retries

    async def fetch(self, session: aiohttp.ClientSession,
                    url: str, **kwargs) -> Optional[str]:
        for attempt in range(self.max_retries):
            proxy = self.pool.get_random()
            if not proxy:
                return None

            try:
                async with session.get(
                    url, proxy=proxy, timeout=aiohttp.ClientTimeout(10), **kwargs
                ) as response:
                    if response.status == 200:
                        self.pool.mark_success(proxy)
                        return await response.text()
                    else:
                        self.pool.mark_failed(proxy)
            except Exception:
                self.pool.mark_failed(proxy)

            await asyncio.sleep(0.5 * (attempt + 1))

        return None

Putting It All Together

async def main():
    # Load your proxy list (from Pineapple Proxy API, file, etc.)
    proxy_list = [
        Proxy("123.45.67.89:8080", "http"),
        Proxy("98.76.54.32:3128", "https"),
        Proxy("11.22.33.44:1080", "socks5"),
    ]

    pool = ProxyPool(proxy_list)
    rotator = ProxyRotator(pool, max_retries=3)

    urls = [
        "https://httpbin.org/ip" for _ in range(20)
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [rotator.fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

    for i, result in enumerate(results):
        print(f"Request {i}: {result[:50] if result else 'Failed'}")

asyncio.run(main())

Advanced Features

1. Weighted Random Selection

Prefer faster proxies by weighting selection by response time:

def get_weighted(self) -> Optional[str]:
    if not self._alive:
        return None
    weights = [1.0 / (p.response_time or 1) for p in self._alive]
    proxy = random.choices(self._alive, weights=weights, k=1)[0]
    return f"{proxy.protocol}://{proxy.url}"

2. Concurrent Request Limiting

Avoid flooding a single proxy:

class ThrottledProxy(Proxy):
    concurrent_requests: int = 0
    max_concurrent: int = 2

3. Geo-Aware Routing

Route requests to proxies in specific countries:

def get_by_country(self, country: str) -> Optional[str]:
    candidates = [p for p in self._alive if p.country == country]
    if not candidates:
        return None
    proxy = random.choice(candidates)
    return f"{proxy.protocol}://{proxy.url}"

Using Pineapple Proxy as Your Source

Instead of maintaining your own proxy discovery and validation, use our Proxy List API to get a pre-verified pool:

import requests

response = requests.get(
    "https://pproxy.tech/api/ProxyDemo/Index",
    headers={"Accept": "application/json"}
)
proxies_data = response.json()

proxy_list = [
    Proxy(
        item["discovered"]["proxyPoint"]["ipPort"],
        item["discovered"]["proxyPoint"]["protocol"].lower()
    )
    for item in proxies_data
]

Conclusion

Building a proxy rotator is straightforward with Python and aiohttp. The key is robust error handling, smart retry logic, and a reliable proxy source. Whether you use our API or public lists, the architecture remains the same.

Need a ready-to-use proxy pool? Browse our verified proxies or check our plans.