Smart Proxy Manager
The SmartProxy class wraps ASocksClient to provide a high-level proxy pool with automatic rotation, health checking, and self-healing.
How it works
Section titled “How it works”- Initialize a pool of proxies matching your criteria (country, protocol, type).
- Request a proxy —
SmartProxyreturns the next healthy proxy URL. - Auto-heal — if a proxy fails, it is automatically replaced. No fatal errors are raised.
Basic usage
Section titled “Basic usage”import asynciofrom asockslib import ASocksClient, SmartProxy
async def main(): async with ASocksClient(api_key="sk-...") as client: smart = SmartProxy(client, country_code="US") await smart.initialize(pool_size=5)
proxy_url = await smart.get_proxy() print(proxy_url) # socks5://user:pass@host:port
asyncio.run(main())Pool management
Section titled “Pool management”Get all proxy URLs
Section titled “Get all proxy URLs”urls = await smart.get_all_proxies()Health-check the entire pool
Section titled “Health-check the entire pool”results = await smart.health_check_all()# {1001: True, 1002: False, 1003: True}Refresh unhealthy proxies
Section titled “Refresh unhealthy proxies”replaced = await smart.refresh_pool()print(f"{replaced} proxies were replaced")Configuration
Section titled “Configuration”| Parameter | Type | Default | Description |
|---|---|---|---|
client | ASocksClient | required | The API client instance |
country_code | str | "" | ISO country code filter |
city | str | "" | City filter |
health_check_url | str | https://api.ipify.org?format=json | URL used for health checks |
health_timeout | float | 10.0 | Health check timeout in seconds |
Using with httpx
Section titled “Using with httpx”import httpxfrom asockslib import ASocksClient, SmartProxy
async def scrape(): async with ASocksClient(api_key="sk-...") as client: smart = SmartProxy(client, country="US") await smart.initialize(pool_size=3)
proxy_url = await smart.get_proxy() async with httpx.AsyncClient(proxy=proxy_url) as http: resp = await http.get("https://example.com") print(resp.status_code)Error handling
Section titled “Error handling”from asockslib.exceptions import NoAvailableProxyError
try: proxy = await smart.get_proxy()except NoAvailableProxyError: print("All proxies failed — check your account or criteria")