#Python#FastAPI#Playwright#Scraping#API

Build a Web Scraping API with Python

Build a web scraping API with Python, FastAPI, Playwright and a queue: resilient extraction, rotating proxies and a clean REST interface.

Build a production-grade scraping API: fetch pages reliably, extract structured data, and expose it over REST — without getting blocked or breaking ToS.

What we’re building

flowchart LR
    Client --> API[FastAPI]
    API --> Queue[(Redis queue)]
    Queue --> Worker[Playwright worker]
    Worker --> Target[Target site]
    Worker --> Extractor[Parsing + validation]
    API --> Client

What you’ll learn

  • Politeness and reliability scraping patterns (rate limits, retries, validation)
  • Structured extraction instead of regex soup
  • An API shape clients can actually build on — with API-key auth, validation and typed errors

Only scrape public data you’re allowed to use: respect robots.txt, ToS and rate limits. This guide builds the tooling, not a lawless process. Blocked domains, login-walled content and personal data are off-limits.

Prerequisites

  • Python 3.12+
  • Redis 7
  • Docker

1. Project setup

mkdir scraper-api && cd scraper-api
python -m venv .venv && source .venv/bin/activate
pip install "fastapi" "uvicorn[standard]" playwright redis rq pydantic tenacity httpx
playwright install --with-deps chromium

2. The API

main.py — Pydantic validation, API-key auth, SSRF guard, typed errors and a global exception handler:

import os
import secrets

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from redis import Redis
from rq import Queue
from rq.job import Job

from schemas import ScrapeRequest

app = FastAPI()
q = Queue("scrape", connection=Redis(host=os.getenv("REDIS_HOST", "localhost")))


def require_key(x_api_key: str = Header(default="")) -> None:
    expected = os.getenv("API_KEY", "")
    if not expected or not secrets.compare_digest(x_api_key, expected):
        raise HTTPException(status_code=401, detail="Invalid or missing API key.")


def is_private(url: str) -> bool:
    try:
        parsed = httpx.URL(url)
        host = parsed.host or ""
        return host in ("localhost",) or host.startswith("127.") or host.startswith("10.") \
            or host.startswith("192.168.") or host.startswith("169.254.") or host.endswith(".local")
    except Exception:
        return True


@app.post("/scrape", dependencies=[Depends(require_key)])
async def scrape(req: ScrapeRequest):
    if is_private(req.url):
        raise HTTPException(status_code=400, detail="Private/internal targets are not allowed (SSRF guard).")
    try:
        job = q.enqueue("worker.scrape_job", req.model_dump(), job_timeout=60, result_ttl=3600)
    except Exception as e:  # Redis down
        raise HTTPException(status_code=503, detail=f"Queue unavailable: {e}")
    return {"job_id": job.id, "status": "/jobs/" + job.id}


@app.get("/jobs/{job_id}", dependencies=[Depends(require_key)])
async def status(job_id: str):
    try:
        job = q.fetch_job(job_id)
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Queue unavailable: {e}")
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found.")
    if job.is_failed:
        raise HTTPException(status_code=502, detail=str(job.exc_info or "Job failed."))
    return {"status": job.get_status(), "result": job.result}


@app.exception_handler(Exception)
async def unhandled(_req, exc):
    return JSONResponse(status_code=500, content={"error": f"Internal error: {exc}"})


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("main:app", host="0.0.0.0", port=8000)

schema.py:

from pydantic import BaseModel, Field, HttpUrl


class ScrapeRequest(BaseModel):
    url: HttpUrl
    selectors: dict[str, str] = Field(default_factory=dict, description="CSS selector per result key")


class ScrapeResult(BaseModel):
    url: str
    values: dict[str, str]
    scraped_at: str

3. The worker

worker.py — one browser launch per job (clean memory), field existence validation, retriable failures:

import os
import time
from datetime import datetime, timezone

from playwright.sync_api import sync_playwright
from redis import Redis
from rq import Queue

from schemas import ScrapeResult

q = Queue("scrape", connection=Redis(host=os.getenv("REDIS_HOST", "localhost")))


class ScrapeError(Exception):
    pass


def _page_text_value(page, selector: str) -> str:
    el = page.locator(selector).first
    if el.count() == 0:
        raise ScrapeError(f"Selector not found: {selector}")
    return el.inner_text(timeout=5_000).strip()


def scrape_job(payload: dict):
    url = payload["url"]
    selectors = payload["selectors"]
    for attempt in range(1, 4):  # 3 attempts, backoff below
        try:
            with sync_playwright() as p:
                browser = p.chromium.launch(headless=True)
                page = browser.new_page(user_agent="Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36")
                page.goto(url, wait_until="domcontentloaded", timeout=30_000)
                values = {key: _page_text_value(page, sel) for key, sel in selectors.items()}
                browser.close()
            if any(not v for v in values.values()):
                raise ScrapeError(f"Empty extraction for: {[k for k, v in values.items() if not v]}")
            return ScrapeResult(url=url, values=values, scraped_at=datetime.now(timezone.utc).isoformat()).model_dump()
        except ScrapeError:
            raise  # not transient: selector/empty — do not retry silently
        except Exception:
            if attempt == 3:
                raise
            time.sleep(2 * attempt)  # 2s, 4s backoff

Per-domain pacing lives in the scheduler, not the worker: enqueue jobs for the same domain at most one per DOMAIN_DELAY seconds (see scheduler.py below). This is what turns scraping into a product instead of a ban magnet.

scheduler.py (optional but recommended) — throttles enqueue per domain:

import hashlib
import os
import time

from redis import Redis
from rq import Queue

r = Redis(host=os.getenv("REDIS_HOST", "localhost"))
q = Queue("scrape", connection=r)
DOMAIN_DELAY = int(os.getenv("DOMAIN_DELAY", "5"))


def enqueue_paced(payload: dict) -> str:
    key = f"last:{hashlib.md5(payload['url'].split('/')[2].encode()).hexdigest()}"
    last = float(r.get(key) or 0)
    wait = DOMAIN_DELAY - (time.time() - last)
    if wait > 0:
        time.sleep(wait)
    job = q.enqueue("worker.scrape_job", payload, job_timeout=60, result_ttl=3600)
    r.set(key, time.time())
    return job.id

Robots.txt check: before the first job for a domain, fetch https://<domain>/robots.txt and cache the disallowed path prefixes in Redis; skip matching jobs with a clear “disallowed by robots.txt” error instead of scraping anyway. Minimal implementation:

# robots.py
import os

import httpx
from redis import Redis

r = Redis(host=os.getenv("REDIS_HOST", "localhost"))


def is_disallowed(url: str, fetch=httpx.get) -> bool:
    host = url.split("/")[2]
    cache_key = f"robots:{host}"
    cached = r.get(cache_key)
    if cached is None:
        try:
            cached = fetch(f"https://{host}/robots.txt", timeout=5).text
        except Exception:
            cached = ""
        r.set(cache_key, cached, ex=3600)  # 1h cache
    disallowed = [line.split(":", 1)[1].strip() for line in cached.splitlines() if line.lower().startswith("disallow:")]
    path = "/" + "/".join(url.split("/")[3:]) if len(url.split("/")) > 3 else "/"
    return any(path.startswith(p) for p in disallowed if p)

Call is_disallowed(req.url) in the API before enqueueing and return 403 with a clear message when it returns True.

4. Resilience (the part that separates a toy from a product)

  • Retries: exponential backoff on transient failures only (network, 5xx, timeouts — never selector/empty extractions, which re-raise immediately).
  • Rate limiting: one job per domain per DOMAIN_DELAY seconds, enforced by the scheduler.
  • Validation: every selector key must resolve to non-empty text before returning; empty = explicit ScrapeError.
  • SSRF guard: the API rejects private/loopback targets before the browser ever starts.

5. Run locally

docker compose up -d redis     # compose file in section 6
export API_KEY=dev-key-123
export REDIS_HOST=localhost
uvicorn main:app --port 8000        # terminal 1
rq worker scrape                    # terminal 2
curl -X POST http://localhost:8000/scrape \
  -H 'content-type: application/json' -H 'x-api-key: dev-key-123' \
  -d '{"url":"https://example.com","selectors":{"title":"h1","price":".price"}}'
# {"job_id":"xyz","status":"/jobs/xyz"}
curl -H 'x-api-key: dev-key-123' http://localhost:8000/jobs/xyz
# {"status":"finished","result":{...}}

Test the guards too: a request without the API key returns 401, and a request for http://127.0.0.1:80 returns 400 before any browser launch.

6. Deploy (2026-08 verified prices)

Dockerfile:

FROM python:3.12-slim
WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt \
    && playwright install --with-deps chromium
COPY . .
RUN useradd -m scraper && chown -R scraper:scraper /app
USER scraper
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

requirements.txt:

fastapi
uvicorn[standard]
playwright
redis
rq
pydantic
tenacity
httpx

docker-compose.yml:

services:
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes
    volumes:
      - redisdata:/data
    restart: unless-stopped

  api:
    build: .
    environment:
      REDIS_HOST: redis
      API_KEY: ${API_KEY}
    ports:
      - "8000:8000"
    depends_on:
      - redis
    restart: unless-stopped

  worker:
    build: .
    command: rq worker scrape --url redis://redis:6379
    environment:
      REDIS_HOST: redis
    depends_on:
      - redis
    restart: unless-stopped

  caddy:
    image: caddy:2
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./Caddyfile:/etc/caddy/Caddyfile:ro
      - caddy_data:/data
    restart: unless-stopped

volumes:
  redisdata:
  caddy_data:

Caddyfile:

YOUR-DOMAIN.com {
    reverse_proxy api:8000
}

On the VPS (4 GB, Playwright needs RAM):

git clone <your-repo> scraper-api && cd scraper-api
cp .env.example .env      # API_KEY=..., REDIS_HOST=redis
docker compose up -d --build
Service Monthly cost
VPS 4 GB (Hetzner CX32) $8.49
Redis (on VPS) $0
Total ≈ $8.50/month

Environment variables

Variable Required Description
API_KEY yes Shared secret required in the x-api-key header on every endpoint.
REDIS_HOST yes Redis host for the queue (localhost locally, redis in compose).
DOMAIN_DELAY no Minimum seconds between jobs for the same domain (default 5).

Production checklist

  • Per-domain rate limiting + politeness delay (scheduler)
  • Blocked responses detected (CAPTCHA/redirect) → clear error, not silent failure
  • Robots.txt check before scrape
  • Job timeout + stale job reaping (rq worker --max-jobs / TTL on results)
  • Logs with structured fields (domain, status, duration)
  • API key rotated on any leak; secret comparison uses secrets.compare_digest
  • Worker restarted daily (Playwright memory) — restart: unless-stopped + cron

Common problems

  • Blocks: capacity planning is politeness. Slow down, rotate user-agents, cache results by URL+date.
  • The hot-news page that changes: selectors break; the empty-value check alerts on extraction gaps instead of returning garbage.
  • Memory in workers: Playwright browsers leak — restart the worker daily (launch-per-job as in worker.py keeps it bounded, at slower throughput).
  • SSRF: private targets must be rejected before the browser starts — the guard in main.py covers loopback/private ranges; extend it to any private range your VPS uses.

Improvements

  • Screenshot capture per page (debugging gold)
  • Pagination iteration (next link following)
  • Webhook delivery of results

Conclusion

A scraping API is a FastAPI shell around a politely-operated Playwright browser. Structured selectors, exponential retries and per-domain pacing turn it into a reliable, billable product for ~$8.50/month.

Related guides

Build an AI-Powered PDF Chat App

Build a PDF chat app with Next.js, PostgreSQL + pgvector and an LLM API: upload, chunk, embed, retrieve and chat with any document.

Build an Automated Invoice Generator

Build an automated invoice generator with Next.js, PostgreSQL and a PDF renderer: templates, Stripe sync and email delivery on a VPS.

Build a SaaS with Next.js, Postgres and Stripe

Build a complete subscription SaaS from scratch with Next.js, PostgreSQL and Stripe: auth, billing, onboarding and deployment to a VPS.