Amazon Scraper API vs Proxy: Choose a Product Data Source
Compare seller APIs, licensed Amazon product data and proxy page checks by record, access scope and reuse rights.
Read guideConfigure sync or async HTTPX proxies, use trust_env=False to ignore inherited settings, and verify the exit IP and response.

Pass the proxy URL to httpx.Client(proxy=...) or httpx.AsyncClient(proxy=...); an explicit proxy takes precedence over environment proxies. Add trust_env=False to stop the other inherited settings too, such as .netrc credentials and certificate variables. proxy=None does not disable proxies: HTTPX still reads HTTP_PROXY and HTTPS_PROXY unless trust_env=False. Accept a run only when the target body has the expected marker and an IP check through the same client returns an address.
A local test showed the difference. With HTTPS_PROXY pointing at a closed port, httpx.Client(proxy=None) failed with ConnectError, while httpx.Client(trust_env=False) returned HTTP 200 directly.
Audience: Python developers who can run a script and adapt a response check. For browser traffic, use the Playwright proxy guide.
| Goal | Configuration | Result |
|---|---|---|
| Use one explicit proxy | proxy="http://<PROXY_HOST>:<PROXY_PORT>" and trust_env=False |
All client requests use that endpoint |
| Bypass every proxy | trust_env=False, no proxy |
Requests go directly from the worker |
| Follow the environment | Omit proxy, keep trust_env=True |
HTTPX reads HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, and NO_PROXY |
Replace <PROXY_HOST> and <PROXY_PORT> with the assigned endpoint and keep its documented scheme. With an http:// proxy, Basic proxy credentials cross the client-to-proxy hop without encryption, even when the target uses HTTPS. Use a trusted network or a supported TLS proxy endpoint.
Warning: Proxy requests can consume paid traffic, and credentials can grant account access. Start with a sandbox allocation and an authorized test target. Keep credentials out of commands, source, and logs; stop the worker and rotate credentials if they appear in output.
Tested with Python 3.12.0 from the official downloads page and HTTPX 0.28.1. The shell commands use Bash or Zsh on macOS or Linux.
python3 -m venv .venv
.venv/bin/python -m pip install 'httpx==0.28.1'
Set the credential-free endpoint. Set PROXY_USER only when the endpoint requires authentication:
export PROXY_SERVER='http://<PROXY_HOST>:<PROXY_PORT>'
export PROXY_USER='<PROXY_USERNAME>'
For an unauthenticated or IP-allowlisted endpoint, leave PROXY_USER unset. The script reads PROXY_PASSWORD from a runtime secret binding when present and otherwise prompts without echoing the password. Do not paste credentials into shell commands or source.
Save the program as httpx-proxy-check.py. Change the target, marker, and IP-check URL near the top only to values approved for the job; the IP service must return JSON with an ip field without redirecting.
import asyncio
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
import getpass
import ipaddress
import json
import math
import os
from urllib.parse import quote, urlsplit, urlunsplit
import httpx
TARGET_URL = "https://example.com/"
EXPECTED_MARKER = "Example Domain"
IP_CHECK_URL = "https://api.ipify.org?format=json"
TIMEOUT = httpx.Timeout(15.0, connect=5.0)
MAX_ATTEMPTS = 3
MAX_BACKOFF_SECONDS = 2.0
RETRY_STATUSES = {502, 503, 504}
RETRY_EXCEPTIONS = (httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadTimeout)
def proxy_url() -> str:
endpoint = urlsplit(os.environ["PROXY_SERVER"])
if (
endpoint.scheme not in {"http", "https"}
or not endpoint.hostname
or not endpoint.port
or endpoint.username is not None
or endpoint.password is not None
or endpoint.path not in {"", "/"}
or endpoint.query
or endpoint.fragment
):
raise ValueError("PROXY_SERVER needs a credential-free HTTP(S) host and port")
username = os.environ.get("PROXY_USER")
if not username:
return urlunsplit((endpoint.scheme, endpoint.netloc, "", "", ""))
password = os.environ.get("PROXY_PASSWORD")
if password is None:
password = getpass.getpass("Proxy password: ")
if not password:
raise ValueError("Proxy password is required when PROXY_USER is set")
authority = f"{quote(username, safe='')}:{quote(password, safe='')}@{endpoint.netloc}"
return urlunsplit((endpoint.scheme, authority, "", "", ""))
def check_response(response: httpx.Response) -> None:
if 300 <= response.status_code < 400:
raise RuntimeError("redirect received; redirects are disabled")
response.raise_for_status()
def retry_after_seconds(value: str | None) -> float | None:
if value is None:
return None
try:
seconds = float(value)
except ValueError:
try:
retry_at = parsedate_to_datetime(value)
if retry_at.tzinfo is None:
retry_at = retry_at.replace(tzinfo=timezone.utc)
seconds = (retry_at - datetime.now(timezone.utc)).total_seconds()
except (TypeError, ValueError, OverflowError):
return None
if not math.isfinite(seconds) or seconds < 0 or seconds > MAX_BACKOFF_SECONDS:
return None
return seconds
async def fetch(client: httpx.AsyncClient, url: str) -> httpx.Response:
for attempt in range(MAX_ATTEMPTS):
try:
response = await client.get(url)
except httpx.ProxyError:
raise RuntimeError("proxy negotiation failed; check scheme and credentials") from None
except RETRY_EXCEPTIONS:
if attempt + 1 == MAX_ATTEMPTS:
raise RuntimeError("connection failed or timed out after configured attempts") from None
await asyncio.sleep(min(0.5 * 2**attempt, MAX_BACKOFF_SECONDS))
continue
except httpx.HTTPError:
raise RuntimeError("request failed; check proxy, target, and network") from None
if response.status_code in RETRY_STATUSES and attempt + 1 < MAX_ATTEMPTS:
header = response.headers.get("Retry-After")
server_delay = retry_after_seconds(header)
if header is not None and server_delay is None:
return response
await response.aclose()
backoff = min(0.5 * 2**attempt, MAX_BACKOFF_SECONDS)
await asyncio.sleep(min(max(server_delay or 0.0, backoff), MAX_BACKOFF_SECONDS))
continue
return response
raise RuntimeError("request ended without a response")
async def main() -> None:
async with httpx.AsyncClient(
proxy=proxy_url(),
trust_env=False,
timeout=TIMEOUT,
limits=httpx.Limits(max_connections=4, max_keepalive_connections=2),
follow_redirects=False,
) as client:
target = await fetch(client, TARGET_URL)
try:
check_response(target)
if EXPECTED_MARKER not in target.text:
raise RuntimeError("target response did not contain the expected marker")
print(f"Target accepted: HTTP {target.status_code}; marker found")
finally:
await target.aclose()
check = await fetch(client, IP_CHECK_URL)
try:
check_response(check)
ip_data = check.json()
ip_value = ip_data.get("ip") if isinstance(ip_data, dict) else None
if not isinstance(ip_value, str):
raise ValueError("IP endpoint must return a JSON object with a string ip")
address = ipaddress.ip_address(ip_value)
print(f"Observed exit address: {address}")
finally:
await check.aclose()
if __name__ == "__main__":
try:
asyncio.run(main())
except (RuntimeError, ValueError) as error:
# These messages come from this script and contain no credentials.
print(f"Check failed: {error}")
raise SystemExit(1) from None
except (httpx.HTTPError, KeyError) as error:
print(f"Check failed ({type(error).__name__})")
raise SystemExit(1) from None
Run it from the shell where PROXY_SERVER is set:
.venv/bin/python httpx-proxy-check.py
Expected output. The address is a documentation example, not a captured exit:
Target accepted: HTTP 200; marker found
Observed exit address: 203.0.113.10
Runs through a local test proxy, with and without proxy authentication, printed both lines with exit status 0. A closed port printed "Check failed: connection failed or timed out after configured attempts", and a wrong password printed "Check failed: proxy negotiation failed; check scheme and credentials". That proves the script logic, not a provider's routing or location.
proxy_url() builds the authenticated endpoint only in memory. Do not log it or attach it to exception telemetry.
The target passes only with a successful status and the configured marker in the body. The IP check shows which address that endpoint observed; it does not prove location, anonymity, or acceptance by another target.
| Limit | Value in the script | What it does not cover |
|---|---|---|
| Attempts per URL | 3, for connection errors, connect or read timeouts, and HTTP 502, 503, or 504 | Proxy errors, 407, and other statuses are never retried |
| Backoff | Capped at 2 seconds; Retry-After honored only within that cap |
Longer server delays go back to the scheduler |
| Timeouts | 5 seconds connect, 15 seconds default | Not a total job deadline; add an outer worker deadline |
| Connection pool | 4 connections, 2 keep-alive | Concurrency across several workers |
| Response size | No cap; each body is buffered | Large downloads need streaming with a byte cap |
Use the retry helper only for idempotent GET requests. With two URLs the script can make up to 6 proxy requests.
For an approved intercepting CA, import ssl and pass verify=ssl.create_default_context(cafile="/path/to/approved-ca.pem") to AsyncClient. With trust_env=False, HTTPX ignores SSL_CERT_FILE and SSL_CERT_DIR, so the CA must be passed explicitly. Keep certificate verification enabled.
Set trust_env=False and leave proxy unset when a request must go directly from the worker:
# Complete direct-request example
import httpx
TARGET_URL = "https://example.com/"
TIMEOUT = httpx.Timeout(15.0, connect=5.0)
with httpx.Client(trust_env=False, timeout=TIMEOUT) as client:
response = client.get(TARGET_URL, follow_redirects=False)
response.raise_for_status()
print(response.status_code)
Keep this direct client separate from the explicit-proxy client so the route is visible in code review.
| Symptom | First check |
|---|---|
"Check failed: proxy negotiation failed" (ProxyError or HTTP 407) |
Proxy scheme, host, port, and credentials, confirmed with the proxy operator |
ConnectError while proxy=None |
An inherited HTTP_PROXY or HTTPS_PROXY; set trust_env=False |
| "Check failed: connection failed or timed out" | Endpoint reachability, DNS, firewall rules, and the connect timeout |
| TLS verification error | Keep checks enabled and pass the approved CA bundle |
| Redirect or HTTP 401, 403, or 429 | Target policy, origin authentication, and rate limits; do not retry automatically |
| HTTP success without the marker | Inspect the body safely; a challenge or empty page is not accepted data |
| Invalid IP response | The IP service must return a JSON ip field without redirecting |
For SOCKS, install httpx[socks] and use the assigned socks5:// URL when the provider supports it. This script's validator accepts only HTTP(S), so allow socks5 there first. HTTP and SOCKS endpoints are not interchangeable.
HTTPX documents proxy routing, environment variables, SSL configuration, timeouts, connection limits, and exceptions.
Related guides: Python Requests proxy setup, aiohttp proxy setup, and proxy error 407.
Scale your workflows with the tools you already use. Rotating or sticky sessions up to 72 hours.
Compare seller APIs, licensed Amazon product data and proxy page checks by record, access scope and reuse rights.
Read guideChoose the right anti-detect browser proxy guide, map the shared verification steps, and keep account access separate from the network route.
Read guideOur picks for market research: ProxyLane for small, intermittent projects, plus options for regional tracking, monthly collection and managed extraction.
Read guide