> ## Documentation Index
> Fetch the complete documentation index at: https://help.miyaip.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Python proxy setup for web scraping with Requests and Scrapy

> Configure proxies for Python Requests and Scrapy, parse pages, export JSON and CSV, and handle retries and failed requests.

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](/proxy/static-residential/overview), and reuse the proxy and session throughout the task. |
| Independent, stateless collection tasks | Configure location and rotation with [dynamic residential proxies](/proxy/dynamic-residential/overview), and rotate between tasks. |
| Regional tests that require a mobile network exit | Select the relevant carrier network with [mobile proxies](/proxy/mobile/overview). |

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

<Steps>
  <Step title="Create a Python environment">
    Run these commands in a new working directory. SOCKS proxies need the optional dependencies included in `requests[socks]`.

    ```powershell theme={null}
    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](/examples/web_scraping.py) and [retry\_after.py](/examples/retry_after.py), and save both files in the same directory.
  </Step>

  <Step title="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.

    ```powershell theme={null}
    $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.
  </Step>

  <Step title="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.

    <CodeGroup>
      ```python web_scraping.py theme={null}
      """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)
      ```

      ```python retry_after.py theme={null}
      """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
      ```
    </CodeGroup>
  </Step>

  <Step title="Run and inspect the results">
    ```powershell theme={null}
    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.
  </Step>
</Steps>

## 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](/examples/scrapy_example.py) beside `retry_after.py`, and reuse the environment variables above.

```powershell theme={null}
python -m pip install scrapy
New-Item -ItemType Directory -Path output -Force
scrapy runspider scrapy_example.py -O output/scrapy-results.json
```

```python scrapy_example.py theme={null}
"""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](https://requests.readthedocs.io/en/latest/user/advanced/#proxies) and [Scrapy HttpProxyMiddleware](https://docs.scrapy.org/en/latest/topics/downloader-middleware.html#httpproxymiddleware). Header formats are defined in [RFC 9110: Retry-After](https://www.rfc-editor.org/rfc/rfc9110.html#name-retry-after).


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.