From 9b3a2c60978b9d5641ad6f39ab5824eaf1625c89 Mon Sep 17 00:00:00 2001 From: Stefano Guerrini Date: Thu, 8 Oct 2026 16:51:01 +0200 Subject: [PATCH 1/2] feat(crawl): add ZenRowsCrawlClient for the Crawl API Add a `zenrows.crawl` package with a typed client for the Crawl API: create, get, list_crawls / iter_crawls, iter_results, stop, get_content, iter_download / download, and a wait helper. Responses parse into pydantic v2 models with open enums; errors raise CrawlAPIError (403 REQS008 reads "Crawl is not enabled for this account"). The shared HTTP transport now takes the error mapping and logger as arguments and gains raw and streaming request helpers. Adds unit tests, an opt-in e2e test (`make test-e2e`, deselected from `make test`), a README section, examples/11_crawl.py and DEVELOPMENT.md notes. Co-Authored-By: Claude Opus 5.5 --- DEVELOPMENT.md | 59 ++++- Makefile | 9 +- README.md | 98 ++++++- examples/11_crawl.py | 70 +++++ examples/README.md | 4 +- pyproject.toml | 6 +- src/zenrows/__init__.py | 8 +- src/zenrows/batch/_transport.py | 83 ++++-- src/zenrows/crawl/__init__.py | 51 ++++ src/zenrows/crawl/client.py | 335 +++++++++++++++++++++++ src/zenrows/crawl/errors.py | 72 +++++ src/zenrows/crawl/models.py | 193 ++++++++++++++ tests/e2e/test_crawl_e2e.py | 109 ++++++++ tests/test_crawl_client.py | 455 ++++++++++++++++++++++++++++++++ 14 files changed, 1519 insertions(+), 33 deletions(-) create mode 100644 examples/11_crawl.py create mode 100644 src/zenrows/crawl/__init__.py create mode 100644 src/zenrows/crawl/client.py create mode 100644 src/zenrows/crawl/errors.py create mode 100644 src/zenrows/crawl/models.py create mode 100644 tests/e2e/test_crawl_e2e.py create mode 100644 tests/test_crawl_client.py diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 7c4b30b..4198496 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -19,14 +19,19 @@ Makefile targets below. ``` src/zenrows/ -├── __init__.py # re-exports both clients +├── __init__.py # re-exports the clients ├── client.py # ZenRowsClient (legacy sync scraper) -└── batch/ - ├── __init__.py # ZenRowsBatchClient + key models +├── batch/ +│ ├── __init__.py # ZenRowsBatchClient + key models +│ ├── client.py # hand-written typed facade +│ ├── _transport.py # httpx wrapper, RFC 7807 → exceptions +│ ├── errors.py # BatchAPIError, ProblemDetail +│ └── models.py # GENERATED — pydantic v2 (do not edit) +└── crawl/ + ├── __init__.py # ZenRowsCrawlClient + models ├── client.py # hand-written typed facade - ├── _transport.py # httpx wrapper, RFC 7807 → exceptions - ├── errors.py # BatchAPIError, ProblemDetail - └── models.py # GENERATED — pydantic v2 (do not edit) + ├── errors.py # CrawlAPIError + └── models.py # hand-written pydantic v2 models ``` The Batch SDK is split deliberately: @@ -42,12 +47,21 @@ This way the wire types stay in lockstep with the OpenAPI document while the ergonomic surface (method names, helpers, retries, URL override) stays in our control. +The Crawl client uses the shared HTTP transport (`_transport.py`: key +header, retries, problem+json) with `CrawlAPIError.from_response` as +its error mapping, and the shared `poll_until` loop. Its models are +hand-written: the surface is a few small schemas, so there is no Crawl +codegen step. +Keep them tolerant: unknown fields are ignored, and every response enum +gets `_missing_ = classmethod(open_enum_missing)`. + ## Common tasks | Make target | What it does | |-----------------|--------------| | `make sync` | `uv sync --all-extras` | -| `make test` | `uv run pytest` | +| `make test` | `uv run pytest` (offline; e2e tests deselected) | +| `make test-e2e` | End-to-end tests against a live API (see below) | | `make check` | `ty check` + `ruff check` + `ruff format --check` (CI mode) | | `make typecheck`| `ty check src` (static types; `models.py` excluded) | | `make lint` | `ruff check --fix` | @@ -56,6 +70,37 @@ override) stays in our control. | `make build` | Build wheel + sdist via hatchling | | `make clean` | Drop caches + build outputs | +## Running the Crawl e2e test + +`tests/e2e/test_crawl_e2e.py` drives a real crawl through the client: +create (with `include_patterns`, `output_format="html"`), wait, read the +results, one page and the NDJSON download, list, stop, and a 404. It +costs a few credits. It is marked `e2e`, which `make test` deselects, +and it skips unless the three required variables below are set. + +| Variable | | +|---|---| +| `ZENROWS_API_KEY` | required: a key with Crawl access | +| `ZENROWS_CRAWL_BASE_URL` | required: the API base, e.g. `https://api.zenrows.com/v1` | +| `ZENROWS_E2E_CRAWL_URL` | required: the page the crawl starts from | +| `ZENROWS_E2E_CRAWL_INCLUDE` | optional: an include pattern; every result must contain it | + +```bash +export ZENROWS_API_KEY=zr_... # a key with Crawl access +export ZENROWS_CRAWL_BASE_URL=https://api.zenrows.com/v1 +export ZENROWS_E2E_CRAWL_URL=https://example.com/products/ +export ZENROWS_E2E_CRAWL_INCLUDE=/product/ # optional +make test-e2e +``` + +Pick a start page whose links include a few pages matching the pattern: +the test crawls at depth 1 with `max_items=3` and `output_format="html"`. + +To test against a local or staging deployment, point +`ZENROWS_CRAWL_BASE_URL` at its `/v1` base instead. When the account +has too many crawls running (429 `too_many_crawls`), the test waits and +retries (up to 5 minutes) before creating its crawl. + ## Refreshing the OpenAPI spec `docs/openapi.yaml` is the SDK-local copy of the spec. `make generate` diff --git a/Makefile b/Makefile index 03233d9..ce7c930 100644 --- a/Makefile +++ b/Makefile @@ -1,13 +1,18 @@ -.PHONY: install sync test lint format typecheck check generate docs clean build bump release +.PHONY: install sync test test-e2e lint format typecheck check generate docs clean build bump release # Bootstrap: install + dev deps, build the local venv. install sync: uv sync --all-extras -# Run the suite. +# Run the suite (offline; e2e tests are deselected). test: uv run pytest +# Run the end-to-end tests against a live API. Needs ZENROWS_API_KEY, +# ZENROWS_CRAWL_BASE_URL and ZENROWS_E2E_CRAWL_URL; see DEVELOPMENT.md. +test-e2e: + uv run pytest -m e2e -s tests/e2e + # Static type check (ty — Astral). Shipped surface only; tests are # covered by the suite. Generated models.py is excluded in pyproject. typecheck: diff --git a/README.md b/README.md index bf15f52..8133127 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ SDK to access [Zenrows](https://www.zenrows.com/) APIs directly from Python. Zenrows handles proxies rotation, headless browsers, and CAPTCHAs for you. -This package ships two clients: +This package ships three clients: - **`ZenRowsClient`** — the original synchronous scraping client. One URL in, one HTML/JSON response out. Best for ad-hoc scraping. @@ -18,6 +18,9 @@ This package ships two clients: thousands of URLs as one job, poll for results, optionally upload a CSV of URLs in one call. Best for offline / bulk pipelines. _(Private beta — [contact support](mailto:support@zenrows.com) for access.)_ + - **`ZenRowsCrawlClient`** — the Crawl API client. Give it one start + URL; it follows links and returns the URLs (and, optionally, the + pages) it finds behind it. ## Table of Contents @@ -37,6 +40,10 @@ This package ships two clients: - [Act on an id without a GET](#act-on-an-id-without-a-get) - [Scheduled jobs & webhooks](#scheduled-jobs--webhooks) - [Error handling](#error-handling) +- [Quickstart — Crawl API (`ZenRowsCrawlClient`)](#quickstart--crawl-api-zenrowscrawlclient) + - [Read the pages](#read-the-pages) + - [List and stop crawls](#list-and-stop-crawls) + - [Crawl errors](#crawl-errors) - [Contributing](#contributing) - [License](#license) @@ -485,6 +492,95 @@ The full Batch surface (jobs, runs, tasks, results, content, history, file_inputs, HMAC keys) is reachable via methods on `ZenRowsBatchClient`. See `src/zenrows/batch/client.py` or `help(ZenRowsBatchClient)`. +## Quickstart — Crawl API (`ZenRowsCrawlClient`) + +A crawl starts from one URL, follows the links on each page up to +`depth` hops, and keeps the URLs that match your patterns, up to +`max_items`. It runs in the background: `create` returns at once with +`status="running"`, and `wait` blocks until it ends. Without +`output_format` a crawl returns URLs only; `output_format="html"` also +returns each kept URL's page. + +```python +import os +from zenrows import ZenRowsCrawlClient + +client = ZenRowsCrawlClient(api_key=os.environ["ZENROWS_API_KEY"]) + +crawl = client.create( + "https://example.com/products/", + depth=1, # link hops from the start URL + max_items=50, # stop after keeping 50 URLs (default 10) + max_pages=100, # stop after fetching 100 pages (default 10) + include_patterns=["/product/"], # keep only URLs containing this +) +crawl = client.wait(crawl.crawl_id, timeout=600) # Crawl, terminal +print(crawl.status.value, crawl.stop_reason, crawl.coverage) + +for result in client.iter_results(crawl.crawl_id): # follows next_cursor + print(result.url) +``` + +`wait` returns a `failed` crawl rather than raising (read `crawl.error`), +and raises `WaiterTimeout` on timeout without stopping the crawl. +`iter_results` stops when the server's `next_cursor` is null, which happens +only once the crawl has ended; on a running crawl it follows new results +live until the crawl ends. `get(crawl_id, cursor=..., limit=...)` returns +one raw page. + +### Read the pages + +Pass `output_format="html"` to also fetch the page of every kept URL: + +```python +crawl = client.create(start_url, depth=1, output_format="html") +client.wait(crawl.crawl_id) + +for result in client.iter_results(crawl.crawl_id): + if result.content_id: # content_status == fetched + html = client.get_content(crawl.crawl_id, result) + +# Or everything in one NDJSON file: {"url", "content_status", "content"} per line. +client.download(crawl.crawl_id, "out/crawl.jsonl") +for line in client.iter_download(crawl.crawl_id): # streamed, parsed + print(line.url, len(line.content or "")) +``` + +### List and stop crawls + +```python +for c in client.iter_crawls(): # newest first, without results + print(c.crawl_id, c.status.value) + +client.stop(crawl_id) # idempotent; read final counts with get() +``` + +### Crawl errors + +Every non-2xx raises `CrawlAPIError`. Branch on `status_code` and `code`: + +```python +from zenrows.crawl import CrawlAPIError + +try: + client.create(start_url, depth=1) +except CrawlAPIError as exc: + if exc.not_enabled: # 403 REQS008 + print("Crawl is not enabled for this account") + elif exc.code == "too_many_crawls": # 429: too many crawls running + print("retry in", exc.retry_after, "s") + elif exc.code in ("invalid_parameter", "invalid_start_url"): # 422 + print(exc.detail) + else: + raise +``` + +`crawl_not_found` / `content_not_found` come with 404. A crawl that +ended `failed` is not an error response: read `crawl.error.code` and +`crawl.error.detail` on it. Response enums (`CrawlStatus`, `StopReason`, +`ContentStatus`, ...) are open: a value added later parses as an +`UNKNOWN` member that keeps the raw value. + ## Contributing Pull requests are welcome. For significant changes, please open an issue first to discuss what you would like to change. diff --git a/examples/11_crawl.py b/examples/11_crawl.py new file mode 100644 index 0000000..1d3246e --- /dev/null +++ b/examples/11_crawl.py @@ -0,0 +1,70 @@ +"""11: Crawl a site → wait → read URLs and pages. + +Demonstrates the Crawl client (`ZenRowsCrawlClient`): + - `create(url, depth=..., include_patterns=..., output_format="html")` + starts a crawl and returns at once, `running`. + - `wait(crawl_id)` blocks until it ends and returns the final `Crawl` + (a `failed` crawl is returned, with `error` set). + - `iter_results(crawl_id)` pages through every kept URL. + - `get_content(crawl_id, result)` reads one kept URL's HTML. + - `download(crawl_id, path)` saves every result as NDJSON. + +Run with: + export ZENROWS_API_KEY=zr_... + python examples/11_crawl.py --url https://example.com/products/ +""" + +import argparse +import os + +from zenrows import ZenRowsCrawlClient +from zenrows.crawl import CrawlAPIError, CrawlStatus + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--url", required=True, help="the page to start from") + parser.add_argument("--depth", type=int, default=1) + parser.add_argument("--max-items", type=int, default=5) + parser.add_argument("--include", action="append", default=["/product/"]) + parser.add_argument("--out", default="crawl.jsonl") + args = parser.parse_args() + + client = ZenRowsCrawlClient(api_key=os.environ["ZENROWS_API_KEY"]) + + try: + crawl = client.create( + args.url, + depth=args.depth, + max_items=args.max_items, + include_patterns=args.include, + output_format="html", + ) + except CrawlAPIError as exc: + if exc.not_enabled: + raise SystemExit("Crawl is not enabled for this account.") from exc + raise + print(f"started {crawl.crawl_id}") + + crawl = client.wait(crawl.crawl_id, timeout=600.0) + cov = crawl.coverage + print( + f"{crawl.crawl_id} {crawl.status.value}: {cov.items_found} URLs kept, " + f"{cov.pages_fetched} pages fetched ({cov.pages_failed} failed)" + ) + if crawl.status is CrawlStatus.FAILED and crawl.error: + raise SystemExit(f"crawl failed: {crawl.error.code.value}: {crawl.error.detail}") + + for result in client.iter_results(crawl.crawl_id): + status = result.content_status.value if result.content_status else "-" + print(f" [{status}] {result.url}") + if result.content_id: + html = client.get_content(crawl.crawl_id, result) + print(f" {len(html)} chars of HTML") + + path = client.download(crawl.crawl_id, args.out) + print(f"wrote {path}") + + +if __name__ == "__main__": + main() diff --git a/examples/README.md b/examples/README.md index ca93890..7f102d6 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,6 +1,7 @@ # Examples -Runnable samples for the Batch API client (`ZenRowsBatchClient`). +Runnable samples for the Batch API client (`ZenRowsBatchClient`) and +the Crawl API client (`ZenRowsCrawlClient`, `11_crawl.py`). Each script reads the API key from the environment: ```bash @@ -27,6 +28,7 @@ uv run python examples/02_download_to_dir.py --job-id 01J... | `08_download_all_results.py` | `download_all_results()` — server-side export zip of a whole run | | `09_scheduled_jobs.py` | scheduled cadences (`Rate`/`Calendar`/`At`), `pause`/`resume`/`update_schedule`, per-fire runs, `retry_failed` | | `10_lightweight_handles.py` | `client.job(id)` / `client.run(id, rid)` — act on a known id without a GET (`.data` lazily fetches) | +| `11_crawl.py` | Crawl: `create` → `wait` → `iter_results` + `get_content` → `download` (NDJSON) | The samples are deliberately small — each demonstrates one feature end-to-end so they double as living documentation. diff --git a/pyproject.toml b/pyproject.toml index f697fb2..fbad15b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -112,5 +112,9 @@ exclude = ["src/zenrows/batch/models.py"] # --- pytest --- [tool.pytest.ini_options] testpaths = ["tests"] -addopts = "-ra --strict-markers" +# e2e tests hit a live API; `make test-e2e` selects them with `-m e2e`. +addopts = "-ra --strict-markers -m 'not e2e'" asyncio_mode = "auto" +markers = [ + "e2e: end-to-end tests against a live ZenRows API (opt-in, see DEVELOPMENT.md)", +] diff --git a/src/zenrows/__init__.py b/src/zenrows/__init__.py index 9bbf8c1..4da0a61 100644 --- a/src/zenrows/__init__.py +++ b/src/zenrows/__init__.py @@ -1,6 +1,6 @@ """Zenrows Python SDK. -Two clients live here: +Three clients live here: - `ZenRowsClient` — the original synchronous scraping API client. Backward-compatible with pre-1.4 usage; freshened-up internals. @@ -8,10 +8,14 @@ - `ZenRowsBatchClient` — the new async-job / batch API client. Built on a generated OpenAPI core with a hand-written ergonomic facade on top. + + - `ZenRowsCrawlClient` — the Crawl API client: start a crawl from + one URL and read back the URLs (and pages) it finds. """ from zenrows.__version__ import __version__ from zenrows.batch import ZenRowsBatchClient from zenrows.client import ZenRowsClient +from zenrows.crawl import ZenRowsCrawlClient -__all__ = ["ZenRowsBatchClient", "ZenRowsClient", "__version__"] +__all__ = ["ZenRowsBatchClient", "ZenRowsClient", "ZenRowsCrawlClient", "__version__"] diff --git a/src/zenrows/batch/_transport.py b/src/zenrows/batch/_transport.py index 20e45c5..f6c95bf 100644 --- a/src/zenrows/batch/_transport.py +++ b/src/zenrows/batch/_transport.py @@ -4,10 +4,18 @@ User-Agent, automatic retries for transient failures, and the RFC 7807 → `BatchAPIError` mapping. The facade (`client.py`) is thin glue over this — one method per endpoint, all typed via pydantic v2 models. + +The Crawl client (`zenrows.crawl`) reuses it with its own error type +and logger: both APIs share the key header, problem+json errors and +retry policy. """ +import json +import logging import random import time +from collections.abc import Callable, Iterator +from contextlib import contextmanager from typing import Any, TypeVar import httpx @@ -78,8 +86,15 @@ def __init__( retries: int = _DEFAULT_RETRIES, verify: bool | str = True, httpx_args: dict[str, Any] | None = None, + error_from_response: Callable[[httpx.Response], Exception] = BatchAPIError.from_response, + log: logging.Logger = _log, ): + """`error_from_response` turns a non-2xx response into the + exception to raise; it should expose `.code` and `.detail` for + the error log. `log` receives the per-request records.""" self._retries = max(0, retries) + self._error_from_response = error_from_response + self._log = log self._client = httpx.Client( base_url=base_url.rstrip("/"), headers={ @@ -185,11 +200,25 @@ def request_json( payload = body.model_dump( mode="json", exclude_unset=True, exclude_none=True, by_alias=True ) - import json - content = json.dumps(payload).encode("utf-8") headers = {**(headers or {}), "Content-Type": "application/json"} + response = self.request(method, path, params=params, headers=headers, content=content) + if response.status_code == 204 or not response.content: + return None + return response.json() + + def request( + self, + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + headers: dict[str, str] | None = None, + content: bytes | None = None, + ) -> httpx.Response: + """Send a request (with retries) and return the raw response, + raising the mapped error on non-2xx. For non-JSON bodies.""" start = time.monotonic() response = self._send( method, @@ -199,25 +228,41 @@ def request_json( content=content, ) elapsed_ms = (time.monotonic() - start) * 1000 - - if response.status_code >= 400: - err = BatchAPIError.from_response(response) - log_error( - _log, - method=method, - path=path, - status=response.status_code, - code=err.code, - detail=err.problem.detail if err.problem else None, - ) - raise err - + self._raise_for_status(method, path, response) log_request( - _log, method=method, path=path, status=response.status_code, elapsed_ms=elapsed_ms + self._log, method=method, path=path, status=response.status_code, elapsed_ms=elapsed_ms ) - if response.status_code == 204 or not response.content: - return None - return response.json() + return response + + @contextmanager + def stream( + self, + method: str, + path: str, + *, + params: dict[str, Any] | None = None, + ) -> Iterator[httpx.Response]: + """Open a streamed response, raising the mapped error on + non-2xx. Not retried: a stream may be partly consumed.""" + with self._client.stream(method, path, params=_drop_none(params)) as response: + if response.status_code >= 400: + response.read() + self._raise_for_status(method, path, response) + yield response + + def _raise_for_status(self, method: str, path: str, response: httpx.Response) -> None: + if response.status_code < 400: + return + err = self._error_from_response(response) + log_error( + self._log, + method=method, + path=path, + status=response.status_code, + code=getattr(err, "code", "internal"), + detail=getattr(err, "detail", None), + ) + raise err def _drop_none(d: dict[str, Any] | None) -> dict[str, Any] | None: diff --git a/src/zenrows/crawl/__init__.py b/src/zenrows/crawl/__init__.py new file mode 100644 index 0000000..2f2021e --- /dev/null +++ b/src/zenrows/crawl/__init__.py @@ -0,0 +1,51 @@ +"""Public surface for the ZenRows Crawl API. + +What's where: + - `client.ZenRowsCrawlClient` — the typed facade: create, get, list, + stop, contents, download, and a `wait` helper. + - `models` — hand-written pydantic v2 response models with open enums. + - `errors.CrawlAPIError` — RFC 9457 problem+json mapping. + +`create(output_format="html")` also returns each kept URL's page; +without it a crawl returns URLs only. +""" + +from zenrows.batch._waiters import WaiterTimeout +from zenrows.crawl.client import DEFAULT_BASE_URL, ZenRowsCrawlClient +from zenrows.crawl.errors import CRAWL_NOT_ENABLED, CrawlAPIError +from zenrows.crawl.models import ( + ContentStatus, + Coverage, + Crawl, + CrawlErrorCode, + CrawlList, + CrawlResult, + CrawlStatus, + CrawlStop, + CrawlWithResults, + DownloadLine, + OutputFormat, + RunError, + StopReason, +) + +__all__ = [ + "CRAWL_NOT_ENABLED", + "DEFAULT_BASE_URL", + "ContentStatus", + "Coverage", + "Crawl", + "CrawlAPIError", + "CrawlErrorCode", + "CrawlList", + "CrawlResult", + "CrawlStatus", + "CrawlStop", + "CrawlWithResults", + "DownloadLine", + "OutputFormat", + "RunError", + "StopReason", + "WaiterTimeout", + "ZenRowsCrawlClient", +] diff --git a/src/zenrows/crawl/client.py b/src/zenrows/crawl/client.py new file mode 100644 index 0000000..30d6549 --- /dev/null +++ b/src/zenrows/crawl/client.py @@ -0,0 +1,335 @@ +"""ZenRowsCrawlClient — the typed facade over the Crawl API. + +A crawl starts from one URL, follows links up to `depth` hops, and +keeps the URLs that match its patterns, up to `max_items`. It runs as +a job: `create` returns at once with `status=running`, and the URLs +grow as pages finish. + + client = ZenRowsCrawlClient(api_key) + crawl = client.create("https://example.com/shop/", depth=1, + include_patterns=["/product/"]) + client.wait(crawl.crawl_id) + for result in client.iter_results(crawl.crawl_id): + print(result.url) + +Requests go through the SDK's shared HTTP transport (auth header, +retries, problem+json mapping), which raises `CrawlAPIError`. +""" + +import json +import logging +import os +import time +from collections.abc import Iterator +from pathlib import Path +from typing import Any, Literal + +import httpx + +from zenrows.__version__ import __version__ +from zenrows.batch._transport import _Transport +from zenrows.batch._waiters import poll_until +from zenrows.crawl.errors import CrawlAPIError +from zenrows.crawl.models import ( + Crawl, + CrawlList, + CrawlResult, + CrawlStatus, + CrawlStop, + CrawlWithResults, + DownloadLine, +) + +# Production endpoint. The `base_url=` kwarg + `ZENROWS_CRAWL_BASE_URL` +# env var are present for advanced use only. +DEFAULT_BASE_URL = "https://api.zenrows.com/v1" +DEFAULT_USER_AGENT = f"zenrows-crawl-python/{__version__}" + +_log = logging.getLogger("zenrows.crawl.transport") + + +class ZenRowsCrawlClient: + """Synchronous, typed client for the ZenRows Crawl API. + + `api_key` is required: + + client = ZenRowsCrawlClient(api_key=os.environ["ZENROWS_API_KEY"]) + + Every non-2xx raises `CrawlAPIError`; branch on its `status_code` + and `code`. An account without Crawl enabled gets 403 `REQS008` + (`CrawlAPIError.not_enabled`). + """ + + def __init__( + self, + api_key: str, + *, + base_url: str | None = None, + timeout: float | httpx.Timeout = 30.0, + retries: int = 3, + verify_ssl: bool | str = True, + user_agent: str = DEFAULT_USER_AGENT, + httpx_args: dict[str, Any] | None = None, + ): + """`retries` bounds automatic retries of transient failures + (HTTP 429/502/503/504 and network errors) on idempotent + requests — GETs, and a `create` that carries an + `idempotency_key`. Retries use jittered exponential backoff and + honor `Retry-After`; set `retries=0` to disable.""" + if not api_key: + raise ValueError("ZenRowsCrawlClient: api_key is required.") + base_url = base_url or os.environ.get("ZENROWS_CRAWL_BASE_URL") or DEFAULT_BASE_URL + self._t = _Transport( + base_url=base_url, + api_key=api_key, + user_agent=user_agent, + timeout=timeout, + retries=retries, + verify=verify_ssl, + httpx_args=httpx_args, + error_from_response=CrawlAPIError.from_response, + log=_log, + ) + + # ----- lifecycle ----- + + def __enter__(self) -> "ZenRowsCrawlClient": + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + def close(self) -> None: + self._t.close() + + @property + def base_url(self) -> str: + return self._t.base_url + + # ===== crawls ===== + + def create( + self, + url: str, + *, + depth: int, + max_items: int | None = None, + max_pages: int | None = None, + include_patterns: list[str] | None = None, + exclude_patterns: list[str] | None = None, + output_format: Literal["html"] | None = None, + idempotency_key: str | None = None, + ) -> Crawl: + """`POST /crawls` — start a crawl and return it, `running`. + + - `depth`: link hops to follow from `url` (1 to 100,000). + - `max_items`: stop after keeping this many URLs (server + default 10). + - `max_pages`: stop after fetching this many pages, which + bounds the cost (server default 10). + - `include_patterns` / `exclude_patterns`: substrings matched + against each normalized URL. A URL is kept when it matches + an include pattern (if any) and no exclude pattern. + - `output_format="html"`: also fetch every kept URL's page, + readable with `get_content` / `download`. Omit for URLs only. + - `idempotency_key`: a retry with the same key and body answers + with the crawl the first request created. + + Only the arguments you pass are sent. An account with too many + crawls running gets 429 `too_many_crawls`; retry after the + error's `retry_after` seconds. + """ + if output_format not in (None, "html"): + raise ValueError(f"create: output_format must be 'html' or None, not {output_format!r}") + body: dict[str, Any] = {"url": url, "depth": depth} + optional: dict[str, Any] = { + "max_items": max_items, + "max_pages": max_pages, + "include_patterns": include_patterns, + "exclude_patterns": exclude_patterns, + "output_format": output_format, + } + body.update({k: v for k, v in optional.items() if v is not None}) + headers = {"Content-Type": "application/json"} + if idempotency_key: + headers["Idempotency-Key"] = idempotency_key + response = self._t.request( + "POST", "/crawls", headers=headers, content=json.dumps(body).encode("utf-8") + ) + return Crawl.model_validate(response.json()) + + def get( + self, + crawl_id: str, + *, + cursor: str | None = None, + limit: int | None = None, + ) -> CrawlWithResults: + """`GET /crawls/{crawl_id}` — the crawl and one page of its + results (`limit` 1 to 10,000, server default 1,000). Pass the + page's `next_cursor` as `cursor` for the URLs kept since. For + every result, prefer `iter_results`.""" + return CrawlWithResults.model_validate( + self._t.request_json( + "GET", f"/crawls/{crawl_id}", params={"cursor": cursor, "limit": limit} + ) + ) + + def list_crawls( + self, + *, + cursor: str | None = None, + limit: int | None = None, + ) -> CrawlList: + """`GET /crawls` — one page of the account's crawls, newest + first (`limit` 1 to 100, server default 20), without results. For + most uses prefer the auto-paginating `iter_crawls`.""" + return CrawlList.model_validate( + self._t.request_json("GET", "/crawls", params={"cursor": cursor, "limit": limit}) + ) + + def iter_crawls(self, *, page_size: int | None = None) -> Iterator[Crawl]: + """Auto-paginate `list_crawls`, newest first.""" + cursor: str | None = None + while True: + page = self.list_crawls(cursor=cursor, limit=page_size) + yield from page.crawls + cursor = page.next_cursor + if not cursor: + return + + def iter_results( + self, + crawl_id: str, + *, + page_size: int | None = None, + poll_interval: float = 2.0, + ) -> Iterator[CrawlResult]: + """Yield every URL the crawl keeps, in order, each once. + + Follows `next_cursor` until the server returns null, which it + does only once the crawl has ended and its last page was read. + On a crawl that is still running, the iterator follows it live: + when it has caught up, it sleeps `poll_interval` seconds and + asks again with the same cursor, so it ends when the crawl ends. + Call `wait` first to read a finished crawl without polling. + """ + cursor: str | None = None + while True: + page = self.get(crawl_id, cursor=cursor, limit=page_size) + yield from page.results + if page.next_cursor is None: + return + if not page.results and page.status is CrawlStatus.RUNNING: + time.sleep(poll_interval) + cursor = page.next_cursor + + def stop(self, crawl_id: str) -> CrawlStop: + """`POST /crawls/{crawl_id}/stop` — stop a running crawl. + + Idempotent: a crawl that already ended answers as it ended. + The answer carries no counts; read the final coverage and + results with `get` once pages already in flight have finished. + """ + return CrawlStop.model_validate(self._t.request_json("POST", f"/crawls/{crawl_id}/stop")) + + # ===== contents ===== + + def get_content(self, crawl_id: str, content: str | CrawlResult) -> str: + """`GET /crawls/{crawl_id}/contents/{content_id}` — one kept + URL's page, as HTML text. + + `content` is a `CrawlResult` whose `content_status` is + `fetched`, or its content id (the last segment of its + `content_url`). Only crawls created with an `output_format` + have contents; anything else is 404 `content_not_found`. + """ + if isinstance(content, CrawlResult): + content_id = content.content_id + if not content_id: + raise ValueError( + f"get_content: {content.url} has no content_url " + f"(content_status={content.content_status})" + ) + else: + content_id = content + return self._t.request("GET", f"/crawls/{crawl_id}/contents/{content_id}").text + + def iter_download(self, crawl_id: str) -> Iterator[DownloadLine]: + """`GET /crawls/{crawl_id}/download` — stream every result as + a parsed `DownloadLine` (url, content_status, content). + + On a running crawl the file holds what the crawl has kept so + far (a warning is logged); call `wait` first for the whole of it. + """ + with self._t.stream("GET", f"/crawls/{crawl_id}/download") as response: + _warn_if_partial(crawl_id, response) + for line in response.iter_lines(): + if line.strip(): + yield DownloadLine.model_validate_json(line) + + def download( + self, + crawl_id: str, + target_path: str | Path, + *, + chunk_size: int = 64 * 1024, + ) -> Path: + """`GET /crawls/{crawl_id}/download` — stream the NDJSON file + (one JSON object per result) to `target_path`, and return it. + + On a running crawl the file holds what the crawl has kept so + far (a warning is logged); call `wait` first for the whole of it. + """ + target = Path(target_path) + target.parent.mkdir(parents=True, exist_ok=True) + with self._t.stream("GET", f"/crawls/{crawl_id}/download") as response: + _warn_if_partial(crawl_id, response) + with target.open("wb") as f: + for chunk in response.iter_bytes(chunk_size): + f.write(chunk) + return target + + # ===== waiter ===== + + def wait( + self, + crawl_id: str, + *, + timeout: float = 300.0, + poll_interval: float = 2.0, + max_poll_interval: float = 15.0, + ) -> Crawl: + """Block until the crawl ends (`completed`, `stopped` or + `failed`) and return it, polling with jittered exponential + backoff from `poll_interval` up to `max_poll_interval` seconds. + + A `failed` crawl is returned, not raised: read `error` on it. + Raises `WaiterTimeout` after `timeout` seconds; the crawl keeps + running (call `stop` to end it). + """ + + def fetch() -> Crawl: + # limit=1 keeps each poll cheap; results are not returned. + page = self.get(crawl_id, limit=1) + return Crawl.model_validate(page.model_dump(exclude={"results", "next_cursor"})) + + return poll_until( + fetch=fetch, + is_done=lambda crawl: crawl.is_terminal, + timeout=timeout, + initial_interval=poll_interval, + max_interval=max_poll_interval, + ) + + +def _warn_if_partial(crawl_id: str, response: httpx.Response) -> None: + status = response.headers.get("X-Crawl-Status") + if status == CrawlStatus.RUNNING.value: + logging.getLogger("zenrows.crawl.client").warning( + "crawl %s is still running: the download holds only the results kept so far", + crawl_id, + ) + + +__all__ = ["DEFAULT_BASE_URL", "DEFAULT_USER_AGENT", "ZenRowsCrawlClient"] diff --git a/src/zenrows/crawl/errors.py b/src/zenrows/crawl/errors.py new file mode 100644 index 0000000..b1f5b9d --- /dev/null +++ b/src/zenrows/crawl/errors.py @@ -0,0 +1,72 @@ +"""RFC 9457 problem+json → `CrawlAPIError`. + +The Crawl API answers errors as `application/problem+json` with a +stable `code`. Branch on +`status_code` + `code`; display `detail`, don't parse it. +""" + +import httpx + +from zenrows.batch.errors import ProblemDetail + +# The code of the 403 answer to a key whose account does not have +# Crawl enabled. +CRAWL_NOT_ENABLED = "REQS008" + + +class CrawlAPIError(Exception): + """A non-2xx response from the Crawl API. + + `code` is the problem's `code` member: `crawl_not_found`, + `content_not_found`, `invalid_parameter`, `too_many_crawls`, ..., + `REQS008` when Crawl is not enabled for the account, and the API's + usual auth and credit codes. `problem` + holds the whole body, or None when it was not JSON. + """ + + def __init__( + self, + status_code: int, + problem: ProblemDetail | None, + raw: bytes, + *, + retry_after: float | None = None, + ): + self.status_code = status_code + # Seconds the server asked to wait (`Retry-After`), e.g. on 429 + # `too_many_crawls`. + self.retry_after = retry_after + self.problem = problem + self.raw = raw + self.code: str = problem.code if problem else "internal" + self.detail: str | None = problem.detail if problem else None + if problem and self.code == CRAWL_NOT_ENABLED: + msg = f"{status_code} Crawl is not enabled for this account ({CRAWL_NOT_ENABLED})" + ( + f": {problem.detail}" if problem.detail else "" + ) + elif problem: + msg = f"{status_code} {problem.title}: {problem.detail or problem.code}" + else: + msg = f"{status_code} (no problem body)" + super().__init__(msg) + + @property + def not_enabled(self) -> bool: + """True when the account does not have Crawl enabled (403 REQS008).""" + return self.status_code == 403 and self.code == CRAWL_NOT_ENABLED + + @classmethod + def from_response(cls, response: httpx.Response) -> "CrawlAPIError": + try: + retry_after: float | None = float(response.headers.get("Retry-After", "")) + except ValueError: + retry_after = None + return cls( + status_code=response.status_code, + problem=ProblemDetail.from_response(response), + raw=response.content, + retry_after=retry_after, + ) + + +__all__ = ["CRAWL_NOT_ENABLED", "CrawlAPIError"] diff --git a/src/zenrows/crawl/models.py b/src/zenrows/crawl/models.py new file mode 100644 index 0000000..e3249e0 --- /dev/null +++ b/src/zenrows/crawl/models.py @@ -0,0 +1,193 @@ +"""Pydantic v2 models for the Crawl API's responses. + +Hand-written: the surface is a handful of small schemas. + +Forward compatible: unknown response fields +are ignored (pydantic's default), and every enum is open: a value the +server adds later parses as an `UNKNOWN` member that keeps the raw +value (see `zenrows.batch._open_enum`). +""" + +from datetime import datetime +from enum import Enum + +from pydantic import BaseModel + +from zenrows.batch._open_enum import open_enum_missing + + +class CrawlStatus(Enum): + """Where a crawl stands. `running` until it ends; the other values + are terminal. Clients must accept values not listed here.""" + + RUNNING = "running" + COMPLETED = "completed" + STOPPED = "stopped" + FAILED = "failed" + + _missing_ = classmethod(open_enum_missing) + + +class StopReason(Enum): + """What ended the crawl before nothing was left to open: a limit + (`max_items` / `max_pages`, with `completed`) or the caller (`user`, + with `stopped`).""" + + MAX_ITEMS = "max_items" + MAX_PAGES = "max_pages" + USER = "user" + + _missing_ = classmethod(open_enum_missing) + + +class ContentStatus(Enum): + """Where a kept URL's page stands, when the crawl has an + `output_format`.""" + + PENDING = "pending" + FETCHED = "fetched" + FAILED = "failed" + + _missing_ = classmethod(open_enum_missing) + + +class OutputFormat(Enum): + """The format the crawl returns each kept URL's page in.""" + + HTML = "html" + + _missing_ = classmethod(open_enum_missing) + + +class CrawlErrorCode(Enum): + """Why a crawl failed (`Crawl.error.code`).""" + + INSUFFICIENT_CREDITS = "insufficient_credits" + SEED_UNREACHABLE = "seed_unreachable" + DOMAIN_NOT_ALLOWED = "domain_not_allowed" + NO_ITEMS_FOUND = "no_items_found" + INTERNAL_ERROR = "internal_error" + + _missing_ = classmethod(open_enum_missing) + + +class RunError(BaseModel): + """Why a crawl failed. Present only when `status` is `failed`; part + of the crawl, not an error response.""" + + code: CrawlErrorCode + detail: str + + +class Coverage(BaseModel): + """How far the crawl has got.""" + + pages_fetched: int + """Pages fetched so far, successful or not.""" + pages_failed: int + """Fetched pages whose fetch failed (not charged).""" + items_found: int + """URLs kept so far, bounded by `max_items`.""" + + +class Crawl(BaseModel): + """One crawl, with the parameters it runs with.""" + + crawl_id: str + status: CrawlStatus + url: str + """The page the crawl started from.""" + depth: int + max_items: int + max_pages: int + coverage: Coverage + created_at: datetime + stop_reason: StopReason | None = None + error: RunError | None = None + include_patterns: list[str] | None = None + exclude_patterns: list[str] | None = None + output_format: OutputFormat | None = None + duplicates_removed: int | None = None + """Links met again and dropped as already seen.""" + finished_at: datetime | None = None + """When the crawl reached a terminal status. Absent while it runs.""" + + @property + def is_terminal(self) -> bool: + """True once the crawl has ended (any status but `running`).""" + return self.status is not CrawlStatus.RUNNING + + +class CrawlResult(BaseModel): + """One URL the crawl kept.""" + + url: str + content_status: ContentStatus | None = None + """Present only when the crawl has an `output_format`.""" + content_url: str | None = None + """The page's path (e.g. `/v1/crawls/c_x/contents/ct_y`), present + once `content_status` is `fetched`. Pass the result itself to + `ZenRowsCrawlClient.get_content`.""" + + @property + def content_id(self) -> str | None: + """The content id, the last segment of `content_url`.""" + if not self.content_url: + return None + return self.content_url.rstrip("/").rsplit("/", 1)[-1] + + +class CrawlWithResults(Crawl): + """A crawl and one page of the URLs it has kept (`GET /crawls/{id}`). + + `next_cursor` is never null while the crawl runs: polling with it + returns only URLs kept since. It is null once the crawl has ended + and this page holds its last URLs.""" + + results: list[CrawlResult] + next_cursor: str | None = None + + +class CrawlList(BaseModel): + """One page of the account's crawls, newest first. `next_cursor` is + absent on the last page.""" + + crawls: list[Crawl] + next_cursor: str | None = None + + +class CrawlStop(BaseModel): + """What `stop` answers: where the crawl stands, without counts. Read + the coverage and results with `get`.""" + + crawl_id: str + status: CrawlStatus + stop_reason: StopReason | None = None + error: RunError | None = None + finished_at: datetime | None = None + + +class DownloadLine(BaseModel): + """One line of a crawl's NDJSON download.""" + + url: str + content_status: ContentStatus | None = None + content: str | None = None + """The page HTML, present when `content_status` is `fetched`.""" + + +__all__ = [ + "ContentStatus", + "Coverage", + "Crawl", + "CrawlErrorCode", + "CrawlList", + "CrawlResult", + "CrawlStatus", + "CrawlStop", + "CrawlWithResults", + "DownloadLine", + "OutputFormat", + "RunError", + "StopReason", +] diff --git a/tests/e2e/test_crawl_e2e.py b/tests/e2e/test_crawl_e2e.py new file mode 100644 index 0000000..d37db0d --- /dev/null +++ b/tests/e2e/test_crawl_e2e.py @@ -0,0 +1,109 @@ +"""End-to-end test of ZenRowsCrawlClient against a live Crawl API. + +Opt-in: runs only with `make test-e2e` (pytest `-m e2e`), and only when +`ZENROWS_API_KEY` (a key with Crawl access), `ZENROWS_CRAWL_BASE_URL` +(the API base, e.g. https://api.zenrows.com/v1) and +`ZENROWS_E2E_CRAWL_URL` (the page to crawl from) are set. Otherwise it is +skipped, so `make test` stays offline. When `ZENROWS_E2E_CRAWL_INCLUDE` +is set, the crawl keeps only URLs containing it, and the test checks +that every result does. It starts one small crawl (a few credits) and +reads it back every way the client can. +""" + +import json +import os +import time + +import pytest + +from zenrows.crawl import CrawlAPIError, CrawlStatus, ZenRowsCrawlClient + +_REQUIRED = ("ZENROWS_API_KEY", "ZENROWS_CRAWL_BASE_URL", "ZENROWS_E2E_CRAWL_URL") + +pytestmark = [ + pytest.mark.e2e, + pytest.mark.skipif( + not all(os.environ.get(v) for v in _REQUIRED), + reason="set " + ", ".join(_REQUIRED) + " to run the Crawl e2e test", + ), +] + +START_URL = os.environ.get("ZENROWS_E2E_CRAWL_URL", "") +INCLUDE = os.environ.get("ZENROWS_E2E_CRAWL_INCLUDE") or None +# An account can run only a few crawls at once, so a create can meet +# 429 too_many_crawls while other runs finish. +SLOT_WAIT_SECONDS = 300 + + +@pytest.fixture(scope="module") +def client(): + with ZenRowsCrawlClient(api_key=os.environ["ZENROWS_API_KEY"]) as c: + yield c + + +def create_when_a_slot_frees(client: ZenRowsCrawlClient): + deadline = time.monotonic() + SLOT_WAIT_SECONDS + while True: + try: + return client.create( + START_URL, + depth=1, + max_items=3, + max_pages=5, + include_patterns=[INCLUDE] if INCLUDE else None, + output_format="html", + ) + except CrawlAPIError as err: + if err.code != "too_many_crawls" or time.monotonic() >= deadline: + raise + wait = err.retry_after or 30.0 + print(f"too_many_crawls: retrying create in {wait:.0f}s") + time.sleep(wait) + + +def test_crawl_end_to_end(client: ZenRowsCrawlClient, tmp_path): + created = create_when_a_slot_frees(client) + print(f"created {created.crawl_id} status={created.status.value}") + assert created.crawl_id + + crawl = client.wait(created.crawl_id, timeout=600) + print( + f"finished {crawl.crawl_id} status={crawl.status.value} " + f"stop_reason={crawl.stop_reason and crawl.stop_reason.value} " + f"pages_fetched={crawl.coverage.pages_fetched} items_found={crawl.coverage.items_found}" + ) + assert crawl.status is CrawlStatus.COMPLETED, crawl.error + + results = list(client.iter_results(crawl.crawl_id)) + print(f"results: {len(results)}") + assert results + if INCLUDE: + assert all(INCLUDE in r.url for r in results), [r.url for r in results] + + fetched = [r for r in results if r.content_id] + assert fetched, "no result has a fetched page" + html = client.get_content(crawl.crawl_id, fetched[0]) + print(f"content of {fetched[0].url}: {len(html)} chars") + assert " ZenRowsCrawlClient: + return ZenRowsCrawlClient(api_key=API_KEY, base_url=BASE_URL, retries=0) + + +@pytest.fixture +def no_sleep(monkeypatch): + """Record + swallow sleeps (waiter, scanner and retry backoff all + call `time.sleep`) so tests run instantly.""" + slept: list[float] = [] + monkeypatch.setattr("zenrows.crawl.client.time.sleep", slept.append) + return slept + + +# ----- construction ----- + + +def test_missing_api_key_raises(): + with pytest.raises(ValueError): + ZenRowsCrawlClient(api_key="") + + +def test_default_base_url(monkeypatch): + monkeypatch.delenv("ZENROWS_CRAWL_BASE_URL", raising=False) + assert ZenRowsCrawlClient(api_key=API_KEY).base_url.rstrip("/") == "https://api.zenrows.com/v1" + + +def test_base_url_env_and_kwarg_precedence(monkeypatch): + monkeypatch.setenv("ZENROWS_CRAWL_BASE_URL", "http://env-set/v1") + assert ZenRowsCrawlClient(api_key=API_KEY).base_url.rstrip("/") == "http://env-set/v1" + assert ZenRowsCrawlClient(api_key=API_KEY, base_url=BASE_URL).base_url.rstrip("/") == BASE_URL + + +# ----- create ----- + + +@respx.mock +def test_create_sends_only_required_fields(client: ZenRowsCrawlClient): + route = respx.post(f"{BASE_URL}/crawls").mock( + return_value=Response(202, headers={"Location": "/v1/crawls/c_1"}, json=crawl_body()) + ) + + crawl = client.create("https://example.com/", depth=1) + + assert crawl.crawl_id == "c_1" + assert crawl.status is CrawlStatus.RUNNING + assert not crawl.is_terminal + sent = route.calls.last.request + assert sent.headers["X-API-Key"] == API_KEY + assert sent.headers["Content-Type"] == "application/json" + assert "Idempotency-Key" not in sent.headers + assert json.loads(sent.content) == {"url": "https://example.com/", "depth": 1} + + +@respx.mock +def test_create_sends_every_option_and_idempotency_key(client: ZenRowsCrawlClient): + route = respx.post(f"{BASE_URL}/crawls").mock( + return_value=Response(202, json=crawl_body(output_format="html")) + ) + + crawl = client.create( + "https://example.com/", + depth=2, + max_items=3, + max_pages=5, + include_patterns=["/product/"], + exclude_patterns=["/cart"], + output_format="html", + idempotency_key="idem-1", + ) + + assert crawl.output_format is OutputFormat.HTML + sent = route.calls.last.request + assert sent.headers["Idempotency-Key"] == "idem-1" + body = json.loads(sent.content) + assert body == { + "url": "https://example.com/", + "depth": 2, + "max_items": 3, + "max_pages": 5, + "include_patterns": ["/product/"], + "exclude_patterns": ["/cart"], + "output_format": "html", + } + assert "discovery" not in body + + +def test_create_output_format_accepts_only_html(client: ZenRowsCrawlClient): + with pytest.raises(ValueError, match="output_format"): + client.create("https://example.com/", depth=1, output_format="json") # type: ignore[arg-type] + + +# ----- get / list / stop ----- + + +@respx.mock +def test_get_parses_results_and_open_enums(client: ZenRowsCrawlClient): + route = respx.get(f"{BASE_URL}/crawls/c_1").mock( + return_value=Response( + 200, + json=crawl_body( + status="completed", + stop_reason="max_items", + output_format="xml", + finished_at="2026-10-08T09:02:00Z", + some_new_field=True, + results=[ + { + "url": "https://example.com/product/a", + "content_status": "fetched", + "content_url": "/v1/crawls/c_1/contents/ct_a", + }, + {"url": "https://example.com/product/b", "content_status": "archived"}, + ], + next_cursor=None, + ), + ) + ) + + page = client.get("c_1", cursor="cur_1", limit=50) + + assert route.calls.last.request.url.params["cursor"] == "cur_1" + assert route.calls.last.request.url.params["limit"] == "50" + assert page.status is CrawlStatus.COMPLETED + assert page.is_terminal + assert page.stop_reason is StopReason.MAX_ITEMS + assert is_unknown(page.output_format) # type: ignore[arg-type] + assert page.next_cursor is None + assert page.results[0].content_status is ContentStatus.FETCHED + assert page.results[0].content_id == "ct_a" + unknown = page.results[1].content_status + assert unknown is not None and is_unknown(unknown) and unknown.value == "archived" + + +@respx.mock +def test_get_parses_failed_crawl_error(client: ZenRowsCrawlClient): + respx.get(f"{BASE_URL}/crawls/c_1").mock( + return_value=Response( + 200, + json=crawl_body( + status="failed", + error={"code": "seed_unreachable", "detail": "The site may be blocking it."}, + results=[], + next_cursor=None, + ), + ) + ) + page = client.get("c_1") + assert page.status is CrawlStatus.FAILED + assert page.error is not None and page.error.code is CrawlErrorCode.SEED_UNREACHABLE + + +@respx.mock +def test_get_omits_unset_query_params(client: ZenRowsCrawlClient): + route = respx.get(f"{BASE_URL}/crawls/c_1").mock( + return_value=Response(200, json=crawl_body(results=[], next_cursor="cur_0")) + ) + client.get("c_1") + assert dict(route.calls.last.request.url.params) == {} + + +@respx.mock +def test_iter_crawls_paginates_until_cursor_absent(client: ZenRowsCrawlClient): + route = respx.get(f"{BASE_URL}/crawls").mock( + side_effect=[ + Response(200, json={"crawls": [crawl_body(crawl_id="c_2")], "next_cursor": "n1"}), + Response(200, json={"crawls": [crawl_body(crawl_id="c_1")]}), + ] + ) + + ids = [c.crawl_id for c in client.iter_crawls(page_size=1)] + + assert ids == ["c_2", "c_1"] + assert route.calls[0].request.url.params["limit"] == "1" + assert "cursor" not in route.calls[0].request.url.params + assert route.calls[1].request.url.params["cursor"] == "n1" + + +@respx.mock +def test_stop_posts_without_body(client: ZenRowsCrawlClient): + route = respx.post(f"{BASE_URL}/crawls/c_1/stop").mock( + return_value=Response( + 200, + json={ + "crawl_id": "c_1", + "status": "stopped", + "stop_reason": "user", + "finished_at": "2026-10-08T09:01:00Z", + }, + ) + ) + + stopped = client.stop("c_1") + + assert stopped.status is CrawlStatus.STOPPED + assert stopped.stop_reason is StopReason.USER + assert route.calls.last.request.content == b"" + + +# ----- results scanner ----- + + +@respx.mock +def test_iter_results_stops_on_null_cursor(client: ZenRowsCrawlClient, no_sleep): + route = respx.get(f"{BASE_URL}/crawls/c_1").mock( + side_effect=[ + Response(200, json=crawl_body(results=[{"url": "https://a"}], next_cursor="k1")), + # Caught up with a running crawl: no results, same cursor back. + Response(200, json=crawl_body(results=[], next_cursor="k1")), + Response( + 200, + json=crawl_body( + status="completed", results=[{"url": "https://b"}], next_cursor=None + ), + ), + ] + ) + + urls = [r.url for r in client.iter_results("c_1", poll_interval=0.5)] + + assert urls == ["https://a", "https://b"] + assert route.call_count == 3 + assert "cursor" not in route.calls[0].request.url.params + assert route.calls[1].request.url.params["cursor"] == "k1" + assert route.calls[2].request.url.params["cursor"] == "k1" + assert no_sleep == [0.5] # slept only on the empty page of a running crawl + + +# ----- contents / download ----- + + +@respx.mock +def test_get_content_accepts_result_or_id(client: ZenRowsCrawlClient): + route = respx.get(f"{BASE_URL}/crawls/c_1/contents/ct_a").mock( + return_value=Response( + 200, headers={"Content-Type": "text/html"}, text="product" + ) + ) + result = CrawlResult( + url="https://a", content_status="fetched", content_url="/v1/crawls/c_1/contents/ct_a" + ) # type: ignore[arg-type] + + assert client.get_content("c_1", result) == "product" + assert client.get_content("c_1", "ct_a") == "product" + assert route.call_count == 2 + + +def test_get_content_without_content_url_raises(client: ZenRowsCrawlClient): + with pytest.raises(ValueError, match="content_url"): + client.get_content("c_1", CrawlResult(url="https://a", content_status="pending")) # type: ignore[arg-type] + + +NDJSON = ( + b'{"url":"https://a","content_status":"fetched","content":"a"}\n' + b'{"url":"https://b","content_status":"failed"}\n' +) + + +@respx.mock +def test_iter_download_parses_lines(client: ZenRowsCrawlClient): + respx.get(f"{BASE_URL}/crawls/c_1/download").mock( + return_value=Response( + 200, + headers={"Content-Type": "application/x-ndjson", "X-Crawl-Status": "completed"}, + content=NDJSON, + ) + ) + lines = list(client.iter_download("c_1")) + assert [line.url for line in lines] == ["https://a", "https://b"] + assert lines[0].content == "a" + assert lines[1].content_status is ContentStatus.FAILED + + +@respx.mock +def test_download_writes_file_and_warns_when_partial(client: ZenRowsCrawlClient, tmp_path, caplog): + respx.get(f"{BASE_URL}/crawls/c_1/download").mock( + return_value=Response(200, headers={"X-Crawl-Status": "running"}, content=NDJSON) + ) + target = client.download("c_1", tmp_path / "out" / "c_1.jsonl") + assert target.read_bytes() == NDJSON + assert "still running" in caplog.text + + +@respx.mock +def test_download_maps_errors(client: ZenRowsCrawlClient): + respx.get(f"{BASE_URL}/crawls/nope/download").mock(return_value=problem(404, "crawl_not_found")) + with pytest.raises(CrawlAPIError) as exc_info: + list(client.iter_download("nope")) + assert exc_info.value.code == "crawl_not_found" + + +# ----- errors ----- + + +@respx.mock +def test_not_found_maps_to_crawl_api_error(client: ZenRowsCrawlClient): + respx.get(f"{BASE_URL}/crawls/nope").mock(return_value=problem(404, "crawl_not_found")) + with pytest.raises(CrawlAPIError) as exc_info: + client.get("nope") + err = exc_info.value + assert err.status_code == 404 + assert err.code == "crawl_not_found" + assert err.detail == "detail for crawl_not_found" + assert not err.not_enabled + + +@respx.mock +def test_validation_error_maps_to_crawl_api_error(client: ZenRowsCrawlClient): + respx.post(f"{BASE_URL}/crawls").mock(return_value=problem(422, "invalid_start_url")) + with pytest.raises(CrawlAPIError) as exc_info: + client.create("http://10.0.0.1/", depth=1) + assert exc_info.value.status_code == 422 + assert exc_info.value.code == "invalid_start_url" + + +@respx.mock +def test_too_many_crawls_carries_retry_after(client: ZenRowsCrawlClient): + respx.post(f"{BASE_URL}/crawls").mock( + return_value=problem(429, "too_many_crawls", headers={"Retry-After": "30"}) + ) + with pytest.raises(CrawlAPIError) as exc_info: + client.create("https://example.com/", depth=1) + assert exc_info.value.code == "too_many_crawls" + assert exc_info.value.retry_after == 30.0 + + +@respx.mock +def test_crawl_not_enabled_has_clear_message(client: ZenRowsCrawlClient): + respx.post(f"{BASE_URL}/crawls").mock( + return_value=problem( + 403, + "REQS008", + title="Crawl is not enabled for this account.", + type="https://docs.zenrows.com/api-error-codes#REQS008", + ) + ) + with pytest.raises(CrawlAPIError) as exc_info: + client.create("https://example.com/", depth=1) + err = exc_info.value + assert err.not_enabled + assert err.code == "REQS008" + assert "Crawl is not enabled for this account" in str(err) + + +@respx.mock +def test_get_retries_transient_status(no_sleep): + client = ZenRowsCrawlClient(api_key=API_KEY, base_url=BASE_URL, retries=2) + route = respx.get(f"{BASE_URL}/crawls/c_1").mock( + side_effect=[Response(503), Response(200, json=crawl_body(results=[], next_cursor="k"))] + ) + client.get("c_1") + assert route.call_count == 2 + assert len(no_sleep) == 1 + + +# ----- waiter ----- + + +@respx.mock +def test_wait_polls_until_terminal(client: ZenRowsCrawlClient, no_sleep): + route = respx.get(f"{BASE_URL}/crawls/c_1").mock( + side_effect=[ + Response(200, json=crawl_body(results=[], next_cursor="k")), + Response( + 200, + json=crawl_body( + status="completed", + finished_at="2026-10-08T09:02:00Z", + results=[{"url": "https://a"}], + next_cursor="k2", + ), + ), + ] + ) + + crawl = client.wait("c_1", poll_interval=1.0) + + assert crawl.status is CrawlStatus.COMPLETED + assert not hasattr(crawl, "results") + assert route.call_count == 2 + assert all(c.request.url.params["limit"] == "1" for c in route.calls) + assert len(no_sleep) == 1 + + +@respx.mock +def test_wait_times_out(client: ZenRowsCrawlClient, no_sleep, monkeypatch): + respx.get(f"{BASE_URL}/crawls/c_1").mock( + return_value=Response(200, json=crawl_body(results=[], next_cursor="k")) + ) + clock = iter(range(0, 1000, 10)) + monkeypatch.setattr("zenrows.batch._waiters.time.monotonic", lambda: next(clock)) + + with pytest.raises(WaiterTimeout): + client.wait("c_1", timeout=25) From b3bfa1adfcc4deca5b0209d30f4e15d64ca93439 Mon Sep 17 00:00:00 2001 From: Stefano Guerrini Date: Fri, 9 Oct 2026 12:37:28 +0200 Subject: [PATCH 2/2] fix(crawl): align the Crawl client with the production API Mark Crawl as Beta in the README and the module and client docstrings. iter_results no longer polls. It returns at the first empty page. iter_download and download return the X-Crawl-Status status with the lines or the saved path. Download lines accept object content. CrawlAPIError.code is None when the body has no code. The client takes the base URL only from the constructor. create never retries 429 too_many_crawls. Batch retries do not change. wait defaults to 600 s. The README lists every Crawl error code. The example lets --include replace the default pattern. Add a py.typed marker so type checkers use the package types. --- DEVELOPMENT.md | 7 +- README.md | 54 +++++++---- examples/11_crawl.py | 19 ++-- src/zenrows/batch/_transport.py | 23 +++-- src/zenrows/batch/errors.py | 8 +- src/zenrows/crawl/__init__.py | 13 ++- src/zenrows/crawl/client.py | 164 +++++++++++++++++++++----------- src/zenrows/crawl/errors.py | 9 +- src/zenrows/crawl/models.py | 6 +- src/zenrows/py.typed | 0 tests/e2e/test_crawl_e2e.py | 17 ++-- tests/test_batch_client.py | 11 +++ tests/test_crawl_client.py | 113 +++++++++++++++++----- 13 files changed, 302 insertions(+), 142 deletions(-) create mode 100644 src/zenrows/py.typed diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 4198496..1589c68 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -81,7 +81,7 @@ and it skips unless the three required variables below are set. | Variable | | |---|---| | `ZENROWS_API_KEY` | required: a key with Crawl access | -| `ZENROWS_CRAWL_BASE_URL` | required: the API base, e.g. `https://api.zenrows.com/v1` | +| `ZENROWS_CRAWL_BASE_URL` | required: the API base, e.g. `https://api.zenrows.com/v1`; the test passes it as `base_url=` | | `ZENROWS_E2E_CRAWL_URL` | required: the page the crawl starts from | | `ZENROWS_E2E_CRAWL_INCLUDE` | optional: an include pattern; every result must contain it | @@ -98,8 +98,9 @@ the test crawls at depth 1 with `max_items=3` and `output_format="html"`. To test against a local or staging deployment, point `ZENROWS_CRAWL_BASE_URL` at its `/v1` base instead. When the account -has too many crawls running (429 `too_many_crawls`), the test waits and -retries (up to 5 minutes) before creating its crawl. +has reached its limit of active jobs (429 `too_many_crawls`: 3 by default, +shared with its Batch jobs), the test waits and retries (up to 5 minutes) +before creating its crawl. ## Refreshing the OpenAPI spec diff --git a/README.md b/README.md index 8133127..5068b99 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ This package ships three clients: _(Private beta — [contact support](mailto:support@zenrows.com) for access.)_ - **`ZenRowsCrawlClient`** — the Crawl API client. Give it one start URL; it follows links and returns the URLs (and, optionally, the - pages) it finds behind it. + pages) it finds behind it. _(Beta.)_ ## Table of Contents @@ -40,7 +40,7 @@ This package ships three clients: - [Act on an id without a GET](#act-on-an-id-without-a-get) - [Scheduled jobs & webhooks](#scheduled-jobs--webhooks) - [Error handling](#error-handling) -- [Quickstart — Crawl API (`ZenRowsCrawlClient`)](#quickstart--crawl-api-zenrowscrawlclient) +- [Quickstart — Crawl API (beta) (`ZenRowsCrawlClient`)](#quickstart--crawl-api-beta-zenrowscrawlclient) - [Read the pages](#read-the-pages) - [List and stop crawls](#list-and-stop-crawls) - [Crawl errors](#crawl-errors) @@ -492,11 +492,14 @@ The full Batch surface (jobs, runs, tasks, results, content, history, file_inputs, HMAC keys) is reachable via methods on `ZenRowsBatchClient`. See `src/zenrows/batch/client.py` or `help(ZenRowsBatchClient)`. -## Quickstart — Crawl API (`ZenRowsCrawlClient`) +## Quickstart — Crawl API (beta) (`ZenRowsCrawlClient`) + +Crawl is in Beta: its API may still change. A crawl starts from one URL, follows the links on each page up to `depth` hops, and keeps the URLs that match your patterns, up to -`max_items`. It runs in the background: `create` returns at once with +`max_items`. It stays on the start URL's registrable domain; +subdomains count. It runs in the background: `create` returns at once with `status="running"`, and `wait` blocks until it ends. Without `output_format` a crawl returns URLs only; `output_format="html"` also returns each kept URL's page. @@ -514,7 +517,7 @@ crawl = client.create( max_pages=100, # stop after fetching 100 pages (default 10) include_patterns=["/product/"], # keep only URLs containing this ) -crawl = client.wait(crawl.crawl_id, timeout=600) # Crawl, terminal +crawl = client.wait(crawl.crawl_id) # Crawl, terminal (timeout 600 s) print(crawl.status.value, crawl.stop_reason, crawl.coverage) for result in client.iter_results(crawl.crawl_id): # follows next_cursor @@ -523,10 +526,10 @@ for result in client.iter_results(crawl.crawl_id): # follows next_cursor `wait` returns a `failed` crawl rather than raising (read `crawl.error`), and raises `WaiterTimeout` on timeout without stopping the crawl. -`iter_results` stops when the server's `next_cursor` is null, which happens -only once the crawl has ended; on a running crawl it follows new results -live until the crawl ends. `get(crawl_id, cursor=..., limit=...)` returns -one raw page. +`iter_results` does not poll: it returns at the first empty page, so on a +running crawl it yields only the URLs kept so far. Call `wait` first to +read every result. `get(crawl_id, cursor=..., limit=...)` returns one raw +page. ### Read the pages @@ -541,11 +544,18 @@ for result in client.iter_results(crawl.crawl_id): html = client.get_content(crawl.crawl_id, result) # Or everything in one NDJSON file: {"url", "content_status", "content"} per line. -client.download(crawl.crawl_id, "out/crawl.jsonl") -for line in client.iter_download(crawl.crawl_id): # streamed, parsed - print(line.url, len(line.content or "")) +saved = client.download(crawl.crawl_id, "out/crawl.jsonl") +print(saved.path, saved.status) # status from X-Crawl-Status + +with client.iter_download(crawl.crawl_id) as download: # streamed, parsed + print(download.status) + for line in download: + print(line.url, len(line.content or "")) ``` +Both downloads carry the crawl's `status` when the file was read. On a +`running` crawl the file holds only the results kept so far. + ### List and stop crawls ```python @@ -567,17 +577,27 @@ try: except CrawlAPIError as exc: if exc.not_enabled: # 403 REQS008 print("Crawl is not enabled for this account") - elif exc.code == "too_many_crawls": # 429: too many crawls running + elif exc.code == "too_many_crawls": # 429: active jobs limit reached print("retry in", exc.retry_after, "s") - elif exc.code in ("invalid_parameter", "invalid_start_url"): # 422 + elif exc.status_code in (400, 422): print(exc.detail) else: raise ``` -`crawl_not_found` / `content_not_found` come with 404. A crawl that -ended `failed` is not an error response: read `crawl.error.code` and -`crawl.error.detail` on it. Response enums (`CrawlStatus`, `StopReason`, +| Status | `code` | What to do | +|---|---|---| +| 400 | `invalid_request`, `unknown_parameter`, `invalid_cursor` | Fix the request. Do not retry it as is. | +| 403 | `REQS008` | Crawl is not enabled for the account (`exc.not_enabled`). | +| 404 | `crawl_not_found`, `content_not_found` | Check the id. | +| 409 | `idempotency_request_in_flight` | Retry after the first request with this key finishes. | +| 422 | `invalid_parameter`, `invalid_start_url`, `domain_not_allowed` | Fix the request. Do not retry it as is. | +| 422 | `idempotency_key_reused` | Use a new key, or no key. Do not retry it as is. | +| 429 | `too_many_crawls` | The account has reached its limit of active jobs (3 by default), shared with its Batch jobs. Retry after `exc.retry_after` seconds. `create` does not retry it. | + +`code` is None when the response has no problem body with a code (for +example a 502 from a proxy). A crawl that ended `failed` is not an error +response: read `crawl.error.code` and `crawl.error.detail` on it. Response enums (`CrawlStatus`, `StopReason`, `ContentStatus`, ...) are open: a value added later parses as an `UNKNOWN` member that keeps the raw value. diff --git a/examples/11_crawl.py b/examples/11_crawl.py index 1d3246e..fbeb982 100644 --- a/examples/11_crawl.py +++ b/examples/11_crawl.py @@ -1,13 +1,14 @@ -"""11: Crawl a site → wait → read URLs and pages. +"""11: Crawl a site → wait → read URLs and pages. Crawl is in Beta. Demonstrates the Crawl client (`ZenRowsCrawlClient`): - `create(url, depth=..., include_patterns=..., output_format="html")` starts a crawl and returns at once, `running`. - `wait(crawl_id)` blocks until it ends and returns the final `Crawl` (a `failed` crawl is returned, with `error` set). - - `iter_results(crawl_id)` pages through every kept URL. + - `iter_results(crawl_id)` pages through the kept URLs. - `get_content(crawl_id, result)` reads one kept URL's HTML. - - `download(crawl_id, path)` saves every result as NDJSON. + - `download(crawl_id, path)` saves every result as NDJSON and + returns the path and the crawl's status. Run with: export ZENROWS_API_KEY=zr_... @@ -26,7 +27,9 @@ def main() -> None: parser.add_argument("--url", required=True, help="the page to start from") parser.add_argument("--depth", type=int, default=1) parser.add_argument("--max-items", type=int, default=5) - parser.add_argument("--include", action="append", default=["/product/"]) + parser.add_argument( + "--include", action="append", help="keep URLs containing this (default /product/)" + ) parser.add_argument("--out", default="crawl.jsonl") args = parser.parse_args() @@ -37,7 +40,7 @@ def main() -> None: args.url, depth=args.depth, max_items=args.max_items, - include_patterns=args.include, + include_patterns=args.include or ["/product/"], output_format="html", ) except CrawlAPIError as exc: @@ -46,7 +49,7 @@ def main() -> None: raise print(f"started {crawl.crawl_id}") - crawl = client.wait(crawl.crawl_id, timeout=600.0) + crawl = client.wait(crawl.crawl_id) cov = crawl.coverage print( f"{crawl.crawl_id} {crawl.status.value}: {cov.items_found} URLs kept, " @@ -62,8 +65,8 @@ def main() -> None: html = client.get_content(crawl.crawl_id, result) print(f" {len(html)} chars of HTML") - path = client.download(crawl.crawl_id, args.out) - print(f"wrote {path}") + saved = client.download(crawl.crawl_id, args.out) + print(f"wrote {saved.path}") if __name__ == "__main__": diff --git a/src/zenrows/batch/_transport.py b/src/zenrows/batch/_transport.py index f6c95bf..1c3e2e5 100644 --- a/src/zenrows/batch/_transport.py +++ b/src/zenrows/batch/_transport.py @@ -117,17 +117,18 @@ def _send( params: dict[str, Any] | None, headers: dict[str, str] | None, content: bytes | None = None, + retry_statuses: frozenset[int] = _RETRYABLE_STATUSES, ) -> httpx.Response: """Issue the request, retrying transient failures on idempotent requests. - Retries `{429, 502, 503, 504}` and transient network errors up - to `retries` times, with jittered exponential backoff (honoring - `Retry-After` when present). Only idempotent requests are - replayed — `GET`/`PUT`/`DELETE`/`HEAD`/`OPTIONS`, plus `POST` - when the caller supplied an `Idempotency-Key`. Our own timeouts - (`httpx.TimeoutException`) are never retried: the caller set - that budget. + Retries `retry_statuses` (default `{429, 502, 503, 504}`) and + transient network errors up to `retries` times, with jittered + exponential backoff (honoring `Retry-After` when present). Only + idempotent requests are replayed — `GET`/`PUT`/`DELETE`/`HEAD`/ + `OPTIONS`, plus `POST` when the caller supplied an + `Idempotency-Key`. Our own timeouts (`httpx.TimeoutException`) + are never retried: the caller set that budget. """ idempotent = method.upper() in _IDEMPOTENT_METHODS or ( method.upper() == "POST" and _has_idempotency_key(headers) @@ -149,11 +150,7 @@ def _send( continue raise - if ( - idempotent - and attempt < self._retries - and response.status_code in _RETRYABLE_STATUSES - ): + if idempotent and attempt < self._retries and response.status_code in retry_statuses: wait_ms = _retry_after_ms(response) or _backoff_ms(attempt) response.close() time.sleep(wait_ms / 1000) @@ -216,6 +213,7 @@ def request( params: dict[str, Any] | None = None, headers: dict[str, str] | None = None, content: bytes | None = None, + retry_statuses: frozenset[int] = _RETRYABLE_STATUSES, ) -> httpx.Response: """Send a request (with retries) and return the raw response, raising the mapped error on non-2xx. For non-JSON bodies.""" @@ -226,6 +224,7 @@ def request( params=_drop_none(params), headers=headers, content=content, + retry_statuses=retry_statuses, ) elapsed_ms = (time.monotonic() - start) * 1000 self._raise_for_status(method, path, response) diff --git a/src/zenrows/batch/errors.py b/src/zenrows/batch/errors.py index 940a2be..40725b4 100644 --- a/src/zenrows/batch/errors.py +++ b/src/zenrows/batch/errors.py @@ -24,7 +24,7 @@ class ProblemDetail: type: str title: str status: int - code: str + code: str | None detail: str | None = None instance: str | None = None extras: dict[str, Any] | None = None @@ -48,7 +48,7 @@ def from_response(cls, response: httpx.Response) -> "ProblemDetail | None": type=body.get("type", "about:blank"), title=body.get("title", "Error"), status=int(body.get("status", response.status_code)), - code=body.get("code", "internal"), + code=body.get("code"), detail=body.get("detail"), instance=body.get("instance"), extras=extras or None, @@ -68,10 +68,10 @@ def __init__(self, status_code: int, problem: ProblemDetail | None, raw: bytes): self.status_code = status_code self.problem = problem self.raw = raw - self.code: str = problem.code if problem else "internal" + self.code: str = (problem.code if problem else None) or "internal" self.detail: str | None = problem.detail if problem else None msg = ( - f"{status_code} {problem.title}: {problem.detail or problem.code}" + f"{status_code} {problem.title}: {problem.detail or self.code}" if problem else f"{status_code} (no problem body)" ) diff --git a/src/zenrows/crawl/__init__.py b/src/zenrows/crawl/__init__.py index 2f2021e..8bfc28f 100644 --- a/src/zenrows/crawl/__init__.py +++ b/src/zenrows/crawl/__init__.py @@ -1,4 +1,6 @@ -"""Public surface for the ZenRows Crawl API. +"""Public surface for the ZenRows Crawl API (Beta). + +Crawl is in Beta: its surface may still change. What's where: - `client.ZenRowsCrawlClient` — the typed facade: create, get, list, @@ -11,7 +13,12 @@ """ from zenrows.batch._waiters import WaiterTimeout -from zenrows.crawl.client import DEFAULT_BASE_URL, ZenRowsCrawlClient +from zenrows.crawl.client import ( + DEFAULT_BASE_URL, + CrawlDownload, + CrawlDownloadFile, + ZenRowsCrawlClient, +) from zenrows.crawl.errors import CRAWL_NOT_ENABLED, CrawlAPIError from zenrows.crawl.models import ( ContentStatus, @@ -36,6 +43,8 @@ "Coverage", "Crawl", "CrawlAPIError", + "CrawlDownload", + "CrawlDownloadFile", "CrawlErrorCode", "CrawlList", "CrawlResult", diff --git a/src/zenrows/crawl/client.py b/src/zenrows/crawl/client.py index 30d6549..6fca9c0 100644 --- a/src/zenrows/crawl/client.py +++ b/src/zenrows/crawl/client.py @@ -1,4 +1,4 @@ -"""ZenRowsCrawlClient — the typed facade over the Crawl API. +"""ZenRowsCrawlClient — the typed facade over the Crawl API (Beta). A crawl starts from one URL, follows links up to `depth` hops, and keeps the URLs that match its patterns, up to `max_items`. It runs as @@ -18,16 +18,16 @@ import json import logging -import os -import time -from collections.abc import Iterator +from collections.abc import Callable, Iterator +from contextlib import ExitStack +from dataclasses import dataclass from pathlib import Path from typing import Any, Literal import httpx from zenrows.__version__ import __version__ -from zenrows.batch._transport import _Transport +from zenrows.batch._transport import _RETRYABLE_STATUSES, _Transport from zenrows.batch._waiters import poll_until from zenrows.crawl.errors import CrawlAPIError from zenrows.crawl.models import ( @@ -40,16 +40,59 @@ DownloadLine, ) -# Production endpoint. The `base_url=` kwarg + `ZENROWS_CRAWL_BASE_URL` -# env var are present for advanced use only. DEFAULT_BASE_URL = "https://api.zenrows.com/v1" DEFAULT_USER_AGENT = f"zenrows-crawl-python/{__version__}" _log = logging.getLogger("zenrows.crawl.transport") +# 429 `too_many_crawls` is a capacity limit, not a transient failure. +_CREATE_RETRY_STATUSES = _RETRYABLE_STATUSES - {429} + + +class CrawlDownload: + """A crawl's NDJSON download, open and ready to read. + + `status` is the crawl's status when the file was read + (`X-Crawl-Status`): `running` means the file holds only what the + crawl has kept so far. Iterate it once for the parsed lines; the + connection closes when the iteration ends. Use it as a context + manager to close it without reading to the end. + """ + + def __init__(self, response: httpx.Response, close: Callable[[], None]): + self.status: CrawlStatus | None = _crawl_status(response) + self._response = response + self._close = close + + def __iter__(self) -> Iterator[DownloadLine]: + try: + for line in self._response.iter_lines(): + if line.strip(): + yield DownloadLine.model_validate_json(line) + finally: + self.close() + + def close(self) -> None: + self._close() + + def __enter__(self) -> "CrawlDownload": + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + +@dataclass(slots=True, frozen=True) +class CrawlDownloadFile: + """A crawl's NDJSON download, saved to `path`. `status` is as on + `CrawlDownload`.""" + + path: Path + status: CrawlStatus | None + class ZenRowsCrawlClient: - """Synchronous, typed client for the ZenRows Crawl API. + """Synchronous, typed client for the ZenRows Crawl API (Beta). `api_key` is required: @@ -71,16 +114,16 @@ def __init__( user_agent: str = DEFAULT_USER_AGENT, httpx_args: dict[str, Any] | None = None, ): - """`retries` bounds automatic retries of transient failures - (HTTP 429/502/503/504 and network errors) on idempotent - requests — GETs, and a `create` that carries an - `idempotency_key`. Retries use jittered exponential backoff and - honor `Retry-After`; set `retries=0` to disable.""" + """`base_url` defaults to `DEFAULT_BASE_URL`. `retries` bounds + automatic retries of transient failures (HTTP 429/502/503/504 + and network errors) on idempotent requests — GETs, and a + `create` that carries an `idempotency_key` (never on 429 for + `create`). Retries use jittered exponential backoff and honor + `Retry-After`; set `retries=0` to disable.""" if not api_key: raise ValueError("ZenRowsCrawlClient: api_key is required.") - base_url = base_url or os.environ.get("ZENROWS_CRAWL_BASE_URL") or DEFAULT_BASE_URL self._t = _Transport( - base_url=base_url, + base_url=base_url or DEFAULT_BASE_URL, api_key=api_key, user_agent=user_agent, timeout=timeout, @@ -129,15 +172,17 @@ def create( bounds the cost (server default 10). - `include_patterns` / `exclude_patterns`: substrings matched against each normalized URL. A URL is kept when it matches - an include pattern (if any) and no exclude pattern. + an include pattern (if any) and no exclude pattern. A crawl + stays on the start URL's registrable domain; subdomains count. - `output_format="html"`: also fetch every kept URL's page, readable with `get_content` / `download`. Omit for URLs only. - `idempotency_key`: a retry with the same key and body answers with the crawl the first request created. - Only the arguments you pass are sent. An account with too many - crawls running gets 429 `too_many_crawls`; retry after the - error's `retry_after` seconds. + Only the arguments you pass are sent. When the account has + reached its limit of active jobs (3 by default), shared with its + Batch jobs, `create` raises 429 `too_many_crawls` without + retrying; retry after the error's `retry_after` seconds. """ if output_format not in (None, "html"): raise ValueError(f"create: output_format must be 'html' or None, not {output_format!r}") @@ -154,7 +199,11 @@ def create( if idempotency_key: headers["Idempotency-Key"] = idempotency_key response = self._t.request( - "POST", "/crawls", headers=headers, content=json.dumps(body).encode("utf-8") + "POST", + "/crawls", + headers=headers, + content=json.dumps(body).encode("utf-8"), + retry_statuses=_CREATE_RETRY_STATUSES, ) return Crawl.model_validate(response.json()) @@ -203,25 +252,20 @@ def iter_results( crawl_id: str, *, page_size: int | None = None, - poll_interval: float = 2.0, ) -> Iterator[CrawlResult]: - """Yield every URL the crawl keeps, in order, each once. - - Follows `next_cursor` until the server returns null, which it - does only once the crawl has ended and its last page was read. - On a crawl that is still running, the iterator follows it live: - when it has caught up, it sleeps `poll_interval` seconds and - asks again with the same cursor, so it ends when the crawl ends. - Call `wait` first to read a finished crawl without polling. + """Yield the URLs the crawl has kept, in order, each once. + + Follows `next_cursor` and returns at the first empty page or + null cursor. It does not poll: on a running crawl it yields + only what the crawl has kept so far. Call `wait` first to read + every result. """ cursor: str | None = None while True: page = self.get(crawl_id, cursor=cursor, limit=page_size) yield from page.results - if page.next_cursor is None: + if not page.results or page.next_cursor is None: return - if not page.results and page.status is CrawlStatus.RUNNING: - time.sleep(poll_interval) cursor = page.next_cursor def stop(self, crawl_id: str) -> CrawlStop: @@ -255,18 +299,18 @@ def get_content(self, crawl_id: str, content: str | CrawlResult) -> str: content_id = content return self._t.request("GET", f"/crawls/{crawl_id}/contents/{content_id}").text - def iter_download(self, crawl_id: str) -> Iterator[DownloadLine]: - """`GET /crawls/{crawl_id}/download` — stream every result as - a parsed `DownloadLine` (url, content_status, content). + def iter_download(self, crawl_id: str) -> CrawlDownload: + """`GET /crawls/{crawl_id}/download` — open the NDJSON file and + return it as a `CrawlDownload`: iterate it for parsed + `DownloadLine`s (url, content_status, content), and read its + `status`. On a running crawl the file holds what the crawl has kept so - far (a warning is logged); call `wait` first for the whole of it. + far, and `status` is `running`; call `wait` first for all of it. """ - with self._t.stream("GET", f"/crawls/{crawl_id}/download") as response: - _warn_if_partial(crawl_id, response) - for line in response.iter_lines(): - if line.strip(): - yield DownloadLine.model_validate_json(line) + stack = ExitStack() + response = stack.enter_context(self._t.stream("GET", f"/crawls/{crawl_id}/download")) + return CrawlDownload(response, stack.close) def download( self, @@ -274,21 +318,23 @@ def download( target_path: str | Path, *, chunk_size: int = 64 * 1024, - ) -> Path: + ) -> CrawlDownloadFile: """`GET /crawls/{crawl_id}/download` — stream the NDJSON file - (one JSON object per result) to `target_path`, and return it. + (one JSON object per result) to `target_path`. Returns its + `path` and the crawl's `status` when the file was read. On a running crawl the file holds what the crawl has kept so - far (a warning is logged); call `wait` first for the whole of it. + far, and `status` is `running`; call `wait` first for all of it. """ target = Path(target_path) target.parent.mkdir(parents=True, exist_ok=True) - with self._t.stream("GET", f"/crawls/{crawl_id}/download") as response: - _warn_if_partial(crawl_id, response) - with target.open("wb") as f: - for chunk in response.iter_bytes(chunk_size): - f.write(chunk) - return target + with ( + self._t.stream("GET", f"/crawls/{crawl_id}/download") as response, + target.open("wb") as f, + ): + for chunk in response.iter_bytes(chunk_size): + f.write(chunk) + return CrawlDownloadFile(path=target, status=_crawl_status(response)) # ===== waiter ===== @@ -296,7 +342,7 @@ def wait( self, crawl_id: str, *, - timeout: float = 300.0, + timeout: float = 600.0, poll_interval: float = 2.0, max_poll_interval: float = 15.0, ) -> Crawl: @@ -323,13 +369,15 @@ def fetch() -> Crawl: ) -def _warn_if_partial(crawl_id: str, response: httpx.Response) -> None: +def _crawl_status(response: httpx.Response) -> CrawlStatus | None: status = response.headers.get("X-Crawl-Status") - if status == CrawlStatus.RUNNING.value: - logging.getLogger("zenrows.crawl.client").warning( - "crawl %s is still running: the download holds only the results kept so far", - crawl_id, - ) + return CrawlStatus(status) if status else None -__all__ = ["DEFAULT_BASE_URL", "DEFAULT_USER_AGENT", "ZenRowsCrawlClient"] +__all__ = [ + "DEFAULT_BASE_URL", + "DEFAULT_USER_AGENT", + "CrawlDownload", + "CrawlDownloadFile", + "ZenRowsCrawlClient", +] diff --git a/src/zenrows/crawl/errors.py b/src/zenrows/crawl/errors.py index b1f5b9d..d81f67b 100644 --- a/src/zenrows/crawl/errors.py +++ b/src/zenrows/crawl/errors.py @@ -20,8 +20,8 @@ class CrawlAPIError(Exception): `code` is the problem's `code` member: `crawl_not_found`, `content_not_found`, `invalid_parameter`, `too_many_crawls`, ..., `REQS008` when Crawl is not enabled for the account, and the API's - usual auth and credit codes. `problem` - holds the whole body, or None when it was not JSON. + usual auth and credit codes; None when the body has no code. + `problem` holds the whole body, or None when it was not JSON. """ def __init__( @@ -38,14 +38,15 @@ def __init__( self.retry_after = retry_after self.problem = problem self.raw = raw - self.code: str = problem.code if problem else "internal" + self.code: str | None = problem.code if problem else None self.detail: str | None = problem.detail if problem else None if problem and self.code == CRAWL_NOT_ENABLED: msg = f"{status_code} Crawl is not enabled for this account ({CRAWL_NOT_ENABLED})" + ( f": {problem.detail}" if problem.detail else "" ) elif problem: - msg = f"{status_code} {problem.title}: {problem.detail or problem.code}" + reason = problem.detail or problem.code + msg = f"{status_code} {problem.title}" + (f": {reason}" if reason else "") else: msg = f"{status_code} (no problem body)" super().__init__(msg) diff --git a/src/zenrows/crawl/models.py b/src/zenrows/crawl/models.py index e3249e0..d6a07e2 100644 --- a/src/zenrows/crawl/models.py +++ b/src/zenrows/crawl/models.py @@ -10,6 +10,7 @@ from datetime import datetime from enum import Enum +from typing import Any from pydantic import BaseModel @@ -172,8 +173,9 @@ class DownloadLine(BaseModel): url: str content_status: ContentStatus | None = None - content: str | None = None - """The page HTML, present when `content_status` is `fetched`.""" + content: str | dict[str, Any] | None = None + """The page, present when `content_status` is `fetched`: HTML text, + or an object for a crawl whose output format returns parsed data.""" __all__ = [ diff --git a/src/zenrows/py.typed b/src/zenrows/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/tests/e2e/test_crawl_e2e.py b/tests/e2e/test_crawl_e2e.py index d37db0d..8357507 100644 --- a/tests/e2e/test_crawl_e2e.py +++ b/tests/e2e/test_crawl_e2e.py @@ -30,14 +30,16 @@ START_URL = os.environ.get("ZENROWS_E2E_CRAWL_URL", "") INCLUDE = os.environ.get("ZENROWS_E2E_CRAWL_INCLUDE") or None -# An account can run only a few crawls at once, so a create can meet -# 429 too_many_crawls while other runs finish. +# An account has a limit of active jobs (3 by default, shared with its +# Batch jobs), so a create can meet 429 too_many_crawls while others finish. SLOT_WAIT_SECONDS = 300 @pytest.fixture(scope="module") def client(): - with ZenRowsCrawlClient(api_key=os.environ["ZENROWS_API_KEY"]) as c: + with ZenRowsCrawlClient( + api_key=os.environ["ZENROWS_API_KEY"], base_url=os.environ["ZENROWS_CRAWL_BASE_URL"] + ) as c: yield c @@ -86,11 +88,14 @@ def test_crawl_end_to_end(client: ZenRowsCrawlClient, tmp_path): print(f"content of {fetched[0].url}: {len(html)} chars") assert " ZenRowsCrawlClient: @pytest.fixture def no_sleep(monkeypatch): - """Record + swallow sleeps (waiter, scanner and retry backoff all - call `time.sleep`) so tests run instantly.""" + """Record + swallow sleeps (waiter and retry backoff both call + `time.sleep`) so tests run instantly.""" slept: list[float] = [] - monkeypatch.setattr("zenrows.crawl.client.time.sleep", slept.append) + monkeypatch.setattr("time.sleep", slept.append) return slept @@ -81,14 +82,9 @@ def test_missing_api_key_raises(): ZenRowsCrawlClient(api_key="") -def test_default_base_url(monkeypatch): - monkeypatch.delenv("ZENROWS_CRAWL_BASE_URL", raising=False) - assert ZenRowsCrawlClient(api_key=API_KEY).base_url.rstrip("/") == "https://api.zenrows.com/v1" - - -def test_base_url_env_and_kwarg_precedence(monkeypatch): +def test_base_url_is_a_constructor_option_only(monkeypatch): monkeypatch.setenv("ZENROWS_CRAWL_BASE_URL", "http://env-set/v1") - assert ZenRowsCrawlClient(api_key=API_KEY).base_url.rstrip("/") == "http://env-set/v1" + assert ZenRowsCrawlClient(api_key=API_KEY).base_url.rstrip("/") == "https://api.zenrows.com/v1" assert ZenRowsCrawlClient(api_key=API_KEY, base_url=BASE_URL).base_url.rstrip("/") == BASE_URL @@ -262,12 +258,10 @@ def test_stop_posts_without_body(client: ZenRowsCrawlClient): @respx.mock -def test_iter_results_stops_on_null_cursor(client: ZenRowsCrawlClient, no_sleep): +def test_iter_results_stops_on_null_cursor(client: ZenRowsCrawlClient): route = respx.get(f"{BASE_URL}/crawls/c_1").mock( side_effect=[ Response(200, json=crawl_body(results=[{"url": "https://a"}], next_cursor="k1")), - # Caught up with a running crawl: no results, same cursor back. - Response(200, json=crawl_body(results=[], next_cursor="k1")), Response( 200, json=crawl_body( @@ -277,14 +271,31 @@ def test_iter_results_stops_on_null_cursor(client: ZenRowsCrawlClient, no_sleep) ] ) - urls = [r.url for r in client.iter_results("c_1", poll_interval=0.5)] + urls = [r.url for r in client.iter_results("c_1")] assert urls == ["https://a", "https://b"] - assert route.call_count == 3 + assert route.call_count == 2 assert "cursor" not in route.calls[0].request.url.params assert route.calls[1].request.url.params["cursor"] == "k1" - assert route.calls[2].request.url.params["cursor"] == "k1" - assert no_sleep == [0.5] # slept only on the empty page of a running crawl + + +@respx.mock +def test_iter_results_returns_at_first_empty_page_of_running_crawl( + client: ZenRowsCrawlClient, no_sleep +): + route = respx.get(f"{BASE_URL}/crawls/c_1").mock( + side_effect=[ + Response(200, json=crawl_body(results=[{"url": "https://a"}], next_cursor="k1")), + # Caught up with a running crawl: no results, same cursor back. + Response(200, json=crawl_body(results=[], next_cursor="k1")), + ] + ) + + urls = [r.url for r in client.iter_results("c_1")] + + assert urls == ["https://a"] + assert route.call_count == 2 + assert no_sleep == [] # ----- contents / download ----- @@ -314,11 +325,12 @@ def test_get_content_without_content_url_raises(client: ZenRowsCrawlClient): NDJSON = ( b'{"url":"https://a","content_status":"fetched","content":"a"}\n' b'{"url":"https://b","content_status":"failed"}\n' + b'{"url":"https://c","content_status":"fetched","content":{"title":"C"}}\n' ) @respx.mock -def test_iter_download_parses_lines(client: ZenRowsCrawlClient): +def test_iter_download_parses_lines_and_status(client: ZenRowsCrawlClient): respx.get(f"{BASE_URL}/crawls/c_1/download").mock( return_value=Response( 200, @@ -326,27 +338,39 @@ def test_iter_download_parses_lines(client: ZenRowsCrawlClient): content=NDJSON, ) ) - lines = list(client.iter_download("c_1")) - assert [line.url for line in lines] == ["https://a", "https://b"] + download = client.iter_download("c_1") + assert download.status is CrawlStatus.COMPLETED + lines = list(download) + assert [line.url for line in lines] == ["https://a", "https://b", "https://c"] assert lines[0].content == "a" assert lines[1].content_status is ContentStatus.FAILED + assert lines[2].content == {"title": "C"} + + +@respx.mock +def test_iter_download_without_status_header(client: ZenRowsCrawlClient): + respx.get(f"{BASE_URL}/crawls/c_1/download").mock(return_value=Response(200, content=b"")) + with client.iter_download("c_1") as download: + assert download.status is None + assert list(download) == [] @respx.mock -def test_download_writes_file_and_warns_when_partial(client: ZenRowsCrawlClient, tmp_path, caplog): +def test_download_writes_file_and_returns_status(client: ZenRowsCrawlClient, tmp_path): respx.get(f"{BASE_URL}/crawls/c_1/download").mock( return_value=Response(200, headers={"X-Crawl-Status": "running"}, content=NDJSON) ) - target = client.download("c_1", tmp_path / "out" / "c_1.jsonl") - assert target.read_bytes() == NDJSON - assert "still running" in caplog.text + target = tmp_path / "out" / "c_1.jsonl" + saved = client.download("c_1", target) + assert saved == CrawlDownloadFile(path=target, status=CrawlStatus.RUNNING) + assert saved.path.read_bytes() == NDJSON @respx.mock def test_download_maps_errors(client: ZenRowsCrawlClient): respx.get(f"{BASE_URL}/crawls/nope/download").mock(return_value=problem(404, "crawl_not_found")) with pytest.raises(CrawlAPIError) as exc_info: - list(client.iter_download("nope")) + client.iter_download("nope") assert exc_info.value.code == "crawl_not_found" @@ -403,6 +427,43 @@ def test_crawl_not_enabled_has_clear_message(client: ZenRowsCrawlClient): assert "Crawl is not enabled for this account" in str(err) +@pytest.mark.parametrize( + "response", + [Response(502, text="bad gateway"), Response(500, json={"title": "Server error"})], +) +@respx.mock +def test_error_without_code_has_none_code(client: ZenRowsCrawlClient, response): + respx.get(f"{BASE_URL}/crawls/c_1").mock(return_value=response) + with pytest.raises(CrawlAPIError) as exc_info: + client.get("c_1") + assert exc_info.value.status_code == response.status_code + assert exc_info.value.code is None + + +@respx.mock +def test_keyed_create_does_not_retry_too_many_crawls(no_sleep): + client = ZenRowsCrawlClient(api_key=API_KEY, base_url=BASE_URL, retries=2) + route = respx.post(f"{BASE_URL}/crawls").mock( + return_value=problem(429, "too_many_crawls", headers={"Retry-After": "1"}) + ) + with pytest.raises(CrawlAPIError) as exc_info: + client.create("https://example.com/", depth=1, idempotency_key="idem-1") + assert exc_info.value.code == "too_many_crawls" + assert route.call_count == 1 + assert no_sleep == [] + + +@respx.mock +def test_keyed_create_retries_transient_status(no_sleep): + client = ZenRowsCrawlClient(api_key=API_KEY, base_url=BASE_URL, retries=2) + route = respx.post(f"{BASE_URL}/crawls").mock( + side_effect=[Response(503), Response(202, json=crawl_body())] + ) + client.create("https://example.com/", depth=1, idempotency_key="idem-1") + assert route.call_count == 2 + assert len(no_sleep) == 1 + + @respx.mock def test_get_retries_transient_status(no_sleep): client = ZenRowsCrawlClient(api_key=API_KEY, base_url=BASE_URL, retries=2)