Python

import requests

PROXY = "http://USERNAME-country-US-session-job1-time-30:PASSWORD@GATEWAY_HOST:58971"
PROXIES = {"http": PROXY, "https": PROXY}

response = requests.get(
    "https://api.ipify.org?format=json",
    proxies=PROXIES,
    timeout=30,
)
response.raise_for_status()
print(response.json())

import requests

PROXY = "http://USERNAME-country-US-session-job1-time-30:PASSWORD@GATEWAY_HOST:58971"

with requests.Session() as session:
    session.proxies = {"http": PROXY, "https": PROXY}
    session.headers["User-Agent"] = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"

    for page in range(1, 6):
        response = session.get(f"https://example.com/list?page={page}", timeout=30)
        print(page, response.status_code)

import requests

PROXY = "http://USERNAME-country-US:PASSWORD@GATEWAY_HOST:58971"
PROXIES = {"http": PROXY, "https": PROXY}

for _ in range(5):
    # New Session each time, closed immediately, so no pool reuse
    with requests.Session() as session:
        print(session.get("https://api.ipify.org", proxies=PROXIES, timeout=30).text)

import concurrent.futures

import requests

GATEWAY = "GATEWAY_HOST:58971"
USER = "USERNAME"
PASSWORD = "PASSWORD"


def proxy_for(session_id: str, country: str = "US", minutes: int = 30) -> str:
    user = f"{USER}-country-{country}-session-{session_id}-time-{minutes}"
    return f"http://{user}:{PASSWORD}@{GATEWAY}"


def fetch(task_id: int) -> tuple[int, str]:
    proxy = proxy_for(f"task.{task_id}")
    with requests.Session() as session:
        session.proxies = {"http": proxy, "https": proxy}
        return task_id, session.get("https://api.ipify.org", timeout=30).text


with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
    for task_id, ip in pool.map(fetch, range(8)):
        print(task_id, ip)

pip install 'requests[socks]'

import requests

# socks5h lets the proxy resolve the hostname
PROXY = "socks5h://USERNAME-country-US:PASSWORD@GATEWAY_HOST:58971"

print(requests.get(
    "https://api.ipify.org",
    proxies={"http": PROXY, "https": PROXY},
    timeout=30,
).text)

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def build_session(proxy: str) -> requests.Session:
    session = requests.Session()
    session.proxies = {"http": proxy, "https": proxy}

    retry = Retry(
        total=3,
        backoff_factor=1.5,
        # Retry only requests safe to replay; handle proxy handshake exceptions separately
        status_forcelist=(429, 502, 503, 504),
        allowed_methods=frozenset(["GET", "HEAD", "POST"]),
    )
    adapter = HTTPAdapter(max_retries=retry, pool_maxsize=32)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


PROXY = "http://USERNAME-country-US:PASSWORD@GATEWAY_HOST:58971"
with build_session(PROXY) as session:
    print(session.get("https://api.ipify.org", timeout=30).text)

import httpx

PROXY = "http://USERNAME-country-US-session-job1-time-30:PASSWORD@GATEWAY_HOST:58971"

with httpx.Client(proxy=PROXY, timeout=30.0) as client:
    print(client.get("https://api.ipify.org").text)

import asyncio

import httpx

PROXY = "http://USERNAME-country-US:PASSWORD@GATEWAY_HOST:58971"


async def main() -> None:
    limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
    async with httpx.AsyncClient(proxy=PROXY, limits=limits, timeout=30.0) as client:
        urls = [f"https://example.com/item/{i}" for i in range(20)]
        responses = await asyncio.gather(*(client.get(url) for url in urls))
        for response in responses:
            print(response.status_code)


asyncio.run(main())

import asyncio

import aiohttp

GATEWAY = "http://GATEWAY_HOST:58971"
AUTH = aiohttp.BasicAuth(
    "USERNAME-country-US-session-job1-time-30",
    "PASSWORD",
)


async def main() -> None:
    timeout = aiohttp.ClientTimeout(total=30)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(
            "https://api.ipify.org",
            proxy=GATEWAY,
            proxy_auth=AUTH,
        ) as response:
            print(await response.text())


asyncio.run(main())

DOWNLOADER_MIDDLEWARES = {
    "scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 750,
    "myproject.middlewares.NextProxyMiddleware": 751,
}

CONCURRENT_REQUESTS = 32
CONCURRENT_REQUESTS_PER_DOMAIN = 16
DOWNLOAD_TIMEOUT = 30
RETRY_TIMES = 3
RETRY_HTTP_CODES = [429, 500, 502, 503, 504]

import base64
import itertools


class NextProxyMiddleware:
    """Assign a rotating session id per request for controlled IP rotation."""

    GATEWAY = "GATEWAY_HOST:58971"
    USER = "USERNAME"
    PASSWORD = "PASSWORD"
    COUNTRY = "US"
    POOL_SIZE = 16

    def __init__(self) -> None:
        self._slots = itertools.cycle(range(self.POOL_SIZE))

    def process_request(self, request, spider) -> None:
        slot = next(self._slots)
        user = (
            f"{self.USER}-country-{self.COUNTRY}"
            f"-session-slot.{slot}-time-10"
        )
        credentials = base64.b64encode(
            f"{user}:{self.PASSWORD}".encode()
        ).decode()

        request.meta["proxy"] = f"http://{self.GATEWAY}"
        request.headers["Proxy-Authorization"] = f"Basic {credentials}"

import requests

API = "https://api.example.com/api/v1/proxy/extract"


def fetch_proxy_pool(api_key: str, count: int = 10) -> list[str]:
    response = requests.get(
        API,
        params={
            "apikey": api_key,
            "num": count,
            "type": "json",
            "country": "US",
            "session": "sticky",
            "time": 10,
        },
        timeout=15,
    )
    response.raise_for_status()
    payload = response.json()
    return [f"http://{item['ip']}:{item['port']}" for item in payload["data"]]


# Passwordless ports, no credentials needed
pool = fetch_proxy_pool("YOUR_API_KEY", count=5)
for proxy in pool:
    print(requests.get(
        "https://api.ipify.org",
        proxies={"http": proxy, "https": proxy},
        timeout=30,
    ).text)