# scrapingpros Python SDK — Complete Reference

> Python SDK for the [Scraping Pros](https://scrapingpros.com) API. Web scraping with browser rendering, proxy rotation, and structured data extraction.

## Installation

**Requires Python 3.10 or higher.**

```bash
pip install scrapingpros
```

## Quick Start

```python
from scrapingpros import SyncClient

client = SyncClient("demo_6x595maoA6GdOdVb")
result = client.scrape("https://example.com")
print(result.content)
```

No signup needed. Demo token: 5,000 credits/month, 30 req/min.

## Credit System

- 1 simple request = 1 credit
- 1 browser request = 5 credits
- Anti-bot protection and proxy rotation included at no extra cost
- Credits are NOT consumed for requests that fail due to infrastructure errors
- `retry_on_block=True` only charges for the successful attempt
- Track credits: `client.credits_charged`, `client.quota_remaining`
- View plans: `client.plans()`, check usage: `client.billing()`

## Plans

| Plan | Price/mo | Credits/mo | Rate Limit |
|------|----------|------------|------------|
| Free | $0 | 1,000 | 30/min |
| Starter | $29 | 25,000 | 30/min |
| Growth | $69 | 100,000 | 60/min |
| Pro | $199 | 500,000 | 120/min |
| Scale | $499 | 2,500,000 | 200/min |
| Enterprise | Custom | Unlimited | 2,000/min |

See https://scrapingpros.com for details.

## Client Initialization

```python
from scrapingpros import SyncClient, AsyncClient

# Sync client
client = SyncClient(
    "your_token",          # or omit and set SP_TOKEN env var
    base_url="https://api.scrapingpros.com",  # default
    timeout=120.0,         # request timeout in seconds
    max_retries=3,         # auto-retry on 429 rate limits
)

# Async client (same API surface)
async with AsyncClient("your_token") as client:
    result = await client.scrape("https://example.com")
```

## Methods Reference

### scrape(url, ...) -> ScrapeResponse

Scrape a single URL. Returns HTML or clean markdown.

```python
result = client.scrape(
    "https://example.com",
    format="markdown",        # "html" (default) or "markdown"
    browser=False,            # True for JS-heavy / protected sites
    use_proxy="any",          # "any" (default), None, country code "US", or ProxyConfig
    screenshot=False,         # True for full-page PNG (requires browser=True)
    actions=None,             # browser automation steps
    language="",              # "en-US", "es-AR", etc.
    http_method=None,         # MethodPOST(payload={...}) for POST requests
    headers=None,             # {"User-Agent": "..."}
    window_size="",           # "desktop", "mobile", "any"
    cookies=None,             # {"name": "value"}
    extract=None,             # CSS/XPath selectors (see below)
    network_capture=None,     # NetworkCaptureConfig(resource_types=["xhr"])
    retry_on_block=False,     # True: auto-retry on CAPTCHA/403, only charges on success
    custom_id=None,           # opaque ID echoed back — for traceability
)
```

The API internally picks the best engine for each site; you don't need to specify which browser to use.

**ScrapeResponse fields:**
- `result.content` — convenience: returns markdown if available, else html
- `result.html` — raw HTML (when format="html")
- `result.markdown` — clean text (when format="markdown")
- `result.statusCode` — HTTP status from target page
- `result.extracted_data` — dict with extracted data
- `result.screenshot` — base64 PNG string
- `result.executionTime` — seconds
- `result.potentiallyBlockedByCaptcha` — bool (legacy, **soft-deprecated v0.5.0**; prefer `result.guidance.success` for the canonical verdict)
- `result.timings` — performance breakdown (always present, never null)
- `result.evaluate_results` — list of JS evaluation results
- `result.network_requests` — captured network activity
- `result.guidance` — `ScrapeGuidance` with error analysis and next steps (see below)

**Binary content (v0.6.0+)** — when the response is a PDF / image / ZIP / etc:
- `result.is_binary` — `bool`, `True` for non-text responses
- `result.content_type` — `str | None`, MIME (`"application/pdf"`, `"image/png"`, …). Not to be confused with `MethodPOST.content_type` (request-side, `"json"`/`"form"`)
- `result.body_base64` — `str | None`, base64 of the raw body; prefer the `body` helper to decode
- `result.body_url` — `str | None`, reserved for future blob offload (>5 MB)
- `result.body` — `bytes`, always-bytes accessor (decodes base64 for binary, UTF-8 encodes text). Raises `ValueError` on offloaded bodies (`body_url`) → use `download_body()`
- `result.text` — `str | None`, returns `None` for binary so a careless `.find()` fails loudly
- `result.save(path)` — write to disk, returns bytes written
- `result.download_body()` (async) / `result.download_body_sync()` — auto-detects inline vs offloaded
- `result.content` returns `None` for binary (use `body` / `save` instead)

Download pattern:
```python
resp = client.scrape("https://x.com/charter.pdf", browser=False)
if resp.is_binary:
    resp.save("charter.pdf")
```

### Response Guidance (ScrapeGuidance)

Every scrape response includes a `guidance` object that explains the result and suggests what to do next.

```python
result = client.scrape("https://hard-site.com")

# Always present — check it after every scrape
g = result.guidance
print(g.success)          # True if content was retrieved
print(g.credits_charged)  # 1 (simple) or 5 (browser), 0 if refunded
```

When a scrape fails, guidance tells you why and what to try:

```python
if not result.guidance.success:
    print(result.guidance.error_type)         # "captcha", "ssl_error", "timeout", "login_wall", etc.
    print(result.guidance.error_provider)      # "cloudflare", "datadome", etc.
    print(result.guidance.next_steps)          # ["Try with browser=true", "Try retry_on_block=true"]
    print(result.guidance.suggested_request)   # ready-to-use params dict: {"url": "...", "browser": True}
    print(result.guidance.credits_refunded)    # True if credits were returned

    # If stop_reason is set, do NOT retry — it won't help
    if result.guidance.stop_reason:
        print("Stop:", result.guidance.stop_reason)  # "Site requires authentication"
    elif result.guidance.suggested_request:
        # Retry with suggested params
        result = client.scrape(**result.guidance.suggested_request)
```

**ScrapeGuidance fields:**
- `success` — `True` if usable content was retrieved
- `error_type` — category: `"captcha"`, `"ssl_error"`, `"timeout"`, `"login_wall"`, `"blocked"`, etc.
- `error_provider` — protection provider: `"cloudflare"`, `"datadome"`, `"recaptcha"`, etc.
- `credits_charged` — credits consumed (0 if refunded)
- `credits_refunded` — `True` if credits were returned due to infrastructure error
- `next_steps` — list of human-readable suggestions
- `suggested_request` — dict of params to retry with (pass directly to `client.scrape(**suggested_request)`)
- `stop_reason` — if set, retrying is futile (e.g. login wall, geo-blocked)

### retry_on_block — Server-Side Anti-Bot Retry

```python
# For protected sites: auto-retry with different IP/fingerprint
result = client.scrape(
    "https://hard-site.com",
    browser=True,
    retry_on_block=True,  # up to 3 retries, only charges on success
)
# Combined with early CAPTCHA detection: blocked pages return in ~5-10s
# instead of waiting for full page load (60-85s)
```

### Browser Rendering

Enable `browser=True` for JavaScript-heavy sites and protected sites. The API picks the best engine for each target domain automatically — you don't configure which engine to use.

```python
# JS-heavy site
result = client.scrape("https://spa-app.com", browser=True)

# Protected site — combine with retry_on_block + resource blocking for hard targets
result = client.scrape(
    "https://www.google.com/shopping/...",
    browser=True,
    retry_on_block=True,
    block_resources=["image", "font", "media"],
)
```

### Resource Blocking (block_resources, block_requests)

Block unnecessary resources to speed up page load and reduce detection surface. Requires `browser=True`.

**`block_resources`** — block by resource type:
```python
result = client.scrape(
    "https://heavy-site.com",
    browser=True,
    block_resources=["image", "font", "media", "stylesheet"],
)
# Valid types: "image", "stylesheet", "font", "media", "script", "document", "xhr", "fetch"
```

**`block_requests`** — block by URL substring (case-insensitive):
```python
result = client.scrape(
    "https://www.google.com/shopping/...",
    browser=True,
    block_requests=["google-analytics", "doubleclick", "googletagmanager", "googlesyndication"],
)
```

**Combined — optimal for hard sites:**
```python
result = client.scrape(
    "https://www.google.com/shopping/...",
    browser=True,
    block_resources=["image", "font", "media"],
    block_requests=["google-analytics", "doubleclick", "googlesyndication"],
    retry_on_block=True,
)
```

### Data Extraction

```python
result = client.scrape(
    "https://quotes.toscrape.com/",
    extract={
        "title": "css:h1",
        "quotes": {"selector": "css:.text", "multiple": True},
        "links": {"selector": "css:a", "attribute": "href", "multiple": True},
        "xpath_example": "xpath://div[@class='quote']",
    },
)
for quote in result.extracted_data["quotes"]:
    print(quote)
```

### Browser Automation (Actions)

```python
from scrapingpros import ClickAction, InputAction, WaitForSelectorAction, EvaluateAction

result = client.scrape(
    "https://example.com/login",
    browser=True,
    actions=[
        InputAction(selector="css:#email", text="user@example.com"),
        InputAction(selector="css:#password", text="secret"),
        ClickAction(selector="css:button[type=submit]", wait_for_navigation=True),
        WaitForSelectorAction(selector="css:.dashboard", time=5000),
    ],
)
```

Available actions: `ClickAction`, `InputAction`, `SelectAction`, `KeyPressAction`, `WaitForSelectorAction`, `WaitForTimeoutAction`, `EvaluateAction`, `CollectAction`, `WhileControl`.

### JavaScript Execution (EvaluateAction)

Run arbitrary JavaScript in the browser page context. Use this to call internal APIs, extract data from JS variables, intercept XHR/fetch responses, or access data that isn't in the DOM.

```python
from scrapingpros import EvaluateAction, WaitForSelectorAction

# Example: call an internal API endpoint from the page context
result = client.scrape(
    "https://example.com/products",
    browser=True,
    actions=[
        WaitForSelectorAction(selector="css:.product-list", time=5000),
        EvaluateAction(script="fetch('/api/products').then(r => r.json())"),
    ],
)
# Results are in evaluate_results, one per EvaluateAction
api_data = result.evaluate_results[0]  # parsed JSON from the fetch

# Example: extract data from JavaScript variables
result = client.scrape(
    "https://example.com/dashboard",
    browser=True,
    actions=[
        EvaluateAction(script="window.__NEXT_DATA__"),        # Next.js page data
        EvaluateAction(script="document.title"),               # simple DOM access
        EvaluateAction(script="""
            // Multi-line scripts work too
            const items = document.querySelectorAll('.item');
            Array.from(items).map(el => ({
                name: el.textContent,
                href: el.querySelector('a')?.href
            }))
        """),
    ],
)
next_data = result.evaluate_results[0]
title = result.evaluate_results[1]
items = result.evaluate_results[2]
```

**EvaluateAction fields:**
- `script` — JavaScript code to execute in the page context. The return value is captured.
- `timeout` — max execution time in ms (default: 30000)

**When to use evaluate vs extract:**
- Use `extract` (CSS/XPath selectors) when the data is visible in the HTML DOM
- Use `EvaluateAction` when you need to: call internal APIs (`fetch`), read JS variables (`window.__DATA__`), execute complex DOM logic, or access data not rendered in HTML

### download(url, ...) -> DownloadResponse

Download a file as base64.

```python
result = client.download("https://example.com/report.pdf")
print(result.contentType)  # "application/pdf"
import base64
with open("report.pdf", "wb") as f:
    f.write(base64.b64decode(result.content))
```

### Batch API — `submit_batch()` + streaming results (v0.4.0+)

**For batches over ~100 URLs, use `submit_batch()` — it's the high-level async API with streaming results, progress tracking, and failure visibility.** Scales up to 50,000+ URLs in a single call.

```python
batch = client.submit_batch("daily-reviews", [
    {"url": tour["url"], "custom_id": tour["id"], "browser": True}
    for tour in my_tours
])

# Stream results as jobs complete, not when everything is done
for result in batch.iter_results():
    if result.guidance and result.guidance.success:
        save_to_db(my_id=result.custom_id, content=result.content)

    # Progress available any time during iteration
    if batch.completed_count % 100 == 0:
        print(f"{batch.pct:.1f}% ({batch.success_count} OK / {batch.failed_count} failed) "
              f"— ETA {batch.eta_seconds:.0f}s" if batch.eta_seconds else "")
```

**Batch properties (read-only, always current):**
- `batch.total`, `batch.completed_count`, `batch.success_count`, `batch.failed_count`, `batch.processing_count`
- `batch.pct` (0..100), `batch.elapsed_seconds`, `batch.jobs_per_second`, `batch.eta_seconds`
- `batch.status` (`"in_progress"`, `"completed"`, `"failed"`, `"cancelled"`), `batch.is_finished`

**Callbacks (chainable):**
```python
batch = (
    client.submit_batch("daily", items)
    .on_result(lambda r: save(r) if r.guidance.success else log_error(r))
    .on_progress(lambda b: print(f"{b.pct:.1f}% — ETA {b.eta_seconds}s"))
    .on_complete(lambda b: notify_slack(f"Done: {b.success_count}/{b.total}"))
)
batch.run_until_complete(timeout=7200)  # blocks, fires callbacks
```

**Tuning for very large batches:**
```python
for result in batch.iter_results(
    poll_interval=10.0,  # override the adaptive default
    timeout=7200,        # raise TimeoutError after N seconds
    fetch_workers=10,    # parallel result fetches per tick (default 5)
    include_failed=True, # yield failed/timeout jobs too (default True)
):
    ...
```

`poll_interval` is **adaptive by default** since v0.7.3 — sized to the batch so large runs don't burn the rate budget polling. Tiers for `iter_results`: `<100 items → 5s, <500 → 10s, <2000 → 15s, ≥2000 → 30s`. `run_and_wait` uses a lighter table (`<100 → 2s, <500 → 5s, <2000 → 10s, ≥2000 → 20s`) since its tick is one cheap GET. Pass `poll_interval=N` to override; the public helper `scrapingpros.adaptive_poll_interval(items, kind="jobs"|"status")` exposes the lookup.

**Resume after client crash:**
```python
# Persist batch.collection_id + batch.run_id on submit
# Later reattach with:
batch = client.get_batch(collection_id, run_id)   # auto-refreshes counters since v0.5.2
print(f"{batch.pct:.0%} ({batch.success_count}/{batch.total})")  # snapshot without iterating
for result in batch.iter_results():
    if result.custom_id not in already_processed_ids:
        save(result)
```

Or use the `client.iter_results(cid, rid)` shortcut (v0.5.2+) when you have IDs but don't need a `Batch` handle:
```python
for r in client.iter_results(saved_cid, saved_rid):
    save(r.custom_id, r.content)
```

**Cross-process resume with a cursor (v0.7.5+):** persist `batch.last_completed_at` alongside the IDs and pass it as `since=` on reattach — the SDK skips jobs with `completed_at <= since` via the server-side filter, no client-side dedup needed:
```python
# Submit + persist as you go
for r in batch.iter_results():
    save(r)
    db.update(cid=batch.collection_id, rid=batch.run_id,
              cursor=batch.last_completed_at, submitted=batch.submitted_count)

# Different process / after a crash
row = db.find(cid)
for r in client.iter_results(row.cid, row.rid,
                              since=row.cursor,
                              submitted_count=row.submitted):
    save(r)
```
Accepts `datetime` or ISO 8601 string. Low-level parity: `iter_run_jobs(..., since_completed_at=)` and `get_run_jobs(..., since_completed_at=)`.

**Embed-body polling (v0.7.7+):** `Batch.iter_results()` drains the result body inline from the listing endpoint and skips the per-job `GET /jobs/{id}/result` round-trip on the happy path. Transparent — no caller change. The SDK falls back to `/result` per job when the inline body is absent (oversize >256 KB, blob miss, legacy server). Low-level callers can opt in via `iter_run_jobs(..., include_results=True)` / `get_run_jobs(..., include_results=True)`; read `job.result` and fall back to `client.get_job_result(...)` only when it's `None`. `status_filter` on those low-level methods also accepts a list / CSV string for draining multiple terminal states in one paginated stream.

**Crash-resilient submit (v0.5.3+):**

Since v0.5.3 `submit_batch` automatically generates an `Idempotency-Key` UUID per call. A network-level retry of the same submit (timeout, transient 5xx) returns the **same** `collection_id` without creating a duplicate run — the server dedupes within 24 h.

```python
from scrapingpros import SyncClient, SubmitTimeout

try:
    batch = client.submit_batch(name, items, submit_timeout=30.0)
except SubmitTimeout:
    # Safe to retry — server replays the original response.
    batch = client.submit_batch(name, items)
```

For belt-and-braces verification, `find_recent_batch` looks up the orphan and reattaches via server-side filters + `list_runs(cid, status_filter="in_progress")`:

```python
batch = client.find_recent_batch(name=name) or client.submit_batch(name, items)
```

Pass `idempotency_key="..."` to control the key yourself (e.g. derive from a DB row id).

**Per-item validation buckets on every submit (v0.7.2+):** when the server rejects items but creates the collection with the rest, the three rejection buckets are surfaced as attributes on the returned `Batch`:

- `batch.blocked_urls` — `list[BlockedURL]` (SSRF: `private_ip`, `invalid_protocol`, `dns_failed`, `blocked_hostname`, `invalid_port`, `malformed_url`, `blocked`).
- `batch.invalid_items` — `list[InvalidItem]` (Pydantic body-validation / parameter rules, e.g. `custom_id` > 255 chars, `screenshot` without `browser`). Each item carries `index`, `url`, and a list of `errors` with `field`, `error_type`, `message`.
- `batch.duplicate_urls` — `list[str]` (URLs skipped as duplicates of an earlier entry in the same submit). The count is also on `batch.duplicates_skipped`.

`submit_batch` emits a `RuntimeWarning` when any of those are non-empty so silent data loss surfaces during testing. `batch.total` already subtracts every rejection bucket.

```python
batch = client.submit_batch(name, items)   # RuntimeWarning if any rejects
for it in batch.invalid_items:
    print(f"  - [{it.index}] {it.url} {it.errors[0].field}={it.errors[0].error_type}")
```

**End-of-run report (`batch.summary()`, v0.7.2+):** single call to get a complete `BatchSummary` snapshot covering both layers — what the server ran (`queued / succeeded / failed`) and what it rejected at submit (`blocked / invalid / duplicates`). `print(batch.summary())` is a multi-line ASCII block. Invariants: `submitted == queued + blocked + invalid + duplicates`; once `is_finished`, `succeeded + failed == queued`. Canonical pattern fires it from `on_complete`:

```python
batch = client.submit_batch("daily", items).on_complete(lambda b: print(b.summary()))
for r in batch.iter_results():
    handle(r)
```

Reattached batches (`client.get_batch(cid, rid)`) carry `summary().submitted = None` — the SDK cannot recover the original payload size from the server. Workaround: persist `len(payload)` alongside `(cid, rid)` in your own storage.

**Lenient submit (`submit_batch_lenient`, v0.5.3+):** does **not** emit the warning. Returns `(batch, blocked)` with the SSRF rejections in the tuple; the other two buckets are still on `batch.invalid_items` and `batch.duplicate_urls`. Use this when partial-success is the expected case (CSV with flaky URLs, user-supplied input):

```python
batch, blocked = client.submit_batch_lenient(name, items)
print(f"submitted {batch.total}, blocked {len(blocked)}, "
      f"invalid {len(batch.invalid_items)}, dup {batch.duplicates_skipped}")
```

**Listing runs of a collection (`list_runs`, v0.5.3+):**

```python
resp = client.list_runs(cid, status_filter="in_progress")   # or "completed"
for run in resp.items:
    print(run.run_id, run.status, run.created_at)
```

Returns a `RunListPublic` ordered by `created_at desc`. Use `status_filter="in_progress"` to find a live run after a client restart — that's what `find_recent_batch` uses internally.

**Typed 404s on `get_job_result` (v0.5.3+):**

```python
from scrapingpros import (
    JobResultError, JobResultPending, JobResultExpired,
    JobResultLost, JobNotFound,
)

try:
    result = client.get_job_result(cid, rid, jid)
except JobResultPending:   # job hasn't reached a terminal state — retry later
    pass
except JobResultExpired:   # > 24 h since completion, blob TTL'd — re-run
    pass
except JobResultLost:      # service incident lost the result — escalate
    pass
except JobNotFound:        # job_id doesn't exist — programming bug
    pass
```

All four inherit from `JobResultError` (which inherits from `APIError`) so existing `except APIError` clauses keep catching them.

**Async version (same API, every method is `async`):**
```python
async with AsyncClient("token") as client:
    batch = await client.submit_batch("daily", items)
    async for result in batch.iter_results():
        await save_async(result)
```

**Choosing between methods**:

| | `scrape()` | `batch_scrape()` | `submit_batch()` |
|---|---|---|---|
| URLs | 1 | Many (Collections-backed) | Many (Collections-backed) |
| Parallelism | None | Server-side | Server-side |
| Returns | `ScrapeResponse` | `list[ScrapeResponse]` | `Batch` (streaming iterator) |
| Progress | n/a | n/a | `pct`, `eta`, success/fail counters |
| Memory | n/a | Holds all results | Constant (streams) |

> ⚠️ **`scrape_many()` was removed in v0.7.0.** Calling it raises `RuntimeError` with the migration recipe inline. Replace with `batch_scrape(urls, ...)` (drop-in, same signature) or `submit_batch(name, urls).iter_results()` (streaming with progress). At N=1000 with `browser=True`, `batch_scrape` is **5× faster** than `scrape_many` was (185 s vs 930 s) and more reliable (998 / 1000 vs 990 / 1000).

### batch_scrape(items, ...) -> list[ScrapeResponse]

Convenience wrapper around `submit_batch()` that blocks until every job is done and returns the full result list. **Drop-in replacement for the removed `scrape_many()`** with server-side scaling.

```python
results = client.batch_scrape([
    {"url": u, "custom_id": product_id, "browser": True}
    for product_id, u in catalog.items()
])
for r in results:
    if r.guidance.success:
        save(r.custom_id, r.content)
```

For very large batches (thousands+) prefer `submit_batch()` directly so you can stream results live.

### Migrating from `scrape_many()`

The migration is mechanical:

```python
# Before (v0.6.x and earlier) — REMOVED in v0.7
results = client.scrape_many(urls, format="markdown", browser=True)

# After (v0.7+)
results = client.batch_scrape(urls, format="markdown", browser=True)

# With actions / blocking — same kwargs as scrape_many had
results = client.batch_scrape(urls,
    browser=True,
    actions=[WaitForSelectorAction(selector="css:.item", time=5000),
             EvaluateAction(script="document.querySelectorAll('.item').length")],
    block_resources=["image", "font", "media"],
    block_requests=["google-analytics", "doubleclick"],
    headers={"Accept-Language": "en-US"},
)

# Heterogeneous: different params per URL — same syntax as before
from scrapingpros import ScrapeRequest
results = client.batch_scrape([
    ScrapeRequest(url="https://site1.com", browser=True, actions=[...]),
    ScrapeRequest(url="https://site2.com", format="markdown"),
    {"url": "https://site3.com", "browser": True, "extract": {"title": "css:h1"}},
])
```

### Batch Processing with Webhooks (collections)

```python
# Create collection with webhook
col = client.create_collection("my-batch", [
    {"url": "https://example.com/1"},
    {"url": "https://example.com/2", "browser": True},
], callback_url="https://your-server.com/webhook")

# Option A: Poll for completion
run = client.run_and_wait(col.id, timeout=120)

# Option B: Receive webhook POST when done
# Your server gets: {"event": "run.completed", "run_id": "...", "job_ids": [...]}
# Signed with HMAC-SHA256 in X-SP-Signature header

# Fetch individual results (HTML available 48h, metadata 90 days)
for job_id in run.job_ids:
    result = client.get_job_result(col.id, run.run_id, job_id)
    print(result.url, result.custom_id, result.content[:100])

# Check webhook delivery status
run = client.get_run(col.id, run.run_id)
print(run.callback_status)  # "sent", "pending", "failed", "retrying"

# Cleanup
client.delete_collection(col.id)
```

### URL traceability with custom_id

Each request can carry a `custom_id` echoed back in the job listing and result, so you can map results to your data without depending on order:

```python
col = client.create_collection("my-batch", [
    {"url": "https://example.com/tour/1", "custom_id": "tour_1"},
    {"url": "https://example.com/tour/2", "custom_id": "tour_2"},
])
run = client.run_and_wait(col.id)

# Iterate jobs with full traceability — each job has url + custom_id
for job in client.iter_run_jobs(col.id, run.run_id):
    if job.status == "completed":
        result = client.get_job_result(col.id, run.run_id, job.job_public_id)
        save_to_db(result, my_id=job.custom_id)
```

`batch_scrape()` results carry `url` and `custom_id` too — same pattern.

### Pagination over jobs

`get_run_jobs()` paginates internally by default (returns all jobs). For very large runs (1000+) prefer `iter_run_jobs()` to stream lazily:

```python
# Default: fetch all jobs in memory
jobs = client.get_run_jobs(col.id, run.run_id)
print(f"{len(jobs.items)} jobs")

# Memory-efficient: stream lazily, optionally filter by status
for job in client.iter_run_jobs(col.id, run.run_id, status_filter="completed"):
    process(job)

# Manual pagination: pass cursor to control fetching
page = client.get_run_jobs(col.id, run.run_id, cursor=None, limit=500)
while page.has_more:
    page = client.get_run_jobs(col.id, run.run_id, cursor=page.cursor_next, limit=500)
```

`JobExecutionPublic` fields: `job_public_id`, `run_public_id`, `collection_id`, `client_id`, `url`, `custom_id`, `status`, `status_code`, `is_success`, `has_extractable_data`, `validator_version`, `queued_at`, `started_at`, `completed_at`, `execution_time_ms`, `retries_attempted`, `block_reason`, `url_truncated`, `protection_stack`, `rule_hits`.

`has_extractable_data: bool | None` (v0.5.0+) — whether the page contained structured data the server could extract (JSON-LD, microdata, OpenGraph, `__NEXT_DATA__`). Independent of `is_success` — a 200 page with usable content can still have no structured payload. `None` for jobs created before the 2026-04-29 migration.

`validator_version: str | None` (v0.5.0+) — version of the HTML Validator that produced `is_success` / `block_reason` / `protection_stack` / `rule_hits` on this job. Pin in tests to detect silent classifier upgrades.

`client_id: str | None` (v0.5.0+) — the client account that owns the job.

### Success classification (`is_success` + `success_criterion`)

Each job carries `is_success: bool | None` — the **server's authoritative verdict** of whether the scrape produced usable content. Use it instead of rolling your own `status_code` check — the server catches soft-blocks (Google CAPTCHA page with 200 + large body, Amazon "Robot Check", etc.) that a naive heuristic misses.

```python
for job in client.iter_run_jobs(col.id, run.run_id):
    if job.is_success:
        save_ok(job)
    elif job.is_success is False:
        log_fail(job.url, reason=job.block_reason or f"http_{job.status_code}")
    else:  # None — legacy run (pre-2026-04-24)
        pass  # SDK's Batch class falls back to client heuristic
```

`batch.iter_results()` already honors this internally — `result.guidance.success` reflects `job.is_success`.

Each run also exposes the active policy:

```python
run = client.get_run(col.id, run.run_id)
run.success_criterion.version  # e.g. "content_success_v1"
run.success_criterion.rules    # human-readable rule list

# Pin it in integration tests to catch silent policy changes:
assert run.success_criterion.version == "content_success_v1"
```

Policy `content_success_v1` (current): `status == "completed"` AND `2xx status_code` AND `potentiallyBlockedByCaptcha is false` AND `block_reason null or "none"`. Invariant: `sum(is_success == True) == run.success_requests`.

### POST to a different URL than the navigation target (MethodPOST.url)

Some sites require navigating to one page (to set cookies, generate session) and POSTing to a different endpoint (an internal API or GraphQL):

```python
from scrapingpros import MethodPOST

result = client.scrape(
    "https://www.example.com/dashboard",     # navigation
    http_method=MethodPOST(
        url="https://api.example.com/graphql", # POST goes here
        payload={"query": "..."},
    ),
)
```

If `MethodPOST.url` is omitted, the POST goes to the same URL as the scrape (legacy behavior).

### POST with form-encoded body (MethodPOST.content_type, v0.5.0+)

OAuth2 `grant_type=client_credentials` and most legacy form-based APIs require `application/x-www-form-urlencoded` request bodies, not JSON. Set `content_type="form"`:

```python
from scrapingpros import MethodPOST

resp = client.scrape(
    "https://api.example.com/v1/oauth2/token",
    http_method=MethodPOST(
        payload={"grant_type": "client_credentials", "scope": "read"},
        content_type="form",   # default is "json"
    ),
    headers={"Authorization": f"Basic {base64_creds}"},
)
```

Accepted values: `"json"` (default if omitted), `"form"`. Any other value raises `ValidationError` client-side.

### Wait for non-visible elements (WaitForSelectorAction.state, v0.5.0+)

By default the server waits for the selector to be **visible** (rendered, non-zero size). For hidden DOM nodes — typically `<script>` tags carrying embedded JSON like `__NEXT_DATA__` — pass `state="attached"` so the wait completes as soon as the element exists in the DOM, regardless of visibility:

```python
from scrapingpros import WaitForSelectorAction

result = client.scrape(url, browser=True, actions=[
    WaitForSelectorAction(selector="css:script#__NEXT_DATA__", time=8000, state="attached"),
])
```

Accepted values: `"visible"` (server default), `"attached"`, `"hidden"`. Leave `state=None` for the default behavior.

### Capture response bodies (NetworkCaptureConfig.url_pattern, v0.5.0+)

Beyond capturing request metadata, the server can also include the **response body** for URLs matching a glob pattern. Useful for extracting tokens from auth flows (Firebase `identitytoolkit`, OAuth) or payloads from GraphQL/Apollo `persistedQuery` calls without re-executing them:

```python
from scrapingpros import NetworkCaptureConfig

result = client.scrape(url, browser=True, network_capture=NetworkCaptureConfig(
    resource_types=["xhr", "fetch"],
    url_pattern="*identitytoolkit.googleapis.com*",
))

for entry in result.network_requests or []:
    if "body" in entry:                # only present when url_pattern matched
        process(entry["body"])
```

When a request URL matches `url_pattern`, the entry in `result.network_requests` gains:
- `body` — response body as utf-8 string (or base64 for binary). Capped at **64 KB**.
- `body_truncated` — `True` if the response was larger than 64 KB.
- `body_error` — optional reason a body could not be captured (e.g. `"global_timeout_5s"`, `"response_disposed"`). Body fetch never hangs the scrape.

Single pattern only. For multiple endpoints, use a broader glob (`*api*`) and filter client-side.

Other collection methods:
- `client.list_collections()` -> list[CollectionPublic]
- `client.get_collection(id)` -> CollectionPublic
- `client.update_collection(id, name, requests)` -> NewCollectionResponse
- `client.create_run(collection_id)` -> RunPublic
- `client.get_run(collection_id, run_id)` -> RunPublic
- `client.get_run_jobs(collection_id, run_id, *, cursor=None, limit=1000, status_filter=None)` -> JobExecutionListPublic
- `client.iter_run_jobs(collection_id, run_id, *, page_size=500, status_filter=None)` -> Iterator[JobExecutionPublic]

### Viability Test

Analyze sites before scraping. Tests each URL with multiple scraping modes to find what works.

```python
# depth controls how many modes are tested:
#   "quick"    — simple HTTP only (1 credit/URL)
#   "standard" — simple + proxy + browser (up to 7 credits/URL)
#   "full"     — all modes including browser+proxy (up to 12 credits/URL)
test = client.create_viability_test(
    ["https://example.com", "https://hard-site.com"],
    depth="full",  # default
)

import time
while True:
    result = client.get_viability_test(test.run_id)
    if result.status == "completed":
        break
    time.sleep(2)

for r in result.results:
    print(f"{r.url}: {r.recommended_strategy}")
    print(f"  Difficulty: {r.difficulty}")  # "easy", "medium", "hard"
    print(f"  Best mode: {r.best_mode}")    # cheapest mode that worked
    print(f"  CAPTCHA: {r.captcha_detected} {r.captcha_providers}")
    print(f"  Browser needed: {r.javascript_required}")
    print(f"  Credits used: {r.total_credits_used}")

    # modes_tested shows per-mode results
    if r.modes_tested:
        for mode, result_data in r.modes_tested.items():
            status = "OK" if result_data.success else "BLOCKED"
            print(f"    {mode}: {status} ({result_data.time_ms}ms, {result_data.credits} credits)")
```

**UrlViabilityResult key fields:**
- `difficulty` — `"easy"`, `"medium"`, or `"hard"`
- `best_mode` — cheapest successful mode (e.g. `"simple"`, `"browser_proxy"`)
- `modes_tested` — dict of mode name to `ViabilityModeResult`
- `total_credits_used` — total credits consumed across all mode tests
- `recommended_strategy` — suggested scraping approach
- `suggested_params` — dict of params to use with `client.scrape()`

**ViabilityModeResult fields:**
- `success` — whether this mode retrieved content
- `blocked` — whether the site blocked this mode
- `captcha_type` — CAPTCHA provider if detected
- `time_ms` — response time in milliseconds
- `credits` — credits consumed for this mode test
- `error` — error message if failed

### Proxy Management

```python
resp = client.list_proxy_countries()
print(resp.countries)  # ["AD", "AE", "AF", ...]

resp = client.request_proxy_country("US", reason="Need US pricing")
print(resp.status)  # "pending" or "already_approved"

status = client.proxy_status()
print(status.approved_countries)

result = client.scrape("https://example.com", use_proxy="US")
```

### Plans & Billing

```python
plans = client.plans()  # public, no auth needed
billing = client.billing(month="2026-04")
metrics = client.client_metrics(date="2026-04", detail="urls")

# After any call:
print(client.credits_charged)       # 1 or 5
print(client.quota_remaining)       # credits left this month
print(client.rate_limit_remaining)  # requests left this minute
```

### Health Check

```python
health = client.health()  # {"status": "healthy", ...}
```

## Async Client

Same API, all methods are `async def`:

```python
from scrapingpros import AsyncClient

async with AsyncClient("demo_6x595maoA6GdOdVb") as client:
    result = await client.scrape("https://example.com", format="markdown")
    print(result.content)

    col = await client.create_collection("batch", [{"url": "https://example.com"}],
                                          callback_url="https://my-server.com/hook")
    run = await client.run_and_wait(col.id)
```

## Error Handling

```python
from scrapingpros import (
    ScrapingPros,
    ScrapingProsError,
    AuthenticationError,
    RateLimitError,
    QuotaExceededError,
    URLBlockedError,
    ConnectionError,
)

try:
    result = client.scrape("https://example.com")
except AuthenticationError:
    # Invalid token — get one at https://scrapingpros.com
    pass
except RateLimitError as e:
    # Rate limited — SDK auto-retries, this means all retries exhausted
    print(f"Retry after {e.retry_after}s")
except QuotaExceededError:
    # Monthly credits exhausted — upgrade at https://scrapingpros.com
    pass
except URLBlockedError:
    # URL blocked by SSRF protection — only public URLs allowed
    pass
except ConnectionError:
    # Network error — DNS, connection refused, etc.
    pass
except ScrapingProsError:
    # Catch-all for any SDK error
    pass
```

Exception hierarchy:
- `ScrapingProsError` — base for all SDK errors
  - `APIError` — HTTP error (has `.status_code`, `.detail`)
    - `AuthenticationError` — 401
    - `CountryProxyNotApproved` — 403
    - `URLBlockedError` — 422 (SSRF)
    - `ValidationError` — 422 (params)
    - `RateLimitError` — 429 (has `.retry_after`)
    - `QuotaExceededError` — 429 (quota, no retry)
  - `ConnectionError` — network error
  - `TimeoutError` — polling timeout

## LangChain Integration

```bash
pip install langchain-scrapingpros
```

```python
from langchain_scrapingpros import ScrapingProsScraper, ScrapingProsExtractor

# Scrape for RAG
scraper = ScrapingProsScraper(api_key="demo_6x595maoA6GdOdVb")
text = scraper.invoke({"url": "https://example.com"})

# Extract structured data
extractor = ScrapingProsExtractor(api_key="demo_6x595maoA6GdOdVb")
data = extractor.invoke({"url": "https://quotes.toscrape.com/", "fields": {"quotes": "css:.text"}})
```

## Key Models

All importable from `from scrapingpros import ...`:

**Request/Response:** `ScrapeRequest`, `ScrapeResponse`, `ScrapeGuidance`, `DownloadRequest`, `DownloadResponse`
**Actions:** `ClickAction`, `InputAction`, `SelectAction`, `KeyPressAction`, `WaitForSelectorAction`, `WaitForTimeoutAction`, `EvaluateAction`, `CollectAction`, `WhileControl`
**Conditions:** `SelectorVisibleCondition`, `SelectorInvisibleCondition`
**Collections:** `CollectionPublic`, `NewCollectionResponse`, `RunPublic`, `RunListPublic`, `SuccessCriterion`, `FailedJobPublic`, `JobExecutionPublic`, `JobExecutionListPublic`, `BlockedURL`, `InvalidItem`, `InvalidItemError`
**Viability:** `ViabilityTestRequest`, `ViabilityRunPublic`, `UrlViabilityResult`, `ViabilityModeResult`
**Proxy:** `ProxyConfig`, `ProxyCountriesResponse`, `ProxyRequestCountryResponse`, `ProxyStatusResponse`
**HTTP:** `MethodGET`, `MethodPOST`
**Constants:** `DEMO_TOKEN`

## Links

- Website: https://scrapingpros.com
- API Docs: https://api.scrapingpros.com/docs
- API Reference: https://api.scrapingpros.com/llms-full.txt
- PyPI: https://pypi.org/project/scrapingpros/
- LangChain: https://pypi.org/project/langchain-scrapingpros/
