Примеры использования

Практические примеры для каждого уровня библиотеки — копируй и запускай.

Установка и ключ

pip install asockslib          # или: uv add asockslib
export ASOCKS_API_KEY="sk-ваш-ключ"

1. One-liner — просто дай прокси

import asyncio
from asockslib import get_proxies, get_proxies_sync

# async
urls = asyncio.run(get_proxies("US", count=10))

# sync (скрипты / notebooks)
urls = get_proxies_sync("US", count=5, verify=True)  # verify=True пингует и оставляет живые
print(urls[0])  # socks5://user:pass@host:port

2. Полный API-клиент

import asyncio
from asockslib import ASocksClient, CreatePortRequest, PortFilterParams

async def main():
    async with ASocksClient(api_key="sk-...") as client:
        balance = await client.get_balance()
        print(f"Баланс: ${balance.balance:.2f}")

        ports = await client.create_ports(
            CreatePortRequest(country_code="US", city="New York", count=5)
        )
        for p in ports:
            print(p.proxy_url)

        existing = await client.list_ports(PortFilterParams(countryName="US"))
        print(f"всего портов: {existing.total}")

asyncio.run(main())

3. Прокси с httpx

import httpx
from asockslib import get_proxies

urls = await get_proxies("DE", count=1)
async with httpx.AsyncClient(proxy=urls[0]) as http:
    resp = await http.get("https://api.ipify.org?format=json")
    print(resp.json())  # IP, который видно через прокси

4. SmartProxy — авто-ротация и самовосстановление

from asockslib import ASocksClient, SmartProxy

async with ASocksClient(api_key="sk-...") as client:
    smart = SmartProxy(client, country_code="US")
    await smart.initialize(pool_size=5)

    url = await smart.get_proxy()   # проверка здоровья; мёртвые прокси меняются сами

5. ProxyPool — 100 000+ аккаунтов, sticky + отказоустойчивость + гео

from asockslib import ASocksClient, ProxyPool, PoolStrategy

async with ASocksClient(api_key="sk-...") as client:
    pool = ProxyPool(
        client,
        country_code="US", city="New York",   # гео-таргетинг: страна + город
        pool_size=500,
        strategy=PoolStrategy.STICKY,          # каждый аккаунт закреплён за своим прокси
    )
    await pool.initialize()

    url = await pool.get_proxy("account_42")   # стабильная привязка на аккаунт
    try:
        ...  # используем url
    except ConnectionError:
        await pool.report_failure("account_42")     # проверка порта через API (без трафика)
        url = await pool.get_proxy("account_42")     # тот же аккаунт, новый прокси того же гео
    await pool.shutdown()

6. Персистентность между перезапусками

class RedisStore:  # храни в Redis / SQL / файле — на твой выбор
    async def load(self) -> dict[str, int]: ...
    async def save(self, account_id: str, port_id: int) -> None: ...
    async def delete(self, account_id: str) -> None: ...

pool = ProxyPool(client, country_code="US", pool_size=1000, store=RedisStore())
await pool.initialize()   # восстанавливает привязки аккаунт -> порт из хранилища

7. Пул по пингу (benchmark on init)

pool = ProxyPool(
    client, country_code="US", pool_size=100,
    benchmark_on_init=True, benchmark_oversample=2.0,  # тестируем ~200, оставляем 100 быстрейших
)
await pool.initialize()

8. Настраиваемые ретраи и типизированные ошибки

from asockslib import ASocksClient, APIConnectionError, RateLimitError, ASocksError

client = ASocksClient(
    api_key="sk-...",
    max_retries=5,          # ретраит 429 / 5xx / сеть с экспоненциальной задержкой
    retry_backoff_base=1.0,
    retry_backoff_max=30.0,
)
try:
    await client.get_balance()
except APIConnectionError:   # сеть недоступна после всех попыток
    ...
except RateLimitError:       # 429 после всех попыток
    ...
except ASocksError as e:     # любая ошибка библиотеки — есть .status_code / .message
    print(e.status_code, e.message)

9. Бенчмарк и выбор самых быстрых

from asockslib import ASocksClient, find_best_proxies

async with ASocksClient(api_key="sk-...") as client:
    result = await find_best_proxies(client, country_code="US", total=100, keep=10)
    print(result.best[0].proxy_url, result.best[0].latency_ms, "мс")