Proxy API Integration with Python: Complete Developer Guide

import requests

# ZentisLabs rotating proxy - new IP per request
PROXY = "http://USER:PASS@basic.resi.zentislabs.com:12321"
proxies = {"http": PROXY, "https": PROXY}

response = requests.get(
    "https://httpbin.org/ip",
    proxies=proxies,
    timeout=15
)
print(response.json())
# {"origin": "185.234.xx.xx"} - residential IP

# Sticky sessions: generate a sticky line in the dashboard or API (e.g. Basic,
# 30-minute lifetime) and use it exactly as generated. It is pinned to one
# gateway node, so the session keeps its IP.
STICKY_LINE = "NODE_IP:12321:USER:PASS_country-us_session-checkout123_lifetime-30m"
host, port, user, password = STICKY_LINE.split(":")
STICKY_PROXY = f"http://{user}:{password}@{host}:{port}"
sticky_proxies = {"http": STICKY_PROXY, "https": STICKY_PROXY}

session = requests.Session()
session.proxies = sticky_proxies

# All requests use the same IP
session.get("https://shop.example.com/")           # Visit homepage
session.get("https://shop.example.com/product/1")   # View product
session.post("https://shop.example.com/cart/add")    # Add to cart

# Country-level targeting
proxy_us = "http://USER:PASS_country-us@basic.resi.zentislabs.com:12321"
proxy_de = "http://USER:PASS_country-de@basic.resi.zentislabs.com:12321"
proxy_jp = "http://USER:PASS_country-jp@basic.resi.zentislabs.com:12321"

# City-level targeting (the city follows its state)
proxy_nyc = "http://USER:PASS_country-us_state-newyork-city-newyork@basic.resi.zentislabs.com:12321"
proxy_london = "http://USER:PASS_country-gb_state-england-city-london@basic.resi.zentislabs.com:12321"

import aiohttp
import asyncio

PROXY = "http://basic.resi.zentislabs.com:12321"
AUTH = aiohttp.BasicAuth("USER", "PASS")

async def fetch(session, url):
    try:
        async with session.get(url, proxy=PROXY, proxy_auth=AUTH,
                               timeout=aiohttp.ClientTimeout(total=15)) as resp:
            return await resp.text()
    except Exception as e:
        print(f"Failed {url}: {e}")
        return None

async def main():
    urls = [f"https://example.com/page/{i}" for i in range(100)]
    async with aiohttp.ClientSession() as session:
        semaphore = asyncio.Semaphore(20)
        async def bounded_fetch(url):
            async with semaphore:
                return await fetch(session, url)
        results = await asyncio.gather(*[bounded_fetch(url) for url in urls])
        print(f"Fetched {len([r for r in results if r])} pages successfully")

asyncio.run(main())

# settings.py
DOWNLOADER_MIDDLEWARES = {
    'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': 110,
}

# In your spider
class ProductSpider(scrapy.Spider):
    name = "products"
    custom_settings = {
        'CONCURRENT_REQUESTS': 16,
        'DOWNLOAD_DELAY': 1.5,
        'RANDOMIZE_DOWNLOAD_DELAY': True,
    }

    def start_requests(self):
        for url in self.urls:
            yield scrapy.Request(
                url,
                meta={'proxy': 'http://USER:PASS@basic.resi.zentislabs.com:12321'},
                callback=self.parse
            )

    def parse(self, response):
        yield {
            'title': response.css('h1::text').get(),
            'price': response.css('.price::text').get(),
        }

import requests
from requests.exceptions import ProxyError, Timeout, ConnectionError
import time

def fetch_with_retry(url, max_retries=3, proxies=None):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, proxies=proxies, timeout=15)
            if response.status_code == 200:
                return response
            if response.status_code == 403:
                print(f"Blocked on attempt {attempt + 1}, rotating IP...")
                time.sleep(2 ** attempt)
                continue
            if response.status_code == 429:
                retry_after = int(response.headers.get('Retry-After', 5))
                time.sleep(retry_after)
                continue
        except (ProxyError, Timeout, ConnectionError) as e:
            print(f"Connection error on attempt {attempt + 1}: {e}")
            time.sleep(2 ** attempt)
    return None

# 1. Only download what you need
response = requests.head(url, proxies=proxies)  # Check headers first

# 2. Use compression
headers = {"Accept-Encoding": "gzip, deflate, br"}

# 3. Skip binary content
if 'image' in response.headers.get('Content-Type', ''):
    return  # Don't download images through proxy

# 4. Use conditional requests
headers = {"If-Modified-Since": "Wed, 01 Jan 2026 00:00:00 GMT"}
response = requests.get(url, headers=headers, proxies=proxies)
if response.status_code == 304:
    print("Not modified, using cached version")