Python proxy setup for web scraping with Requests and Scrapy

Use this tutorial to configure a proxy, request an authorized public page, parse its fields, and export files. Validate the flow with example.com, then substitute an approved target and your operational contact address. Follow the target’s published request limits.

Choose a proxy and session

Task	Selection and configuration
Stable public endpoints that permit collection	Start with a data center proxy or one static endpoint to simplify monitoring and cost accounting.
Tasks that need a fixed region and continuous identity	Extract connection details from static residential proxies, and reuse the proxy and session throughout the task.
Independent, stateless collection tasks	Configure location and rotation with dynamic residential proxies, and rotate between tasks.
Regional tests that require a mobile network exit	Select the relevant carrier network with mobile proxies.

A proxy selects the network exit. You still need request limits, reliable selectors, and record validation. HTTP 200 alone does not establish that the collected fields are complete.

Collect and export with Requests

Create a Python environment

Run these commands in a new working directory. SOCKS proxies need the optional dependencies included in requests[socks].

python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install 'requests[socks]'

On macOS or Linux, activate with . .venv/bin/activate. Download web_scraping.py and retry_after.py, and save both files in the same directory.

Set the target and proxy connection details

The following values are placeholders. Inject the complete proxy URL into SCRAPING_PROXY_URL. Match the protocol, host, port, and authentication details to the proxy you extracted. Use http:// for an HTTP proxy or socks5h:// for a SOCKS5 proxy. Percent-encode special characters in the URL’s username and password.

$env:SCRAPING_PROXY_URL = 'http://YOUR_USERNAME:YOUR_PASSWORD@proxy.example.com:1234'
$env:TARGET_URL = 'https://example.com/'
$env:ALLOWED_HOSTS = 'example.com'
python -c "import os; assert os.environ.get('SCRAPING_PROXY_URL')"

Inject real credentials through your secret-management tool for production tasks. Keep them out of shared scripts and logs. Update ALLOWED_HOSTS when changing the target; the example allows only approved HTTPS hosts on the default port. Replace contact@example.com in the script with your operational contact address.

Save the complete scripts

The main script reuses one requests.Session, sets the proxy for both HTTP and HTTPS requests, extracts the first title and h1, and verifies nonempty fields before exporting JSON and CSV. It disables automatic redirects to prevent an unchecked target change.

"""Collect an approved public HTML page through an explicitly configured proxy."""

import csv
import json
import os
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import urlsplit

import requests

from retry_after import (
    RetryAfterOutOfRange,
    RetryDeferred,
    retry_delay,
    wait_before_retry,
)

MAX_ATTEMPTS = 3
MAX_WAIT_SECONDS = 60  # Waiting budget, not a cap on Retry-After.
TIMEOUT = (10, 30)
TRANSIENT_STATUSES = {429, 500, 502, 503, 504}


class FetchFailed(RuntimeError):
    pass


class PageParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.capture = None
        self.parts = []
        self.title = ""
        self.h1 = ""

    def handle_starttag(self, tag, attrs):
        if not self.capture and tag in ("title", "h1") and not getattr(self, tag):
            self.capture = tag
            self.parts = []

    def handle_data(self, data):
        if self.capture:
            self.parts.append(data)

    def handle_endtag(self, tag):
        if tag == self.capture:
            setattr(self, tag, " ".join("".join(self.parts).split()))
            self.capture = None
            self.parts = []


def fetch(session, url, proxies, wait_budget=None):
    wait_budget = wait_budget if wait_budget is not None else {
        "remaining_seconds": MAX_WAIT_SECONDS
    }
    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            response = session.get(
                url, proxies=proxies, timeout=TIMEOUT, allow_redirects=False
            )
        except requests.exceptions.ProxyError:
            raise FetchFailed("proxy_connection_or_authentication_failed") from None
        except requests.exceptions.Timeout:
            if attempt == MAX_ATTEMPTS:
                raise FetchFailed("timeout_attempts_exhausted") from None
            wait_before_retry(2 ** (attempt - 1), wait_budget)
            continue
        except requests.exceptions.RequestException:
            raise FetchFailed("request_failed") from None

        status = response.status_code
        if status == 407:
            response.close()
            raise FetchFailed("proxy_authentication_failed")
        if status in TRANSIENT_STATUSES:
            delay = retry_delay(response.headers.get("Retry-After"), 2 ** (attempt - 1))
            response.close()
            if attempt == MAX_ATTEMPTS:
                raise RetryDeferred(delay, "maximum_attempts_reached")
            wait_before_retry(delay, wait_budget)
            continue
        if not 200 <= status < 300:
            response.close()
            raise FetchFailed("http_status_{}".format(status))
        return response
    raise FetchFailed("maximum_attempts_reached")


def parse(response, url):
    parser = PageParser()
    parser.feed(response.text)
    if not parser.title or not parser.h1:
        raise FetchFailed("required_fields_missing")
    return {"url": url, "title": parser.title, "h1": parser.h1}


def save(records, output_dir="output"):
    output = Path(output_dir)
    output.mkdir(parents=True, exist_ok=True)
    (output / "results.json").write_text(
        json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    with (output / "results.csv").open("w", newline="", encoding="utf-8-sig") as file:
        writer = csv.DictWriter(file, fieldnames=["url", "title", "h1"])
        writer.writeheader()
        writer.writerows(records)


def main():
    proxy_url = os.environ.get("SCRAPING_PROXY_URL", "")
    proxy = urlsplit(proxy_url)
    if proxy.scheme not in {"http", "https", "socks5", "socks5h"} or not proxy.hostname:
        raise FetchFailed("invalid_or_missing_proxy_configuration")
    url = os.environ.get("TARGET_URL", "https://example.com/")
    target = urlsplit(url)
    allowed_hosts = set(os.environ.get("ALLOWED_HOSTS", "example.com").split(","))
    if target.scheme != "https" or target.hostname not in allowed_hosts:
        raise FetchFailed("target_not_allowlisted")
    if target.username or target.password or target.port not in {None, 443}:
        raise FetchFailed("invalid_target")

    budget = {"remaining_seconds": MAX_WAIT_SECONDS}
    proxies = {"http": proxy_url, "https": proxy_url}
    with requests.Session() as session:
        session.trust_env = False
        session.headers["User-Agent"] = "AuthorizedResearchBot/1.0 (+contact@example.com)"
        response = fetch(session, url, proxies, budget)
        try:
            records = [parse(response, url)]
        finally:
            response.close()
    save(records)
    print(json.dumps({"outcome": "success", "status": response.status_code,
                      "records": len(records), "output": "output/results.json,output/results.csv"}))


if __name__ == "__main__":
    try:
        main()
    except RetryDeferred as error:
        print(json.dumps({"outcome": "deferred", "retry_at": error.retry_at,
                          "reason": error.reason}))
        raise SystemExit(2)
    except RetryAfterOutOfRange:
        print(json.dumps({"outcome": "failed", "reason": "retry_after_out_of_range",
                          "action": "manual_rescheduling_required"}))
        raise SystemExit(1)
    except Exception as error:
        reason = str(error) if isinstance(error, FetchFailed) else type(error).__name__
        print(json.dumps({"outcome": "failed", "reason": reason}))
        raise SystemExit(1)

"""Retry-After parsing shared by the Requests and Scrapy examples."""

import math
import time
from datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime


class RetryAfterOutOfRange(RuntimeError):
    pass


def utc_now():
    return datetime.now(timezone.utc)


def retry_delay(value, fallback, now=None):
    """Return the full requested delay; never shorten a valid Retry-After."""
    now = now or utc_now()
    text = "" if value is None else str(value).strip()
    if text.isascii() and text.isdecimal():
        digits = text.lstrip("0") or "0"
        available = datetime.max.replace(tzinfo=timezone.utc) - now
        largest = str(available.days * 86400 + available.seconds)
        if len(digits) > len(largest) or (len(digits) == len(largest) and digits > largest):
            raise RetryAfterOutOfRange("manual_rescheduling_required")
        return int(digits)
    if not text:
        return fallback
    try:
        retry_at = parsedate_to_datetime(text)
        if retry_at.tzinfo is None:
            retry_at = retry_at.replace(tzinfo=timezone.utc)
        return max(math.ceil((retry_at - now).total_seconds()), 0)
    except (TypeError, ValueError, OverflowError):
        return fallback


class RetryDeferred(RuntimeError):
    def __init__(self, delay, reason, now=None):
        try:
            retry_at = (now or utc_now()) + timedelta(seconds=delay)
        except (OverflowError, ValueError):
            raise RetryAfterOutOfRange("manual_rescheduling_required") from None
        self.retry_at = retry_at.isoformat().replace("+00:00", "Z")
        self.reason = reason
        super().__init__("retry_deferred")


def wait_before_retry(delay, wait_budget, sleep=None):
    if delay > wait_budget["remaining_seconds"]:
        raise RetryDeferred(delay, "waiting_budget_exhausted")
    if delay > 0:
        (sleep or time.sleep)(delay)
    wait_budget["remaining_seconds"] -= delay

Run and inspect the results

python web_scraping.py
python --version
python -m pip show requests

After a successful run, open output/results.json and output/results.csv. Check the url, title, and h1 values, encoding, and record count. The console’s outcome=success reflects your actual run. Apply your duplicate-record rules before sending data to another system.

Use Scrapy

This single-file spider sets the proxy through an actual request’s meta["proxy"]. Its downloader middleware also routes auxiliary requests, including robots rules, through that proxy. This path uses Scrapy’s HttpProxyMiddleware, so provide an HTTP or HTTPS proxy URL. Save scrapy_example.py beside retry_after.py, and reuse the environment variables above.

python -m pip install scrapy
New-Item -ItemType Directory -Path output -Force
scrapy runspider scrapy_example.py -O output/scrapy-results.json

scrapy_example.py

"""Run with: scrapy runspider scrapy_example.py -O output/scrapy-results.json"""

import json
import os
from urllib.parse import urlsplit

import scrapy
from scrapy.exceptions import CloseSpider, IgnoreRequest
from scrapy.utils.defer import deferred_from_coro

from retry_after import RetryAfterOutOfRange, RetryDeferred, retry_delay


class ProxyRouteMiddleware:
    def process_request(self, request, spider):
        if getattr(spider, "_stopped", False):
            raise IgnoreRequest("task_already_stopped")
        # Also proxy framework-generated requests, including robots.txt.
        request.meta["proxy"] = os.environ["SCRAPING_PROXY_URL"]

    def process_response(self, request, response, spider):
        # Apply server waits to auxiliary requests before their own callbacks run.
        if response.status in {407, 429, 500, 502, 503, 504}:
            reason = spider.stop_response(response)
            # Do not await closing here: this download is still active.
            deferred_from_coro(spider.crawler.engine.close_spider_async(reason=reason))
            raise IgnoreRequest(reason)
        return response


class ExampleSpider(scrapy.Spider):
    name = "authorized_example"
    handle_httpstatus_list = [407, 429, 500, 502, 503, 504]
    custom_settings = {
        "HTTPPROXY_ENABLED": True,
        "DOWNLOADER_MIDDLEWARES": {
            __name__ + ".ProxyRouteMiddleware": 740,
            "scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 750,
        },
        "ROBOTSTXT_OBEY": True,
        "AUTOTHROTTLE_ENABLED": True,
        "AUTOTHROTTLE_START_DELAY": 2.0,
        "AUTOTHROTTLE_MAX_DELAY": 60.0,
        "AUTOTHROTTLE_TARGET_CONCURRENCY": 1.0,
        "CONCURRENT_REQUESTS_PER_DOMAIN": 2,
        "DOWNLOAD_DELAY": 1.0,
        "DOWNLOAD_TIMEOUT": 30,
        "RETRY_ENABLED": False,
        "REDIRECT_ENABLED": False,
        "LOG_ENABLED": False,
    }

    async def start(self):
        proxy_url = os.environ.get("SCRAPING_PROXY_URL", "")
        proxy = urlsplit(proxy_url)
        if proxy.scheme not in {"http", "https"} or not proxy.hostname:
            print(json.dumps({"outcome": "failed", "reason": "HTTP_proxy_required"}))
            raise CloseSpider("invalid_proxy_configuration")
        url = os.environ.get("TARGET_URL", "https://example.com/")
        target = urlsplit(url)
        allowed_hosts = set(os.environ.get("ALLOWED_HOSTS", "example.com").split(","))
        if target.scheme != "https" or target.hostname not in allowed_hosts:
            print(json.dumps({"outcome": "failed", "reason": "target_not_allowlisted"}))
            raise CloseSpider("target_not_allowlisted")
        if target.username or target.password or target.port not in {None, 443}:
            raise CloseSpider("invalid_target")
        yield scrapy.Request(
            url, meta={"proxy": proxy_url}, callback=self.parse, errback=self.failed
        )

    def failed(self, failure):
        if getattr(self, "_stopped", False):
            return
        print(json.dumps({"outcome": "failed", "reason": "request_failed"}))
        raise CloseSpider("request_failed")

    def stop_response(self, response):
        reason = "proxy_authentication_failed" if response.status == 407 else "reschedule_required"
        if getattr(self, "_stopped", False):
            return reason
        self._stopped = True
        if response.status == 407:
            print(json.dumps({"outcome": "failed", "reason": "proxy_authentication_failed"}))
        else:
            header = response.headers.get(b"Retry-After", b"").decode("ascii", "replace")
            try:
                deferred = RetryDeferred(retry_delay(header, 1), "reschedule_required")
                print(json.dumps({"outcome": "deferred", "retry_at": deferred.retry_at,
                                  "status": response.status, "reason": deferred.reason}))
            except RetryAfterOutOfRange:
                print(json.dumps({"outcome": "failed", "reason": "retry_after_out_of_range"}))
        return reason

    def parse(self, response):
        if getattr(self, "_stopped", False):
            return
        if response.status in {407, 429, 500, 502, 503, 504}:
            raise CloseSpider(self.stop_response(response))
        record = {"url": response.url,
                  "title": response.css("title::text").get(default="").strip(),
                  "h1": response.css("h1::text").get(default="").strip()}
        if not record["title"] or not record["h1"]:
            raise CloseSpider("required_fields_missing")
        print(json.dumps({"outcome": "success", "status": response.status, "records": 1}))
        yield record

The spider uses asynchronous start(), robots rules, AutoThrottle, and per-domain concurrency limits. Middleware checks every response for authentication failure, 429, and the listed transient server errors, including auxiliary requests such as robots.txt. Rate limits or transient errors report a UTC retry_at, set a stop flag, and schedule crawler closure. Subsequent requests are rejected so an external scheduler can resume later. Automatic retries and redirects are disabled to prevent another component from resending too early. Raw framework logs are disabled; the console reports defined error categories and actual export results.

Control retries and handle failures

The Requests example makes at most 3 attempts per URL. Its connect and read timeouts are 10 and 30 seconds, and its cumulative retry waiting budget is 60 seconds per task. This budget limits how long the current process waits; it does not shorten a valid Retry-After to 60 seconds. The header can contain seconds or an HTTP date. If its delay exceeds the remaining budget, or the last attempt still returns a transient error, the script reports outcome=deferred, a UTC retry_at, and a reason, then exits with code 2. Only missing or invalid headers use bounded exponential backoff. Resume at or after retry_at; the example does not create a schedule.

Observation	Action
407 or `requests.exceptions.ProxyError`	Stop and check protocol, host, port, username format, percent-encoding, account status, and the IP allowlist for your chosen authentication mode instead of resending unchanged invalid configuration.
429, 500, 502, 503, 504	Honor the full `Retry-After`; check concurrency, download intervals, and per-domain requests, and lower the request rate. Defer tasks that exceed the budget instead of immediately changing IP and retrying.
Request timeout	Check gateway connectivity, DNS, and routing, and distinguish connection time from read time. Back off within the attempt and waiting limits, then stop when attempts are exhausted.
Missing `title` or `h1`	Check page structure, content type, target URL, language, and selectors. Isolate empty records and update the parser before accepting data.
`retry_after_out_of_range`	The requested delay exceeds the runtime’s UTC date range; stop and reschedule manually.

Validate before adding a task queue

Start with one authorized request and separately check the actual exit region. Inspect both output files for fields, encoding, and duplicate handling. Use controlled responses to test 407, 429, HTTP dates, and timeouts, and confirm deferred tasks do not retry early. Production tasks also need a queue, target rules, rate limits, session ownership, record validation, and isolation for parsing failures. Keep timestamps, status categories, and durations in logs while excluding proxy credentials. Increase concurrency only after valid-record rates, target responses, and costs satisfy your task constraints. For proxy configuration, see Requests documentation and Scrapy HttpProxyMiddleware. Header formats are defined in RFC 9110: Retry-After.

Getting started

Proxy products

Proxy fundamentals

Relay proxy

Native proxy

AdsPower

Developer integrations

FAQ

Python proxy setup for web scraping with Requests and Scrapy

Choose a proxy and session

Collect and export with Requests

Use Scrapy

Control retries and handle failures

Validate before adding a task queue

​Choose a proxy and session

​Collect and export with Requests

​Use Scrapy

​Control retries and handle failures

​Validate before adding a task queue

Choose a proxy and session

Collect and export with Requests

Use Scrapy

Control retries and handle failures

Validate before adding a task queue