diff --git a/README.md b/README.md index 68a15b1..58c6cf8 100644 --- a/README.md +++ b/README.md @@ -245,6 +245,12 @@ count = run.download_to_dir("./out") print(f"downloaded {count} results to ./out/") ``` +`run.wait()` returns once the run is terminal (`completed`, `stopped` or +`failed`; read `failure_reason` on a failed run). A paused run never +progresses, so the waiter raises `RunPaused` (carrying the run) instead +of polling until its timeout; pass `wait_while_paused=True` to keep +polling. + > Need full control over the request body? `client.submit_job({...})` accepts a raw dict (the > wire shape) or a typed `SubmitJobRequest`, and returns the same `JobRef`. @@ -268,7 +274,8 @@ job = client.submit_regular( An Extract task's result carries two keys — `html` (the raw page) and `parsed` (the structured data). It costs the same as a regular task, so `estimate_cost` -prices it correctly. `extract_fields` is not supported in Batch yet. +prices it correctly. `extract_fields` narrows the returned fields, as on +the regular API. ### Upload URLs from a CSV @@ -310,11 +317,11 @@ print(est.format()) # per-tier breakdown table > it lives on the client (`client.estimate_cost`) to keep your call > sites stable. -After a run finishes, read the **actual** credits it used from -`run.stats.spend` (`{credits, cost}`). Only successful requests are -charged, so realized spend is ≤ the estimate. (Per-task `task.spend` is -indicative and often absent on successful rows — rely on the run-level -rollup for actuals.) +After a run finishes, read the credits it used from `run.stats.spend` +(`{credits, cost}`), an indicative figure; your account's usage page is +authoritative. Failed tasks are not charged, except target `404` and +`410` responses, which are billed as on the regular API. Result rows +carry no per-task spend. ### Retry only the failed tasks @@ -335,6 +342,11 @@ run = client.job(job_id).retry_failed() run.wait() ``` +Start a rerun only once the previous run is terminal, or the API answers +`409 run_not_terminal`. A failed-only (partial) rerun works within 10 +days of the start of the run it retries; after that it raises +`409 partial_rerun_window_expired`, and a full `rerun()` still works. + ### Download results The Batch API doesn't ship page bodies inline — `results()` gives you @@ -382,9 +394,24 @@ artifact: run.download_all_results("results.zip") ``` -> The server-side zip is capped at **1 GiB** per run. For larger runs — -> or one file per task — use `download_to_dir`, which fetches bodies -> client-side with **no size limit** (slower; tune with `concurrency=`). +> The server caps one export at **100 GiB** of results. For larger runs, +> or one file per task, use `download_to_dir`, which fetches bodies +> client-side with no export cap (slower; tune with `concurrency=`). + +**Expired results.** Results are kept for 14 days by default. After that +a successful task reads as `expired`, has no `result_url`, and still +matches `status="successful"`. The bulk helpers skip such rows (and +failed ones) instead of stopping, log a WARNING on +`zenrows.batch.download` with the counts, and pass each one to +`on_skip` if you set it: + +```python +from zenrows.batch import TaskStatus + +skipped = [] +run.download_to_dir("./out", on_skip=skipped.append) +expired = [r for r in skipped if r.status is TaskStatus.EXPIRED] +``` **One task at a time.** When you're iterating `results()` and want only some bodies (a Python-side filter the `status=` argument can't express), @@ -399,8 +426,9 @@ for task in run.results(status="successful"): `download_task_to_file` also accepts an open binary file object, not just a path. Every result's `result_url` is a plain presigned URL, so -you can equally GET it with your own HTTP client (no auth header) — it's -TTL-limited, so fetch it promptly. See examples 02 and 08. +you can equally GET it with your own HTTP client (no auth header). It is +valid for 2 hours; list the results again for a fresh one. See examples +02 and 08. ### Act on an id without a GET @@ -435,19 +463,47 @@ job.schedule.resume() # fire again job.schedule.update(Rate(every=1, unit="day")) # replace the schedule ``` -Register a completion webhook at submit time, or manage it later via -`PUT` / `DELETE /jobs/{id}/webhook`. Set `signature=True` to have each -delivery HMAC-signed (manage signing keys via the `/hmac/keys` -endpoints); it defaults to unsigned. Deliveries are at-least-once, so -dedup on the `X-ZenRows-Event-Id` header. +Register a webhook at submit time, or manage it later with +`job.set_webhook(url, signature=...)` / `job.delete_webhook()`. You get +one event per run: `run.completed` when it finishes on its own, or +`run.failed` when an account error (credits, subscription, key cap) +ends it. A run you stop yourself sends none. A failed delivery is +retried with backoff for up to 3 days, so an event can arrive more than +once: deduplicate on the `X-ZenRows-Event-Id` header. + +Set `signature=True` to have each delivery HMAC-signed; it defaults to +unsigned. Create the signing key first with `client.rotate_hmac_key()` +(the first key is active at once; a later rotation stages a new key you +promote with `finalize_hmac_key()`), and store the `secret` it returns: +it is shown once, base64-encoded. `verify_webhook_signature` checks the +`X-Signature` header (`t=...,v1=...,kid=...`): it decodes the secret, +recomputes HMAC-SHA256 over `.`, compares in constant time +and rejects timestamps more than 5 minutes off. ```python +from zenrows.batch import WEBHOOK_SIGNATURE_HEADER, WebhookEvent, verify_webhook_signature + client.submit_regular( ["https://example.com/a"], webhook={"url": "https://hooks.example.com/zr", "signature": True}, ) + +# kid -> the base64 secret rotate_hmac_key() returned +SECRETS = {"": ""} + + +def receive(raw_body: bytes, headers) -> int: + # Verify the raw bytes before parsing them. + if not verify_webhook_signature(raw_body, headers[WEBHOOK_SIGNATURE_HEADER], SECRETS): + return 401 + event = WebhookEvent.model_validate_json(raw_body) + print(event.event_type, event.job_id, event.run_id) + return 200 ``` +During a key rotation, pass both secrets keyed by their `kid`; each +delivery names the key that signed it. + > No webhook receiver? Just poll for completion with `job.run.wait()` / > `run.wait()`. @@ -457,18 +513,37 @@ Every non-2xx surfaces as `BatchAPIError`; the `code` attribute carries the stable code from the RFC 7807 body (`file_input_not_found`, `idempotency_key_conflict`, `api_key_cap_reached`, etc.) and `detail` the human-readable explanation (for a credit cap: which cap, and when it resets). +`BatchErrorCode` names every code; its members compare equal to `code`. ```python -from zenrows.batch import BatchAPIError +from zenrows.batch import BatchAPIError, BatchErrorCode try: client.get_job("does-not-exist") except BatchAPIError as exc: - if exc.code == "not_found": + if exc.code == BatchErrorCode.NOT_FOUND: ... raise ``` +The client retries network errors and `429` / `502` / `503` / `504` on +requests that are safe to repeat: reads, and a submit, rerun or +`add_tasks` call that carries an `idempotency_key`. Within 24 hours a +repeat with the same key and body returns the first answer instead of +doing the work twice. A `429 quota_exceeded` (3 active jobs per account, +shared with Crawl, or 5 scheduled jobs) is not retried: it clears only +when a job finishes, is closed or is deleted. + +```python +import uuid + +job = client.submit_open(["https://example.com/a"]) +job.add_tasks( + {"tasks": [{"url": "https://example.com/b"}], "last_batch": True}, + idempotency_key=str(uuid.uuid4()), # one key per logical call, reused on retry +) +``` + ### Forward-compatible enums Every enum the Batch API returns (`RunStatus`, `FailureReason`, ...) is open: diff --git a/docs/batch-client-reference.md b/docs/batch-client-reference.md index 4f4ced6..413025a 100644 --- a/docs/batch-client-reference.md +++ b/docs/batch-client-reference.md @@ -179,6 +179,10 @@ def submit_scheduled(schedule: Schedule | JobScheduleDict, Submit a scheduled job. +Creating a scheduled job starts no run (the first fire creates +Run 1), so there is nothing to ingest and no `wait_for_ingest` +option. + `schedule` is one of the typed builders (`At`, `Rate`, `Calendar`) or a raw `JobScheduleDict` for power users. The typed builders validate their inputs in `__post_init__`; @@ -404,7 +408,8 @@ def wait_for_run(job_id: str, timeout: float = 300.0, poll_interval: float = 2.0, max_poll_interval: float = 15.0, - progress: bool = False) -> Run + progress: bool = False, + wait_while_paused: bool = False) -> Run ``` Block until a run reaches one of `target_statuses`, polling @@ -416,9 +421,14 @@ included, so a run the API auto-failed is returned (read until `timeout`. Pass `failure_statuses={"failed"}` to have the waiter raise `WaiterError` on it instead. +A paused run (`pause_state: paused`) never progresses until +someone resumes it, so the waiter raises `RunPaused` (a +`WaiterError`, carrying the run) as soon as it sees one that is +not yet in a target status. Pass `wait_while_paused=True` to +keep polling instead, for example when another process will +resume it. + `progress=True` shows a tqdm bar with totals as they advance. -`None` inherits the client-level `progress` setting (which -itself defaults to off unless `ZENROWS_BATCH_PROGRESS=true`). @@ -526,9 +536,9 @@ carrying the just-issued `export_id`; call `.load()` or block on `.wait()` for the download URL. Failure modes: 404 if the job/run is missing or not yours. -The worker fails the export (status=failed, -error="results are larger then 1 gb") if the pre-zip total -exceeds the server's 1 GiB cap. +The server fails the export (status=failed, +error="results are larger than N gb") if the run's results +total more than the export cap (100 GiB). @@ -564,7 +574,7 @@ Block until an export reaches a terminal state. Defaults to check `.status` and (on `completed`) `.download_url`. `failed` is NOT raised; the caller decides whether -`error="results are larger then 1 gb"` is fatal or expected. +`error="results are larger than N gb"` is fatal or expected. @@ -591,9 +601,9 @@ Steps: Raises `WaiterTimeout` if the export doesn't reach a terminal state within `wait_timeout`. Raises `BatchAPIError` with the server's error message on `status=failed` (e.g. -`"results are larger then 1 gb"`). Returns the `Path` written. +`"results are larger than N gb"`). Returns the `Path` written. -The server-side export is capped at 1 GiB per run. For larger +The server caps one export at 100 GiB of results. For larger runs (or one file per task), use `download_to_dir` instead: it fetches each body client-side with no size limit, at the cost of being slower (one body at a time, tunable via `concurrency=`). @@ -668,6 +678,50 @@ def delete() -> None ``DELETE /jobs/{id}/runs/{run_id}`` — scrub one run only. + + +#### download\_to\_dir + +```python +def download_to_dir( + target_dir: str | Path, + *, + status: str | None = "successful", + name_fn: Callable[[TaskResult], str] | None = None, + use_external_id: bool = False, + concurrency: int = 1, + progress: bool = False, + max_files: int | None = None, + max_bytes_per_file: int | None = None, + on_skip: Callable[[TaskResult], None] | None = None) -> int +``` + +Write every matching task's body to ``target_dir``, one file +per task. Returns the number of files written. Failed and +``expired`` rows have no body: they are skipped and passed to +``on_skip`` when you set it. + + + +#### download\_to\_memory + +```python +def download_to_memory( + *, + status: str | None = "successful", + concurrency: int = 1, + progress: bool = False, + max_count: int | None = None, + max_total_bytes: int | None = None, + max_bytes_per_file: int | None = None, + on_skip: Callable[[TaskResult], None] | None = None +) -> "list[DownloadedResult]" +``` + +Load every matching task's body into memory. Failed and +``expired`` rows have no body: they are skipped and passed to +``on_skip`` when you set it. + #### download\_task\_to\_file @@ -702,12 +756,15 @@ def wait(*, failure_statuses: set[str] | frozenset[str] | None = None, timeout: float = 300.0, poll_interval: float = 2.0, - progress: bool = False) -> "RunHandle" + progress: bool = False, + wait_while_paused: bool = False) -> "RunHandle" ``` Block until this run reaches a target state. Returns a fresh loaded :class:`RunHandle` so chains like ``run.wait().download_to_dir(...)`` work without re-wrapping. +A paused run raises :class:`RunPaused` unless +``wait_while_paused=True``. @@ -748,9 +805,9 @@ Start an export of this run's results, wait for it, and save the zip to ``target_path``. See ``ZenRowsBatchClient.download_all_results`` for the contract. -Capped at 1 GiB per run server-side; for larger runs use -``download_to_dir`` (no size limit, but slower — one body at a -time, tunable via ``concurrency=``). +The server caps one export at 100 GiB of results. For a larger +run use ``download_to_dir``: it has no export cap, but is slower +(one body at a time, tunable via ``concurrency=``). @@ -798,7 +855,8 @@ A reference to a results-export by id. A results export is async: ``start_export()`` returns immediately with a ``pending`` ref; the server zips the run in the background and it reaches ``completed`` (download URL ready) or ``failed`` -(with an ``error`` message — e.g. the 1 GiB size cap). Call +(with an ``error`` message, e.g. the run is over the 100 GiB +export cap). Call ``load()`` / ``wait()`` for an :class:`ExportHandle` with data. @@ -946,14 +1004,17 @@ def wait(*, failure_statuses: set[str] | frozenset[str] | None = None, timeout: float = 300.0, poll_interval: float = 2.0, - progress: bool = False) -> "RunHandle" + progress: bool = False, + wait_while_paused: bool = False) -> "RunHandle" ``` Block until the current run reaches a target state. Returns a loaded :class:`RunHandle`, so chains like ``job.run.wait().download_to_dir(...)`` work cleanly. -See ``ZenRowsBatchClient.wait_for_run`` for the full contract. +A paused run raises :class:`RunPaused` instead of being polled +until ``timeout``, unless ``wait_while_paused=True``. See +``ZenRowsBatchClient.wait_for_run`` for the full contract. @@ -975,6 +1036,50 @@ def task_history(task_id: str) -> TaskHistoryResponse Current-run per-attempt event log for one task. + + +#### download\_to\_dir + +```python +def download_to_dir( + target_dir: str | Path, + *, + status: str | None = "successful", + name_fn: Callable[[TaskResult], str] | None = None, + use_external_id: bool = False, + concurrency: int = 1, + progress: bool = False, + max_files: int | None = None, + max_bytes_per_file: int | None = None, + on_skip: Callable[[TaskResult], None] | None = None) -> int +``` + +Write every matching task's body to ``target_dir``, one file +per task. Returns the number of files written. Failed and +``expired`` rows have no body: they are skipped and passed to +``on_skip`` when you set it. + + + +#### download\_to\_memory + +```python +def download_to_memory( + *, + status: str | None = "successful", + concurrency: int = 1, + progress: bool = False, + max_count: int | None = None, + max_total_bytes: int | None = None, + max_bytes_per_file: int | None = None, + on_skip: Callable[[TaskResult], None] | None = None +) -> "list[DownloadedResult]" +``` + +Load every matching task's body into memory. Failed and +``expired`` rows have no body: they are skipped and passed to +``on_skip`` when you set it. + #### download\_task\_to\_file @@ -1028,9 +1133,9 @@ Start an export of the current run's results, wait for it, and save the zip to ``target_path``. See ``ZenRowsBatchClient.download_all_results`` for the contract. -Capped at 1 GiB per run server-side; for larger runs use -``download_to_dir`` (no size limit, but slower — one body at a -time, tunable via ``concurrency=``). +The server caps one export at 100 GiB of results. For a larger +run use ``download_to_dir``: it has no export cap, but is slower +(one body at a time, tunable via ``concurrency=``). @@ -1188,7 +1293,8 @@ def delete() -> None ```python def rerun(*, status: str | list[str] | None = None, - idempotency_key: str | None = None) -> "RunHandle" + idempotency_key: str | None = None, + wait_for_ingest: bool = False) -> "RunHandle" ``` ``POST /jobs/{id}/rerun[?status=...]`` — start a new run. @@ -1200,7 +1306,22 @@ from the template). No source-child link. With ``status`` (single value like ``"failed"`` or list like ``["failed", "pending"]``): partial retry — matching statuses are reset to ``pending`` and re-enqueued; everything else is -inherited verbatim with ``source_run_id`` stamped. +inherited verbatim with ``source_run_id`` stamped. A partial +retry works within 10 days of the start of the run it retries; +after that the API answers 409 ``partial_rerun_window_expired`` +and only a full rerun works. + +Every rerun is accepted asynchronously (``202``): the new run's +task rows keep arriving until ``ingest_status`` is ``done``, so +``results()`` read straight away can be partial. Pass +``wait_for_ingest=True`` to poll the run this call returned +(``GET /jobs/{id}/runs/{run_id}``, for up to 300 s) until its +``ingest_status`` is no longer ``pending``, and get that fresh +snapshot back. This also holds when the call replays an earlier +answer for the same ``idempotency_key``. A run stopped while it +was ingesting comes back ``stopped`` once ingestion has wound +down, with only the rows written so far (``stats.total`` is cut +to match). Otherwise use ``run.wait()`` before reading results. Returns a :class:`RunHandle` for the newly-created run. @@ -1211,7 +1332,8 @@ Returns a :class:`RunHandle` for the newly-created run. ```python def retry_failed(*, include_pending: bool = False, - idempotency_key: str | None = None) -> "RunHandle" + idempotency_key: str | None = None, + wait_for_ingest: bool = False) -> "RunHandle" ``` Start a new run that re-executes only the previous run's @@ -1224,6 +1346,9 @@ Shortcut for ``rerun(status="failed")``. Set started (``status="failed,pending"``) — the usual move after a ``stop()`` left orphan ``pending`` rows. +``wait_for_ingest=True`` returns only once the new run's task +rows are all written, as on :meth:`rerun`. + Returns a :class:`RunHandle` for the new run. Requires the previous run to be terminal (``completed`` / ``stopped`` / ``failed``); raises @@ -1235,10 +1360,22 @@ previous run to be terminal (``completed`` / ``stopped`` / #### add\_tasks ```python -def add_tasks(body: AddTasksRequest | AddTasksDict) -> AddTasksResponse +def add_tasks(body: AddTasksRequest | AddTasksDict, + *, + idempotency_key: str | None = None) -> AddTasksResponse ``` -``POST /jobs/{id}/tasks`` — append to the open initial run. +``POST /jobs/{id}/tasks``: append up to 10,000 tasks to the +open initial run. + +Pass ``idempotency_key`` (up to 128 characters, unique per +logical call) to make the call safe to retry. Within 24 hours, +a repeat with the same key and body returns the first answer +instead of adding the tasks again, and the client then retries +transient failures (network errors, 429, 502, 503, 504) on its +own. A repeat with a different body, or one sent while the first +call is still running, raises 409 ``idempotency_key_conflict``. +Without a key every call appends its tasks and is sent once. @@ -1322,11 +1459,16 @@ finished writing task rows. Large submissions return ``202 Accepted`` and stream task rows into storage off the request path; until that finishes, results pages may be partial and ``add_tasks`` on an open job -is rejected with 409. This polls ``GET /jobs/{id}`` until -``latest_run.ingest_status`` leaves ``pending`` (also satisfied -by a terminal run — a mid-ingest stop flips the field to -``done``). Runs that never ingested asynchronously (any 201 -submission) are done on the first poll. +is rejected with 409. This polls until the run's +``ingest_status`` is no longer ``pending``. On a ref from a +``submit_*`` call it polls the run that submit returned +(``GET /jobs/{id}/runs/{run_id}``), so a replayed submit never +follows a newer run; otherwise it polls the job's latest run. +The run status does not end the wait: a run stopped mid-ingest +is terminal at once but keeps ``ingest_status: pending`` until +the API has settled its rows and cut ``stats.total`` to match. +Runs that never ingested asynchronously (any 201 submission) +are done on the first poll. Raises ``WaiterTimeout`` after ``timeout`` seconds. Returns the fresh loaded handle, so chains like @@ -1393,6 +1535,22 @@ class WaiterError(RuntimeError) Raised when the resource entered a `failure_states` value. + + +## RunPaused Objects + +```python +class RunPaused(WaiterError) +``` + +Raised by a run waiter when the run is paused +(`pause_state: paused`) and not yet in a target status. + +A paused run makes no progress until someone resumes it, so the +waiter stops at once instead of polling until its timeout. `run` is +the run as last read. Resume it with `job.run.resume()` and wait +again, or pass `wait_while_paused=True` to keep polling. + #### poll\_until @@ -1453,6 +1611,14 @@ so we never page faster than we drain. importable; falls back to no-op if it isn't (so the SDK works in minimal envs without a hard dep). +Rows with no body to fetch are skipped by the bulk helpers: failed +tasks, and successful tasks whose result is past the job's retention +window (14 days by default), which the API returns as `expired` even +under a `status="successful"` filter. Pass `on_skip=` to be called +with each of those rows; a WARNING on `zenrows.batch.download` gives the +counts. Any other row without a `result_url` still raises, so a result +never goes missing silently. + ## DownloadedResult Objects @@ -1503,11 +1669,16 @@ def download_to_dir( concurrency: int = 1, progress: bool = False, max_files: int = DEFAULT_MAX_FILES, - max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE) -> int + max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE, + on_skip: Callable[[TaskResult], None] | None = None) -> int ``` Stream every task's body into `target_dir`. Returns the count. +Failed and `expired` rows have no body: they are skipped, passed +to `on_skip` when you set it, and don't count toward +`max_files`. + `concurrency` parallelises the body-fetch + write. With the default 1, behaviour is exactly serial. With N>1, up to N bodies are fetched in flight; iteration of the result list stays @@ -1557,12 +1728,17 @@ def download_to_memory( progress: bool = False, max_count: int = DEFAULT_MAX_COUNT_IN_MEMORY, max_total_bytes: int = DEFAULT_MAX_TOTAL_BYTES_IN_MEMORY, - max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE + max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE, + on_skip: Callable[[TaskResult], None] | None = None ) -> list[DownloadedResult] ``` Load every (matching) task body into a list and return it. +Failed and `expired` rows have no body: they are skipped, passed +to `on_skip` when you set it, and don't count toward +`max_count`. + Three independent caps that all raise `DownloadLimitExceeded`: - `max_count` — how many rows total - `max_total_bytes` — running sum of body sizes @@ -1830,6 +2006,78 @@ daily / weekly / monthly cadence. `"09:30"`). `cadence` is exactly one of `Daily()`, `Weekly(...)`, or `Monthly(...)`. `timezone` is mandatory (IANA name). +# Webhook signatures + +Verify the signature on a Batch webhook delivery. + +When a job's webhook has `signature: true`, every delivery carries an +`X-Signature` header shaped `t=,v1=,kid=`. +The digest is HMAC-SHA256 over the bytes `.`, keyed +with your signing secret. The API returns that secret once, base64-encoded, +from `rotate_hmac_key()`; `verify_webhook_signature` decodes it for you. + + from zenrows.batch import WEBHOOK_SIGNATURE_HEADER, verify_webhook_signature + + ok = verify_webhook_signature( + request.body, # raw bytes, before any JSON parsing + request.headers[WEBHOOK_SIGNATURE_HEADER], + {"": ""}, # every key you may still receive + ) + +During a key rotation, keep both the old and the new secret in the +mapping: the header's `kid` picks the one that signed. Deliveries can +repeat, so deduplicate on the `X-ZenRows-Event-Id` header +(`WEBHOOK_EVENT_ID_HEADER`). + + + +#### WEBHOOK\_SIGNATURE\_HEADER + +The header that carries a signed delivery's signature. + + + +#### WEBHOOK\_EVENT\_ID\_HEADER + +The header that carries a delivery's event id. Deduplicate on it. + + + +#### DEFAULT\_WEBHOOK\_TOLERANCE\_SECONDS + +Seconds a signature's timestamp may be from the current time. + + + +#### verify\_webhook\_signature + +```python +def verify_webhook_signature( + raw_body: bytes | str, + signature_header: str, + secrets: Mapping[str, str], + *, + tolerance_seconds: float = DEFAULT_WEBHOOK_TOLERANCE_SECONDS, + now: float | None = None) -> bool +``` + +Return True when `signature_header` is a valid signature of `raw_body`. + +- `raw_body`: the request body exactly as received (a `str` is + encoded as utf-8). Verify before parsing the JSON. +- `signature_header`: the value of the `X-Signature` header. +- `secrets`: maps each key id (`kid`) to its secret as the API + returned it: a base64 string that decodes to 32 bytes. Any other + secret, including an empty one, never verifies. +- `tolerance_seconds`: the largest accepted gap, in seconds, between the + signature's timestamp and `now`, in either direction. `0` turns + the check off; a negative value rejects every delivery. +- `now`: the current Unix time in seconds (default: `time.time()`). + +The digests are compared in constant time. Returns False, and never +raises, for a malformed header, an unknown `kid`, a bad secret, a +wrong digest or a timestamp out of range. + # Errors RFC 7807 Problem JSON → friendly Python exceptions. @@ -1867,6 +2115,157 @@ Parse a Problem body. Returns None if the body isn't JSON. Tolerant — production servers occasionally return non-JSON errors (e.g. ALB blocked at the edge); we degrade gracefully. + + +## BatchErrorCode Objects + +```python +class BatchErrorCode(str, Enum) +``` + +The stable `code` values the Batch API sends in its problem+json +errors, and in the `error` of a failed task. + +Members are strings, so they compare equal to `BatchAPIError.code`: + + except BatchAPIError as exc: + if exc.code == BatchErrorCode.QUOTA_EXCEEDED: + ... + +The set is open: the API may add codes, and `BatchAPIError.code` +stays a plain `str`, so keep a fallback branch. + + + +#### INVALID\_ARGUMENT + +400: the request failed validation. `extras["invalid_tasks"]` +lists per-task reasons when the problem is in `tasks`. + + + +#### NOT\_FOUND + +404: no such resource, or it is not yours. + + + +#### CONFLICT + +409: the request conflicts with the resource's current state. + + + +#### UNAUTHENTICATED + +401: missing or invalid API key. + + + +#### SUBSCRIPTION\_INACTIVE + +401: the key is valid, but the subscription is not active +(past due, unpaid, paused). Fix it in Billing. + + + +#### PAYMENT\_REQUIRED + +402: the account is out of credits. + + + +#### API\_KEY\_CAP\_REACHED + +402: this API key reached one of its credit caps. `detail` says +which cap, and when it resets. + + + +#### PERMISSION\_DENIED + +403: the account may not use this feature. + + + +#### QUOTA\_EXCEEDED + +429: an account limit was reached: active jobs (3, shared with +Crawl), scheduled jobs (5) or an open job's task cap (100,000). Not +transient; the client does not retry it. + + + +#### INTERNAL + +500: an unexpected server error. Also the `code` of a +`BatchAPIError` whose response had no problem body. + + + +#### NOT\_IMPLEMENTED + +501: the feature is not available. + + + +#### RUN\_NOT\_TERMINAL + +The previous run is still pending or running (for example on a +rerun), or a large submit is still ingesting (on add-tasks). + + + +#### NO\_MATCHING\_TASKS + +A partial rerun's `status` filter matched no task. + + + +#### IDEMPOTENCY\_KEY\_CONFLICT + +The `Idempotency-Key` was used with a different body, or the +first call with it is still running. + + + +#### PARTIAL\_RERUN\_WINDOW\_EXPIRED + +A partial rerun (`status=`) more than 10 days after the run it +retries started. A full rerun still works. + + + +#### RESULT\_EXPIRED + +410: the result is past the job's retention window (14 days by +default). + + + +#### FILE\_INPUT\_NOT\_FOUND + +404: no such file input, or its 24 h slot expired. + + + +#### FILE\_INPUT\_NOT\_UPLOADED + +409: the CSV was never uploaded to the slot. + + + +#### FILE\_INPUT\_UNSUPPORTED\_FOR\_JOB\_TYPE + +409: CSV input works only for closed regular jobs and scheduled +jobs. + + + +#### GATEWAY\_UNREACHABLE + +The scraping API could not be reached for this task. + ## BatchAPIError Objects @@ -1879,8 +2278,9 @@ A non-2xx response from the Batch API. `code` is the RFC 7807 `code` member (e.g. `file_input_not_found`, `idempotency_key_conflict`, `api_key_cap_reached`). Stable; safe to -branch on. `detail` is the human-readable explanation (e.g. which -credit cap was reached and when it resets): display it, don't parse it. +branch on; `BatchErrorCode` names every code the API sends. `detail` +is the human-readable explanation (e.g. which credit cap was reached +and when it resets): display it, don't parse it. # Models @@ -1985,6 +2385,12 @@ Clients must accept values not listed here. class TaskStatus(Enum) ``` +- `expired`: a successful task whose result is past the job's + retention window (14 days by default). The row has no + `result_url`, and the content endpoint answers + `410 result_expired`. A `status=successful` filter still + returns these rows. + Clients must accept values not listed here. @@ -2144,11 +2550,9 @@ cost?" indicator, not for reconciliation. class TaskSpend(BaseModel) ``` -Per-task indicative spend with two roll-ups: `total` -accumulates across every attempt (including retries), -`last_attempt` carries just the most recent gateway call. -On a task that succeeded on its first try the two are -equal; on a retried task they diverge. +Deprecated: result rows never carry a per-task spend, so this +model is never filled. Kept so existing imports keep working. +Read the run-level `stats.spend` instead. @@ -2182,8 +2586,10 @@ runs that predate the feature. Vocabulary (the only keys that appear): - `auth_failed` — credentials or billing -- `blocked` — anti-bot / policy denials -- `bad_target` — target URL is the problem (bad host, 404, 410, too large) +- `blocked`: anti-bot / policy denials, including a bot wall + Extract met on the fetched page +- `bad_target`: target URL is the problem (bad host, 404, 410, + too large, or a page Extract does not cover or found no data on) - `rate_limited` — target throttled the request - `timeout` — the scrape didn't complete in time - `gateway_error` — Zenrows-side transport / 5xx @@ -2767,8 +3173,9 @@ RFC 7807 Problem Details. #### code -Stable machine-readable error code, e.g. `payment_required` -or `api_key_cap_reached`. Branch on this, not on `detail`. +Stable machine-readable error code, e.g. `payment_required`, +`api_key_cap_reached` or `subscription_inactive`. Branch on +this, not on `detail`. Clients must accept values not listed here. @@ -2834,7 +3241,7 @@ re-fetch the URL right before you download. Non-empty only when `status = failed`. Stable strings — e.g. `"results are larger than N gb"` when the combined -results exceed the export byte cap (20 GiB). +results exceed the export byte cap (100 GiB). @@ -2949,9 +3356,11 @@ responses. #### result\_url -24-hour presigned download URL for the result body, or a -`/v1/jobs//runs//tasks//content` URL you can -fetch directly. Empty for non-successful tasks. +Presigned download URL for the result body, valid for 2 +hours; list the results again for a fresh one. Empty for +failed and `expired` tasks. The body is also served by +`GET /v1/jobs/{job_id}/runs/{run_id}/tasks/{task_id}/content`, +which needs your API key. @@ -2972,6 +3381,14 @@ source run's stored result. On chained retries, `source_run_id` chases back to the run that actually owns the result. Empty for normally-executed rows. + + +#### spend + +Deprecated: result rows never carry a per-task spend, so +this is always empty. Read the run-level `stats.spend` +instead. + #### created\_at @@ -3153,7 +3570,7 @@ ignored otherwise. #### tasks -Required for closed jobs (1–1000) unless `file_input_id` +Required for closed jobs (1 to 100,000) unless `file_input_id` is provided. Optional for open jobs — may be empty if the caller will follow up with `addTasks`. Mutually exclusive with `file_input_id`. diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 4c80d2f..bd89dd5 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -86,9 +86,6 @@ paths: `ingest_status` field reports the ingest state: poll `GET /v1/jobs/{id}` until `latest_run.ingest_status` is `done` to know every task row is visible. - - `type: scheduled` is reserved for a future feature and - currently returns `501 not_implemented`. parameters: - $ref: '#/components/parameters/IdempotencyKey' requestBody: @@ -127,7 +124,7 @@ paths: '401': { $ref: '#/components/responses/Unauthenticated' } '402': { $ref: '#/components/responses/PaymentRequired' } '409': { $ref: '#/components/responses/IdempotencyConflict' } - '501': { $ref: '#/components/responses/NotImplemented' } + '429': { $ref: '#/components/responses/QuotaExceeded' } '503': { $ref: '#/components/responses/Unavailable' } get: @@ -204,6 +201,15 @@ paths: Setting `last_batch: true` closes the job (status → `closed`) and signals the run to flip to `completed` once tasks drain. + + Honors `Idempotency-Key`, scoped to the caller and the job: a + repeat of a call that answered `200` replays that answer, and a + repeat of a call that failed part-way (a `503` after its tasks + were stored, or a lost response) finishes it without adding, + enqueuing or counting any task twice. Without the header every + call appends its tasks. + parameters: + - $ref: '#/components/parameters/IdempotencyKey' requestBody: required: true content: @@ -219,12 +225,17 @@ paths: '401': { $ref: '#/components/responses/Unauthenticated' } '402': { $ref: '#/components/responses/PaymentRequired' } '404': { $ref: '#/components/responses/NotFound' } + '429': { $ref: '#/components/responses/QuotaExceeded' } '409': description: | Job is not open, last batch already received, or the initial run's task ingestion is still in progress (large `202`-accepted submissions) — in the last case, retry once ingest completes. + + Also returned when an `Idempotency-Key` is reused with a + different body, or while an earlier call with the same key + is still running (retry shortly). content: application/problem+json: schema: { $ref: '#/components/schemas/Problem' } @@ -333,7 +344,7 @@ paths: latest run is terminal (use `/rerun` to start fresh) and 404 on a deleted job. - Billing note: in-flight gateway calls still consume credits. + Billing note: requests already in flight still consume credits. Use `/cancel` for a hard cost-stop. responses: '200': @@ -1002,7 +1013,7 @@ paths: The export fails with `error: "results are larger than N gb"` if the combined size - of the results exceeds the export byte cap (20 GiB). + of the results exceeds the export byte cap (100 GiB). responses: '202': description: Export accepted; the zip is produced asynchronously. @@ -1078,9 +1089,11 @@ components: in: header required: false description: | - Optional opaque token scoping a single submit / rerun - request. Re-submitting with the same key returns the - original response (or 409 on body mismatch). + Optional opaque token scoping a single submit, rerun or + add-tasks request. Sending the same key again with the same + body replays the original status and body. A different body, + or a repeat while the original is still running, returns + `409 idempotency_key_conflict`. Keys are kept for 24 hours. schema: { type: string, maxLength: 128 } ResultStatus: name: status @@ -1138,6 +1151,15 @@ components: content: application/problem+json: schema: { $ref: '#/components/schemas/Problem' } + QuotaExceeded: + description: | + Account limit reached (`quota_exceeded`): too many active jobs + (3 per account, shared with Crawl), too many scheduled jobs (5), + or an open job's task cap (100,000). Not transient: retrying + does not help until a job finishes, is closed or is deleted. + content: + application/problem+json: + schema: { $ref: '#/components/schemas/Problem' } NotImplemented: description: Feature is reserved but not yet implemented. content: @@ -1236,8 +1258,15 @@ components: TaskStatus: type: string x-extensible-enum: true - enum: [pending, processing, successful, failed] - description: Clients must accept values not listed here. + enum: [pending, processing, successful, failed, expired] + description: | + - `expired`: a successful task whose result is past the job's + retention window (14 days by default). The row has no + `result_url`, and the content endpoint answers + `410 result_expired`. A `status=successful` filter still + returns these rows. + + Clients must accept values not listed here. ResultType: type: string @@ -1310,7 +1339,7 @@ components: header sanitisation. At most 32 headers / 8 KB serialised; names must be valid HTTP tokens. Header filtering and browser-header sanitisation are owned by the scraper API, - not Conveyor. A task-level + not the Batch API. A task-level `custom_headers` replaces the job-level object wholesale — no per-name merging. additionalProperties: @@ -1398,12 +1427,12 @@ components: tasks: type: array description: | - Required for closed jobs (1–1000) unless `file_input_id` + Required for closed jobs (1 to 100,000) unless `file_input_id` is provided. Optional for open jobs — may be empty if the caller will follow up with `addTasks`. Mutually exclusive with `file_input_id`. minItems: 0 - maxItems: 1000 + maxItems: 100000 items: { $ref: '#/components/schemas/TaskInput' } file_input_id: type: string @@ -1468,7 +1497,7 @@ components: tasks: type: array minItems: 1 - maxItems: 1000 + maxItems: 10000 items: { $ref: '#/components/schemas/TaskInput' } last_batch: type: boolean @@ -1524,12 +1553,11 @@ components: TaskSpend: type: object + deprecated: true description: | - Per-task indicative spend with two roll-ups: `total` - accumulates across every attempt (including retries), - `last_attempt` carries just the most recent gateway call. - On a task that succeeded on its first try the two are - equal; on a retried task they diverge. + Deprecated: result rows never carry a per-task spend, so this + model is never filled. Kept so existing imports keep working. + Read the run-level `stats.spend` instead. required: [total, last_attempt] properties: total: { $ref: '#/components/schemas/Spend' } @@ -1554,8 +1582,10 @@ components: Vocabulary (the only keys that appear): - `auth_failed` — credentials or billing - - `blocked` — anti-bot / policy denials - - `bad_target` — target URL is the problem (bad host, 404, 410, too large) + - `blocked`: anti-bot / policy denials, including a bot wall + Extract met on the fetched page + - `bad_target`: target URL is the problem (bad host, 404, 410, + too large, or a page Extract does not cover or found no data on) - `rate_limited` — target throttled the request - `timeout` — the scrape didn't complete in time - `gateway_error` — Zenrows-side transport / 5xx @@ -1581,6 +1611,7 @@ components: job_id: { type: string } run_sequence: { type: integer, minimum: 1 } status: { $ref: '#/components/schemas/RunStatus' } + trigger: { $ref: '#/components/schemas/RunTrigger' } stats: { $ref: '#/components/schemas/RunStats' } last_batch_received: type: boolean @@ -1850,9 +1881,11 @@ components: type: string nullable: true description: | - 24-hour presigned download URL for the result body, or a - `/v1/jobs//runs//tasks//content` URL you can - fetch directly. Empty for non-successful tasks. + Presigned download URL for the result body, valid for 2 + hours; list the results again for a fresh one. Empty for + failed and `expired` tasks. The body is also served by + `GET /v1/jobs/{job_id}/runs/{run_id}/tasks/{task_id}/content`, + which needs your API key. error: $ref: '#/components/schemas/Problem' description: | @@ -1868,7 +1901,12 @@ components: source run's stored result. On chained retries, `source_run_id` chases back to the run that actually owns the result. Empty for normally-executed rows. - spend: { $ref: '#/components/schemas/TaskSpend' } + spend: + $ref: '#/components/schemas/TaskSpend' + description: | + Deprecated: result rows never carry a per-task spend, so + this is always empty. Read the run-level `stats.spend` + instead. created_at: type: string format: date-time @@ -2198,8 +2236,9 @@ components: code: type: string description: | - Stable machine-readable error code, e.g. `payment_required` - or `api_key_cap_reached`. Branch on this, not on `detail`. + Stable machine-readable error code, e.g. `payment_required`, + `api_key_cap_reached` or `subscription_inactive`. Branch on + this, not on `detail`. Clients must accept values not listed here. instance: { type: string } invalid_tasks: @@ -2284,7 +2323,7 @@ components: description: | Non-empty only when `status = failed`. Stable strings — e.g. `"results are larger than N gb"` when the combined - results exceed the export byte cap (20 GiB). + results exceed the export byte cap (100 GiB). download_url: type: string format: uri diff --git a/examples/02_download_to_dir.py b/examples/02_download_to_dir.py index 3c185f8..ad2842e 100644 --- a/examples/02_download_to_dir.py +++ b/examples/02_download_to_dir.py @@ -23,6 +23,9 @@ fill the disk. - `status="successful"` (the default) skips failed rows; pass `status=None` to grab everything. + - Results are kept for 14 days by default. After that a successful + task reads as `expired` and has no body; the download skips it + (and any failed row) and passes it to `on_skip=`. Run with: export ZENROWS_API_KEY=zr_... @@ -33,7 +36,7 @@ import os from zenrows import ZenRowsBatchClient -from zenrows.batch import DownloadLimitExceeded +from zenrows.batch import DownloadLimitExceeded, TaskResult def main() -> None: @@ -60,6 +63,7 @@ def main() -> None: job = client.get_job(args.job_id) run = job.run.wait(timeout=600.0, progress=True) + skipped: list[TaskResult] = [] try: written = run.download_to_dir( args.out, @@ -68,8 +72,9 @@ def main() -> None: progress=True, # live tqdm bar max_files=20_000, # cap row count max_bytes_per_file=10 * 1024 * 1024, # cap single body @ 10 MiB + on_skip=skipped.append, # expired or failed rows, which have no body ) - print(f"wrote {written} files to {args.out}/") + print(f"wrote {written} files to {args.out}/ ({len(skipped)} skipped)") except DownloadLimitExceeded as exc: # `limit_name` is one of: max_files, max_bytes_per_file. # `limit` + `observed` for diagnosis. diff --git a/examples/06_retry_failed.py b/examples/06_retry_failed.py index 6582127..9eb6f0d 100644 --- a/examples/06_retry_failed.py +++ b/examples/06_retry_failed.py @@ -1,7 +1,7 @@ """06: Retry only the failed tasks of a finished run. `job.retry_failed()` starts a fresh run that re-executes ONLY the -previous run's failed tasks (partial retry, SPEC §3.5). Successful +previous run's failed tasks (partial retry). Successful tasks are inherited verbatim — you don't pay to re-scrape them, and the new run's totals already carry the prior successes. Pass `include_pending=True` to also re-enqueue tasks that never started @@ -11,7 +11,9 @@ Requires the previous run to be terminal (`completed` / `stopped` / `failed`); otherwise the API returns `409 run_not_terminal` — call -`job.run.stop()` first if it's still live. +`job.run.stop()` first if it's still live. A partial retry works within +10 days of the start of the run it retries; after that the API returns +`409 partial_rerun_window_expired` and only a full `job.rerun()` works. Run with: export ZENROWS_API_KEY=zr_... @@ -22,7 +24,7 @@ import os from zenrows import ZenRowsBatchClient -from zenrows.batch import BatchAPIError +from zenrows.batch import BatchAPIError, BatchErrorCode def main() -> None: @@ -46,6 +48,9 @@ def main() -> None: try: run = job.retry_failed(include_pending=args.include_pending) except BatchAPIError as e: + if e.code == BatchErrorCode.PARTIAL_RERUN_WINDOW_EXPIRED: + print("too late for a partial retry; use job.rerun() to run every task again") + return # e.g. no_matching_tasks (nothing failed) or run_not_terminal. print(f"retry skipped: {e.code}") return diff --git a/examples/07_hmac_rotation.py b/examples/07_hmac_rotation.py index d3aea5d..9500944 100644 --- a/examples/07_hmac_rotation.py +++ b/examples/07_hmac_rotation.py @@ -13,6 +13,11 @@ this is your one chance to capture it. Subsequent reads give you just `kid` + `created_at`. +The secret is base64-encoded. Keep it as returned and pass it, keyed +by `kid`, to `verify_webhook_signature`, which decodes it before +computing the HMAC. If you verify by hand, base64-decode it first: +the HMAC key is the decoded 32 bytes, not the text. + Typical client flow: 1. `rotate_hmac_key()` → install K2 alongside K1 in your verifier 2. (deploy K2 everywhere) @@ -48,6 +53,9 @@ def main() -> None: return raise + # Your receiver checks deliveries with every key it may still see: + # verify_webhook_signature(raw_body, headers[WEBHOOK_SIGNATURE_HEADER], + # {old_kid: old_secret, rotated.kid: rotated.secret}) # Once your verifiers accept the new key, promote it: # client.finalize_hmac_key() # ...or back out: diff --git a/examples/08_download_all_results.py b/examples/08_download_all_results.py index 26cfb79..8f77798 100644 --- a/examples/08_download_all_results.py +++ b/examples/08_download_all_results.py @@ -6,11 +6,12 @@ to disk. Use this when you want a single artifact; for one file per task instead, see `download_to_dir` (example 02). -The zip has a 12h TTL and is capped at 1 GiB server-side — an -oversize run fails the export and raises `BatchAPIError`. For larger -runs (or one file per task), use the iterative client-side download -`download_to_dir` (example 02): it has no size limit but is slower, -fetching bodies one at a time (tunable via `concurrency=`). +The export lives 12 h and its download link 2 h. The server caps one +export at 100 GiB of results: an oversize run fails the export and +raises `BatchAPIError`. For larger runs (or one file per task), use +the iterative client-side download `download_to_dir` (example 02): it +has no export cap but is slower, fetching bodies one at a time +(tunable via `concurrency=`). `job.run.download_all_results` targets the job's current run; address a specific run with `client.get_run(job_id, run_id=...).download_all_results(...)`. @@ -40,7 +41,7 @@ def main() -> None: path = job.run.download_all_results(args.out) print(f"wrote {path}") except BatchAPIError as e: - # e.g. the run exceeded the 1 GiB export cap. + # e.g. the run exceeded the 100 GiB export cap. print(f"export failed: {e}") diff --git a/examples/09_scheduled_jobs.py b/examples/09_scheduled_jobs.py index fe6df1f..5991f5a 100644 --- a/examples/09_scheduled_jobs.py +++ b/examples/09_scheduled_jobs.py @@ -22,7 +22,7 @@ """ import os -from datetime import datetime +from datetime import datetime, timedelta from zenrows import ZenRowsBatchClient from zenrows.batch import At, Calendar, JobHandle, Monthly, Rate, Weekly @@ -81,13 +81,17 @@ def main() -> None: ) print(f"created {monthly.job_id} — 1st + 15th at midnight UTC") - # At: a single future fire (naive local datetime + timezone). + # At: a single future fire (naive local datetime + timezone). The + # time must be in the future, so this picks 09:00 thirty days from now. + launch = (datetime.now() + timedelta(days=30)).replace( + hour=9, minute=0, second=0, microsecond=0 + ) oneshot = client.submit_scheduled( - At(datetime(2026, 9, 1, 9, 0), timezone="Europe/Berlin"), + At(launch, timezone="Europe/Berlin"), ["https://example.com/launch-day"], name="launch-day", ) - print(f"created {oneshot.job_id} — one-shot 2026-09-01 09:00 Berlin") + print(f"created {oneshot.job_id}: one-shot {launch:%Y-%m-%d %H:%M} Berlin") # --- 3. Manage the schedule (via the `job.schedule` facet) ---------- # Pause: the schedule keeps ticking server-side but fires are dropped diff --git a/scripts/build_reference.py b/scripts/build_reference.py index 6a2d38b..6ae8030 100644 --- a/scripts/build_reference.py +++ b/scripts/build_reference.py @@ -28,6 +28,7 @@ ("zenrows.batch._download", "Downloads"), ("zenrows.batch._estimate", "Cost estimation"), ("zenrows.batch._schedule", "Schedule builders"), + ("zenrows.batch._webhooks", "Webhook signatures"), ("zenrows.batch.errors", "Errors"), ("zenrows.batch.models", "Models"), ] diff --git a/src/zenrows/batch/__init__.py b/src/zenrows/batch/__init__.py index 4eccfb7..61d3ca6 100644 --- a/src/zenrows/batch/__init__.py +++ b/src/zenrows/batch/__init__.py @@ -11,8 +11,9 @@ Re-exports below are the curated, stable surface. Anything in `models` that callers need but isn't here can be imported directly. -Completion is surfaced via waiters (`job.wait()` / `run.wait()`) and -webhooks — poll or get notified; there's no callback to wire up. +Completion is surfaced via waiters (`job.run.wait()` / `run.wait()`) and +webhooks: poll or get notified; there's no callback to wire up. Check a +signed delivery with `verify_webhook_signature`. """ from zenrows.batch._download import DownloadedResult, DownloadLimitExceeded @@ -52,9 +53,15 @@ TaskInputDict, WebhookDict, ) -from zenrows.batch._waiters import WaiterError, WaiterTimeout +from zenrows.batch._waiters import RunPaused, WaiterError, WaiterTimeout +from zenrows.batch._webhooks import ( + DEFAULT_WEBHOOK_TOLERANCE_SECONDS, + WEBHOOK_EVENT_ID_HEADER, + WEBHOOK_SIGNATURE_HEADER, + verify_webhook_signature, +) from zenrows.batch.client import TERMINAL_RUN_STATUSES, ZenRowsBatchClient -from zenrows.batch.errors import BatchAPIError, ProblemDetail +from zenrows.batch.errors import BatchAPIError, BatchErrorCode, ProblemDetail from zenrows.batch.models import ( AddTasksRequest, CreateJobInputRequest, @@ -73,14 +80,20 @@ TestWebhookRequest, TestWebhookResponse, WebhookConfig, + WebhookEvent, + WebhookEventType, ) __all__ = [ + "DEFAULT_WEBHOOK_TOLERANCE_SECONDS", "TERMINAL_RUN_STATUSES", + "WEBHOOK_EVENT_ID_HEADER", + "WEBHOOK_SIGNATURE_HEADER", "AddTasksDict", "AddTasksRequest", "At", "BatchAPIError", + "BatchErrorCode", "CSVFieldsDict", "CSVSpecDict", "Cadence", @@ -108,6 +121,7 @@ "Rate", "Run", "RunHandle", + "RunPaused", "RunRef", "RunStatus", "Schedule", @@ -127,6 +141,9 @@ "WaiterTimeout", "WebhookConfig", "WebhookDict", + "WebhookEvent", + "WebhookEventType", "Weekly", "ZenRowsBatchClient", + "verify_webhook_signature", ] diff --git a/src/zenrows/batch/_download.py b/src/zenrows/batch/_download.py index b06ff50..ecc8a35 100644 --- a/src/zenrows/batch/_download.py +++ b/src/zenrows/batch/_download.py @@ -28,6 +28,14 @@ `progress=True` installs a `rich.progress` bar if `rich` is importable; falls back to no-op if it isn't (so the SDK works in minimal envs without a hard dep). + +Rows with no body to fetch are skipped by the bulk helpers: failed +tasks, and successful tasks whose result is past the job's retention +window (14 days by default), which the API returns as `expired` even +under a `status="successful"` filter. Pass `on_skip=` to be called +with each of those rows; a WARNING on `zenrows.batch.download` gives the +counts. Any other row without a `result_url` still raises, so a result +never goes missing silently. """ import os @@ -41,9 +49,16 @@ import httpx +from zenrows.batch._logging import logger from zenrows.batch.errors import BatchAPIError from zenrows.batch.models import TaskResult, TaskStatus +_log = logger("download") + +# Statuses whose rows never have a body to fetch: a failed task, or a +# successful one whose result aged out of the retention window. +_NO_BODY_STATUSES = frozenset({TaskStatus.FAILED, TaskStatus.EXPIRED}) + class _BodyFetcher(Protocol): """The slice of `ZenRowsBatchClient` the helpers need — just result @@ -107,9 +122,14 @@ def download_to_dir( progress: bool = False, max_files: int = DEFAULT_MAX_FILES, max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE, + on_skip: Callable[[TaskResult], None] | None = None, ) -> int: """Stream every task's body into `target_dir`. Returns the count. + Failed and `expired` rows have no body: they are skipped, passed + to `on_skip` when you set it, and don't count toward + `max_files`. + `concurrency` parallelises the body-fetch + write. With the default 1, behaviour is exactly serial. With N>1, up to N bodies are fetched in flight; iteration of the result list stays @@ -154,6 +174,7 @@ def handle(row: TaskResult) -> int: max_rows=max_files, max_rows_limit_name="max_files", progress_description="downloading to disk", + on_skip=on_skip, ) @@ -237,10 +258,15 @@ def _fetch_result_url(task: TaskResult) -> tuple[bytes, str]: """GET a task's presigned `result_url` directly — no API round-trip and no auth header (it's a signed storage URL). Returns `(body, content_type)`. Raises `ValueError` when the task carries no - result (e.g. a failed task).""" + result (a failed or `expired` task).""" if not task.result_url: + if task.status is TaskStatus.EXPIRED: + raise ValueError( + f"task {task.task_id!r} has expired: its result is past the " + "job's retention window and can no longer be downloaded" + ) raise ValueError( - f"task {task.task_id!r} has no result_url — " + f"task {task.task_id!r} has no result_url: " "only successful tasks have a downloadable body" ) with httpx.Client(timeout=httpx.Timeout(60.0)) as bare: @@ -283,9 +309,14 @@ def download_to_memory( max_count: int = DEFAULT_MAX_COUNT_IN_MEMORY, max_total_bytes: int = DEFAULT_MAX_TOTAL_BYTES_IN_MEMORY, max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE, + on_skip: Callable[[TaskResult], None] | None = None, ) -> list[DownloadedResult]: """Load every (matching) task body into a list and return it. + Failed and `expired` rows have no body: they are skipped, passed + to `on_skip` when you set it, and don't count toward + `max_count`. + Three independent caps that all raise `DownloadLimitExceeded`: - `max_count` — how many rows total - `max_total_bytes` — running sum of body sizes @@ -325,6 +356,7 @@ def handle(row: TaskResult) -> int: max_rows=max_count, max_rows_limit_name="max_count", progress_description="downloading to memory", + on_skip=on_skip, ) return out @@ -344,16 +376,74 @@ def _run_bulk( max_rows: int, max_rows_limit_name: str, progress_description: str, + on_skip: Callable[[TaskResult], None] | None = None, ) -> int: """Iterate results + apply `per_row` to each, with optional parallelism + progress bar. - Returns the number of rows processed. + Returns the number of rows processed. Rows with no body (see + `_with_bodies`) never reach `per_row` and don't count. Cap checking is done *here* rather than in `per_row` so the count is authoritative even when the closures run out of order. """ - rows = client._iter_results_raw(job_id, run_id=run_id, status=status) + counts: dict[str, int] = {} + rows = _with_bodies( + client._iter_results_raw(job_id, run_id=run_id, status=status), + on_skip=on_skip, + counts=counts, + ) + try: + return _drive( + rows, + per_row=per_row, + concurrency=concurrency, + progress=progress, + max_rows=max_rows, + max_rows_limit_name=max_rows_limit_name, + progress_description=progress_description, + ) + finally: + if counts: + _log.warning( + "skipped %d result(s) with no body to download (%s)", + sum(counts.values()), + ", ".join(f"{n} {st}" for st, n in sorted(counts.items())), + extra={"skipped": dict(counts)}, + ) + + +def _with_bodies( + rows: Iterable[TaskResult], + *, + on_skip: Callable[[TaskResult], None] | None, + counts: dict[str, int], +) -> Iterator[TaskResult]: + """Drop the rows that have no body: no `result_url` and a status of + `failed` or `expired`. Each one is passed to `on_skip` (when set) + and counted per status in `counts`. Every other row passes through, + so a `successful` row without a URL still fails loudly in the fetch. + Runs on the iterating thread, so `on_skip` is never called + concurrently.""" + for row in rows: + if not row.result_url and row.status in _NO_BODY_STATUSES: + if on_skip is not None: + on_skip(row) + counts[row.status.value] = counts.get(row.status.value, 0) + 1 + continue + yield row + + +def _drive( + rows: Iterable[TaskResult], + *, + per_row: Callable[[TaskResult], int], + concurrency: int, + progress: bool, + max_rows: int, + max_rows_limit_name: str, + progress_description: str, +) -> int: with _maybe_progress(progress, progress_description) as advance: if concurrency <= 1: diff --git a/src/zenrows/batch/_estimate.py b/src/zenrows/batch/_estimate.py index 9deb559..03dc755 100644 --- a/src/zenrows/batch/_estimate.py +++ b/src/zenrows/batch/_estimate.py @@ -51,9 +51,9 @@ AUTO_MIN_CREDITS = 1 AUTO_MAX_CREDITS = 25 -# Param values arrive as str | bool | int. These spellings -# count as "on" for the boolean flags. -_TRUTHY = frozenset({"true", "1", "yes", "on"}) +# Param values arrive as str | bool | int. The API accepts only `true` / +# `"true"` (any case) as "on" for a boolean flag and rejects `1`, `"yes"` +# or `"on"` with a 400, so only those count: a priced job is a valid one. # What `Tier` keys can appear, in the order a breakdown should render. ParamValue = str | bool | int | dict[str, str] # dict form: `custom_headers` map @@ -170,19 +170,18 @@ def format(self) -> str: def _truthy(value: ParamValue | None) -> bool: - """Coerce a scraper-param scalar to a boolean. Booleans pass - through; ints are nonzero-truthy; strings match the on-spellings.""" + """True only for the values the API reads as "on": `True`, or the + string `"true"` in any case.""" if isinstance(value, bool): return value - if isinstance(value, int): - return value != 0 if isinstance(value, str): - return value.strip().lower() in _TRUTHY + return value.lower() == "true" return False def _is_auto(params: ParamMap) -> bool: - return str(params.get("mode", "")).strip().lower() == "auto" + """The API accepts only the exact value `"auto"` for `mode`.""" + return params.get("mode") == "auto" def _cost_for_params(params: ParamMap) -> TaskCost: diff --git a/src/zenrows/batch/_resources.py b/src/zenrows/batch/_resources.py index 00f852d..3f3b4e7 100644 --- a/src/zenrows/batch/_resources.py +++ b/src/zenrows/batch/_resources.py @@ -46,6 +46,7 @@ AddTasksResponse, Export, ExportStatus, + IngestStatus, Job, JobStatus, RerunJobResponse, @@ -128,7 +129,12 @@ def download_to_dir( progress: bool = False, max_files: int | None = None, max_bytes_per_file: int | None = None, + on_skip: Callable[[TaskResult], None] | None = None, ) -> int: + """Write every matching task's body to ``target_dir``, one file + per task. Returns the number of files written. Failed and + ``expired`` rows have no body: they are skipped and passed to + ``on_skip`` when you set it.""" return self._client._download_dir( self.job_id, self.run_id, @@ -140,6 +146,7 @@ def download_to_dir( progress=progress, max_files=max_files, max_bytes_per_file=max_bytes_per_file, + on_skip=on_skip, ) def download_to_memory( @@ -151,7 +158,11 @@ def download_to_memory( max_count: int | None = None, max_total_bytes: int | None = None, max_bytes_per_file: int | None = None, + on_skip: Callable[[TaskResult], None] | None = None, ) -> "list[DownloadedResult]": + """Load every matching task's body into memory. Failed and + ``expired`` rows have no body: they are skipped and passed to + ``on_skip`` when you set it.""" return self._client._download_memory( self.job_id, self.run_id, @@ -161,6 +172,7 @@ def download_to_memory( max_count=max_count, max_total_bytes=max_total_bytes, max_bytes_per_file=max_bytes_per_file, + on_skip=on_skip, ) # ----- single-task download ----- @@ -186,10 +198,13 @@ def wait( timeout: float = 300.0, poll_interval: float = 2.0, progress: bool = False, + wait_while_paused: bool = False, ) -> "RunHandle": """Block until this run reaches a target state. Returns a fresh loaded :class:`RunHandle` so chains like - ``run.wait().download_to_dir(...)`` work without re-wrapping.""" + ``run.wait().download_to_dir(...)`` work without re-wrapping. + A paused run raises :class:`RunPaused` unless + ``wait_while_paused=True``.""" run = self._client._wait_for_run_raw( self.job_id, run_id=self.run_id, @@ -198,6 +213,7 @@ def wait( timeout=timeout, poll_interval=poll_interval, progress=progress, + wait_while_paused=wait_while_paused, ) return RunHandle(self._client, self.job_id, self.run_id, run) @@ -226,9 +242,9 @@ def download_all_results( the zip to ``target_path``. See ``ZenRowsBatchClient.download_all_results`` for the contract. - Capped at 1 GiB per run server-side; for larger runs use - ``download_to_dir`` (no size limit, but slower — one body at a - time, tunable via ``concurrency=``).""" + The server caps one export at 100 GiB of results. For a larger + run use ``download_to_dir``: it has no export cap, but is slower + (one body at a time, tunable via ``concurrency=``).""" return self._client.download_all_results( self.job_id, self.run_id, @@ -279,7 +295,8 @@ class ExportRef: A results export is async: ``start_export()`` returns immediately with a ``pending`` ref; the server zips the run in the background and it reaches ``completed`` (download URL ready) or ``failed`` - (with an ``error`` message — e.g. the 1 GiB size cap). Call + (with an ``error`` message, e.g. the run is over the 100 GiB + export cap). Call ``load()`` / ``wait()`` for an :class:`ExportHandle` with data. """ @@ -485,12 +502,15 @@ def wait( timeout: float = 300.0, poll_interval: float = 2.0, progress: bool = False, + wait_while_paused: bool = False, ) -> "RunHandle": """Block until the current run reaches a target state. Returns a loaded :class:`RunHandle`, so chains like ``job.run.wait().download_to_dir(...)`` work cleanly. - See ``ZenRowsBatchClient.wait_for_run`` for the full contract.""" + A paused run raises :class:`RunPaused` instead of being polled + until ``timeout``, unless ``wait_while_paused=True``. See + ``ZenRowsBatchClient.wait_for_run`` for the full contract.""" run = self._client._wait_for_run_raw( self.job_id, run_id=None, @@ -499,6 +519,7 @@ def wait( timeout=timeout, poll_interval=poll_interval, progress=progress, + wait_while_paused=wait_while_paused, ) return RunHandle(self._client, self.job_id, run.run_id, run) @@ -525,7 +546,12 @@ def download_to_dir( progress: bool = False, max_files: int | None = None, max_bytes_per_file: int | None = None, + on_skip: Callable[[TaskResult], None] | None = None, ) -> int: + """Write every matching task's body to ``target_dir``, one file + per task. Returns the number of files written. Failed and + ``expired`` rows have no body: they are skipped and passed to + ``on_skip`` when you set it.""" return self._client._download_dir( self.job_id, None, @@ -537,6 +563,7 @@ def download_to_dir( progress=progress, max_files=max_files, max_bytes_per_file=max_bytes_per_file, + on_skip=on_skip, ) def download_to_memory( @@ -548,7 +575,11 @@ def download_to_memory( max_count: int | None = None, max_total_bytes: int | None = None, max_bytes_per_file: int | None = None, + on_skip: Callable[[TaskResult], None] | None = None, ) -> "list[DownloadedResult]": + """Load every matching task's body into memory. Failed and + ``expired`` rows have no body: they are skipped and passed to + ``on_skip`` when you set it.""" return self._client._download_memory( self.job_id, None, @@ -558,6 +589,7 @@ def download_to_memory( max_count=max_count, max_total_bytes=max_total_bytes, max_bytes_per_file=max_bytes_per_file, + on_skip=on_skip, ) # ----- single-task download ----- @@ -594,9 +626,9 @@ def download_all_results( and save the zip to ``target_path``. See ``ZenRowsBatchClient.download_all_results`` for the contract. - Capped at 1 GiB per run server-side; for larger runs use - ``download_to_dir`` (no size limit, but slower — one body at a - time, tunable via ``concurrency=``).""" + The server caps one export at 100 GiB of results. For a larger + run use ``download_to_dir``: it has no export cap, but is slower + (one body at a time, tunable via ``concurrency=``).""" return self._client.download_all_results( self.job_id, self._current_run_id(), @@ -741,6 +773,7 @@ def rerun( *, status: str | list[str] | None = None, idempotency_key: str | None = None, + wait_for_ingest: bool = False, ) -> "RunHandle": """``POST /jobs/{id}/rerun[?status=...]`` — start a new run. @@ -751,19 +784,40 @@ def rerun( With ``status`` (single value like ``"failed"`` or list like ``["failed", "pending"]``): partial retry — matching statuses are reset to ``pending`` and re-enqueued; everything else is - inherited verbatim with ``source_run_id`` stamped. + inherited verbatim with ``source_run_id`` stamped. A partial + retry works within 10 days of the start of the run it retries; + after that the API answers 409 ``partial_rerun_window_expired`` + and only a full rerun works. + + Every rerun is accepted asynchronously (``202``): the new run's + task rows keep arriving until ``ingest_status`` is ``done``, so + ``results()`` read straight away can be partial. Pass + ``wait_for_ingest=True`` to poll the run this call returned + (``GET /jobs/{id}/runs/{run_id}``, for up to 300 s) until its + ``ingest_status`` is no longer ``pending``, and get that fresh + snapshot back. This also holds when the call replays an earlier + answer for the same ``idempotency_key``. A run stopped while it + was ingesting comes back ``stopped`` once ingestion has wound + down, with only the rows written so far (``stats.total`` is cut + to match). Otherwise use ``run.wait()`` before reading results. Returns a :class:`RunHandle` for the newly-created run.""" resp: RerunJobResponse = self._client._post_rerun( self.job_id, status=status, idempotency_key=idempotency_key ) - return RunHandle(self._client, self.job_id, resp.latest_run.run_id, resp.latest_run) + run = resp.latest_run + if wait_for_ingest and run.ingest_status is IngestStatus.PENDING: + run = self._client._wait_for_run_ingest_raw( + self.job_id, run.run_id, timeout=300.0, poll_interval=2.0 + ) + return RunHandle(self._client, self.job_id, run.run_id, run) def retry_failed( self, *, include_pending: bool = False, idempotency_key: str | None = None, + wait_for_ingest: bool = False, ) -> "RunHandle": """Start a new run that re-executes only the previous run's **failed** tasks (partial retry). Successful tasks @@ -775,6 +829,9 @@ def retry_failed( started (``status="failed,pending"``) — the usual move after a ``stop()`` left orphan ``pending`` rows. + ``wait_for_ingest=True`` returns only once the new run's task + rows are all written, as on :meth:`rerun`. + Returns a :class:`RunHandle` for the new run. Requires the previous run to be terminal (``completed`` / ``stopped`` / ``failed``); raises @@ -782,13 +839,30 @@ def retry_failed( (409 ``no_matching_tasks``) when nothing matched the filter. """ statuses = ["failed", "pending"] if include_pending else "failed" - return self.rerun(status=statuses, idempotency_key=idempotency_key) + return self.rerun( + status=statuses, idempotency_key=idempotency_key, wait_for_ingest=wait_for_ingest + ) # ----- tasks (write path) ----- - def add_tasks(self, body: AddTasksRequest | AddTasksDict) -> AddTasksResponse: - """``POST /jobs/{id}/tasks`` — append to the open initial run.""" - return self._client._post_tasks(self.job_id, body) + def add_tasks( + self, + body: AddTasksRequest | AddTasksDict, + *, + idempotency_key: str | None = None, + ) -> AddTasksResponse: + """``POST /jobs/{id}/tasks``: append up to 10,000 tasks to the + open initial run. + + Pass ``idempotency_key`` (up to 128 characters, unique per + logical call) to make the call safe to retry. Within 24 hours, + a repeat with the same key and body returns the first answer + instead of adding the tasks again, and the client then retries + transient failures (network errors, 429, 502, 503, 504) on its + own. A repeat with a different body, or one sent while the first + call is still running, raises 409 ``idempotency_key_conflict``. + Without a key every call appends its tasks and is sent once.""" + return self._client._post_tasks(self.job_id, body, idempotency_key=idempotency_key) # ----- runs (list) ----- @@ -856,26 +930,39 @@ def wait_for_ingest( Large submissions return ``202 Accepted`` and stream task rows into storage off the request path; until that finishes, results pages may be partial and ``add_tasks`` on an open job - is rejected with 409. This polls ``GET /jobs/{id}`` until - ``latest_run.ingest_status`` leaves ``pending`` (also satisfied - by a terminal run — a mid-ingest stop flips the field to - ``done``). Runs that never ingested asynchronously (any 201 - submission) are done on the first poll. + is rejected with 409. This polls until the run's + ``ingest_status`` is no longer ``pending``. On a ref from a + ``submit_*`` call it polls the run that submit returned + (``GET /jobs/{id}/runs/{run_id}``), so a replayed submit never + follows a newer run; otherwise it polls the job's latest run. + The run status does not end the wait: a run stopped mid-ingest + is terminal at once but keeps ``ingest_status: pending`` until + the API has settled its rows and cut ``stats.total`` to match. + Runs that never ingested asynchronously (any 201 submission) + are done on the first poll. Raises ``WaiterTimeout`` after ``timeout`` seconds. Returns the fresh loaded handle, so chains like ``job.wait_for_ingest().add_tasks(...)`` work. Or opt in at submit time via ``submit_job(..., wait_for_ingest=True)``.""" - return JobHandle( - self._client, - self.job_id, - self._client._wait_for_ingest_raw( + submitted = self.submit_response.latest_run if self.submit_response else None + if submitted is not None: + self._client._wait_for_run_ingest_raw( self.job_id, + submitted.run_id, timeout=timeout, poll_interval=poll_interval, max_poll_interval=max_poll_interval, - ), - ) + ) + data = self._client._get_job_data(self.job_id) + else: + data = self._client._wait_for_ingest_raw( + self.job_id, + timeout=timeout, + poll_interval=poll_interval, + max_poll_interval=max_poll_interval, + ) + return JobHandle(self._client, self.job_id, data, submit_response=self.submit_response) class JobHandle(JobRef): diff --git a/src/zenrows/batch/_transport.py b/src/zenrows/batch/_transport.py index b29ccc7..7c98218 100644 --- a/src/zenrows/batch/_transport.py +++ b/src/zenrows/batch/_transport.py @@ -34,7 +34,7 @@ _RETRYABLE_STATUSES = frozenset({429, 502, 503, 504}) # Methods safe to replay without side effects. POST is added only when -# the caller supplied an Idempotency-Key (submit / rerun). +# the caller supplied an Idempotency-Key (submit, rerun, add-tasks). _IDEMPOTENT_METHODS = frozenset({"GET", "PUT", "DELETE", "HEAD", "OPTIONS"}) # Retry tuning (mirrors the Node SDK): ~250ms · 2**attempt, ±20% jitter, @@ -47,6 +47,23 @@ _DEFAULT_RETRIES = 3 +# A 429 with this problem code is an account limit (active or scheduled +# jobs, or an open job's task cap), not rate limiting: it clears only when +# a job finishes, is closed or is deleted, so a retry only adds delay. +_NOT_RETRYABLE_429_CODES = frozenset({"quota_exceeded"}) + + +def _is_account_limit(response: httpx.Response) -> bool: + """True for a 429 whose problem body names an account limit.""" + if response.status_code != 429: + return False + try: + body = json.loads(response.content or b"{}") + except (json.JSONDecodeError, UnicodeDecodeError): + return False + return isinstance(body, dict) and body.get("code") in _NOT_RETRYABLE_429_CODES + + def _has_idempotency_key(headers: dict[str, str] | None) -> bool: return bool(headers) and any(k.lower() == "idempotency-key" for k in headers) @@ -135,7 +152,9 @@ def _send( 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. + are never retried: the caller set that budget. A 429 + `quota_exceeded` is an account limit, not rate limiting, and is + returned at once. """ idempotent = method.upper() in _IDEMPOTENT_METHODS or ( method.upper() == "POST" and _has_idempotency_key(headers) @@ -157,7 +176,12 @@ def _send( continue raise - if idempotent and attempt < self._retries and response.status_code in retry_statuses: + if ( + idempotent + and attempt < self._retries + and response.status_code in retry_statuses + and not _is_account_limit(response) + ): wait_ms = _retry_after_ms(response) or _backoff_ms(attempt) response.close() time.sleep(wait_ms / 1000) diff --git a/src/zenrows/batch/_waiters.py b/src/zenrows/batch/_waiters.py index e75ba27..1b6e3cc 100644 --- a/src/zenrows/batch/_waiters.py +++ b/src/zenrows/batch/_waiters.py @@ -15,7 +15,10 @@ import random import time from collections.abc import Callable -from typing import TypeVar +from typing import TYPE_CHECKING, TypeVar + +if TYPE_CHECKING: + from zenrows.batch.models import Run T = TypeVar("T") @@ -28,6 +31,23 @@ class WaiterError(RuntimeError): """Raised when the resource entered a `failure_states` value.""" +class RunPaused(WaiterError): + """Raised by a run waiter when the run is paused + (`pause_state: paused`) and not yet in a target status. + + A paused run makes no progress until someone resumes it, so the + waiter stops at once instead of polling until its timeout. `run` is + the run as last read. Resume it with `job.run.resume()` and wait + again, or pass `wait_while_paused=True` to keep polling. + """ + + def __init__(self, run: "Run"): + self.run = run + super().__init__( + f"waiter: run {run.run_id} is paused; resume it with job.run.resume() and wait again" + ) + + def poll_until( fetch: Callable[[], T], *, diff --git a/src/zenrows/batch/_webhooks.py b/src/zenrows/batch/_webhooks.py new file mode 100644 index 0000000..d2c2fdb --- /dev/null +++ b/src/zenrows/batch/_webhooks.py @@ -0,0 +1,137 @@ +"""Verify the signature on a Batch webhook delivery. + +When a job's webhook has `signature: true`, every delivery carries an +`X-Signature` header shaped `t=,v1=,kid=`. +The digest is HMAC-SHA256 over the bytes `.`, keyed +with your signing secret. The API returns that secret once, base64-encoded, +from `rotate_hmac_key()`; `verify_webhook_signature` decodes it for you. + + from zenrows.batch import WEBHOOK_SIGNATURE_HEADER, verify_webhook_signature + + ok = verify_webhook_signature( + request.body, # raw bytes, before any JSON parsing + request.headers[WEBHOOK_SIGNATURE_HEADER], + {"": ""}, # every key you may still receive + ) + +During a key rotation, keep both the old and the new secret in the +mapping: the header's `kid` picks the one that signed. Deliveries can +repeat, so deduplicate on the `X-ZenRows-Event-Id` header +(`WEBHOOK_EVENT_ID_HEADER`). +""" + +import base64 +import binascii +import hashlib +import hmac +import math +import re +import time +from collections.abc import Mapping + +WEBHOOK_SIGNATURE_HEADER = "X-Signature" +"""The header that carries a signed delivery's signature.""" + +WEBHOOK_EVENT_ID_HEADER = "X-ZenRows-Event-Id" +"""The header that carries a delivery's event id. Deduplicate on it.""" + +DEFAULT_WEBHOOK_TOLERANCE_SECONDS = 300 +"""Seconds a signature's timestamp may be from the current time.""" + +# The API issues signing secrets of exactly 32 random bytes. Refusing any +# other length also refuses an empty key, which anyone could sign with. +_SECRET_BYTES = 32 + +# A Unix timestamp in seconds, bounded so the arithmetic can't overflow, +# and a hex HMAC-SHA256 digest (64 characters). +_TIMESTAMP = re.compile(r"[0-9]{1,12}", re.ASCII) +_DIGEST = re.compile(r"[0-9a-fA-F]{64}", re.ASCII) + + +def verify_webhook_signature( + raw_body: bytes | str, + signature_header: str, + secrets: Mapping[str, str], + *, + tolerance_seconds: float = DEFAULT_WEBHOOK_TOLERANCE_SECONDS, + now: float | None = None, +) -> bool: + """Return True when `signature_header` is a valid signature of `raw_body`. + + - `raw_body`: the request body exactly as received (a `str` is + encoded as utf-8). Verify before parsing the JSON. + - `signature_header`: the value of the `X-Signature` header. + - `secrets`: maps each key id (`kid`) to its secret as the API + returned it: a base64 string that decodes to 32 bytes. Any other + secret, including an empty one, never verifies. + - `tolerance_seconds`: the largest accepted gap, in seconds, between the + signature's timestamp and `now`, in either direction. `0` turns + the check off; a negative value rejects every delivery. + - `now`: the current Unix time in seconds (default: `time.time()`). + + The digests are compared in constant time. Returns False, and never + raises, for a malformed header, an unknown `kid`, a bad secret, a + wrong digest or a timestamp out of range. + """ + if not isinstance(signature_header, str) or not isinstance(secrets, Mapping): + return False + if not isinstance(raw_body, (bytes, bytearray, memoryview, str)): + return False + if not _finite_number(tolerance_seconds) or tolerance_seconds < 0: + return False + if now is not None and not _finite_number(now): + return False + + parts: dict[str, str] = {} + for item in signature_header.split(","): + key, sep, value = item.strip().partition("=") + if sep: + parts[key] = value + t, digest, kid = parts.get("t"), parts.get("v1"), parts.get("kid") + if not t or not digest or not kid: + return False + if not _TIMESTAMP.fullmatch(t) or not _DIGEST.fullmatch(digest): + return False + + try: + secret_b64 = secrets.get(kid) + except Exception: + return False + if not isinstance(secret_b64, str): + return False + try: + key = base64.b64decode(secret_b64, validate=True) + except (ValueError, binascii.Error): + return False + if len(key) != _SECRET_BYTES: + return False + + if tolerance_seconds > 0: + current = time.time() if now is None else now + if abs(current - int(t)) > tolerance_seconds: + return False + + try: + body = raw_body.encode("utf-8") if isinstance(raw_body, str) else bytes(raw_body) + except (TypeError, ValueError, UnicodeError): + return False + expected = hmac.new(key, t.encode("ascii") + b"." + body, hashlib.sha256).hexdigest() + return hmac.compare_digest(expected, digest.lower()) + + +def _finite_number(value: object) -> bool: + """A real int or float (not a bool) that is finite.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return False + try: + return math.isfinite(value) + except OverflowError: + return False + + +__all__ = [ + "DEFAULT_WEBHOOK_TOLERANCE_SECONDS", + "WEBHOOK_EVENT_ID_HEADER", + "WEBHOOK_SIGNATURE_HEADER", + "verify_webhook_signature", +] diff --git a/src/zenrows/batch/client.py b/src/zenrows/batch/client.py index af09969..2e975e1 100644 --- a/src/zenrows/batch/client.py +++ b/src/zenrows/batch/client.py @@ -65,7 +65,7 @@ TaskInputDict, WebhookDict, ) -from zenrows.batch._waiters import poll_until +from zenrows.batch._waiters import RunPaused, poll_until from zenrows.batch.errors import BatchAPIError from zenrows.batch.models import ( AddTasksRequest, @@ -89,6 +89,7 @@ ListJobRunsResponse, ListJobsResponse, ListResultsResponse, + PauseState, RerunJobResponse, Run, RunStatus, @@ -334,6 +335,10 @@ def submit_scheduled( ) -> JobRef: """Submit a scheduled job. + Creating a scheduled job starts no run (the first fire creates + Run 1), so there is nothing to ingest and no `wait_for_ingest` + option. + `schedule` is one of the typed builders (`At`, `Rate`, `Calendar`) or a raw `JobScheduleDict` for power users. The typed builders validate their inputs in `__post_init__`; @@ -572,6 +577,7 @@ def wait_for_run( poll_interval: float = 2.0, max_poll_interval: float = 15.0, progress: bool = False, + wait_while_paused: bool = False, ) -> Run: """Block until a run reaches one of `target_statuses`, polling with jittered exponential backoff. @@ -582,9 +588,14 @@ def wait_for_run( until `timeout`. Pass `failure_statuses={"failed"}` to have the waiter raise `WaiterError` on it instead. + A paused run (`pause_state: paused`) never progresses until + someone resumes it, so the waiter raises `RunPaused` (a + `WaiterError`, carrying the run) as soon as it sees one that is + not yet in a target status. Pass `wait_while_paused=True` to + keep polling instead, for example when another process will + resume it. + `progress=True` shows a tqdm bar with totals as they advance. - `None` inherits the client-level `progress` setting (which - itself defaults to off unless `ZENROWS_BATCH_PROGRESS=true`). """ return self._wait_for_run_raw( job_id, @@ -595,6 +606,7 @@ def wait_for_run( poll_interval=poll_interval, max_poll_interval=max_poll_interval, progress=progress, + wait_while_paused=wait_while_paused, ) # ===== file inputs (CSV uploads) ===== @@ -720,9 +732,9 @@ def start_results_export(self, job_id: str, run_id: str) -> ExportRef: `.wait()` for the download URL. Failure modes: 404 if the job/run is missing or not yours. - The worker fails the export (status=failed, - error="results are larger then 1 gb") if the pre-zip total - exceeds the server's 1 GiB cap. + The server fails the export (status=failed, + error="results are larger than N gb") if the run's results + total more than the export cap (100 GiB). """ resp = self._post_export_start(job_id, run_id) return ExportRef(self, job_id, run_id, resp.export_id, start_response=resp) @@ -750,7 +762,7 @@ def wait_for_export( check `.status` and (on `completed`) `.download_url`. `failed` is NOT raised; the caller decides whether - `error="results are larger then 1 gb"` is fatal or expected. + `error="results are larger than N gb"` is fatal or expected. """ return self._wait_for_export_raw( job_id, @@ -835,10 +847,17 @@ def _post_schedule_state(self, job_id: str, state: str) -> Job: Job, ) - def _post_tasks(self, job_id: str, body: AddTasksRequest | AddTasksDict) -> AddTasksResponse: + def _post_tasks( + self, + job_id: str, + body: AddTasksRequest | AddTasksDict, + *, + idempotency_key: str | None = None, + ) -> AddTasksResponse: body = _as_model(body, AddTasksRequest) + headers = {"Idempotency-Key": idempotency_key} if idempotency_key else None return _parse( - self._t.request_json("POST", f"/jobs/{job_id}/tasks", body=body), + self._t.request_json("POST", f"/jobs/{job_id}/tasks", body=body, headers=headers), AddTasksResponse, ) @@ -930,9 +949,9 @@ def download_all_results( Raises `WaiterTimeout` if the export doesn't reach a terminal state within `wait_timeout`. Raises `BatchAPIError` with the server's error message on `status=failed` (e.g. - `"results are larger then 1 gb"`). Returns the `Path` written. + `"results are larger than N gb"`). Returns the `Path` written. - The server-side export is capped at 1 GiB per run. For larger + The server caps one export at 100 GiB of results. For larger runs (or one file per task), use `download_to_dir` instead: it fetches each body client-side with no size limit, at the cost of being slower (one body at a time, tunable via `concurrency=`). @@ -984,6 +1003,7 @@ def _wait_for_run_raw( poll_interval: float, max_poll_interval: float = 15.0, progress: bool = False, + wait_while_paused: bool = False, ) -> Run: target = target_statuses or TERMINAL_RUN_STATUSES if failure_statuses: @@ -1015,6 +1035,14 @@ def is_failure(run: Run) -> bool: def fetch_with_progress() -> Run: run = fetch() on_poll(run) + if ( + not wait_while_paused + and not isinstance(run, _PendingRun) + and run.pause_state is PauseState.PAUSED + and not is_done(run) + and not is_failure(run) + ): + raise RunPaused(run) return run return poll_until( @@ -1026,6 +1054,29 @@ def fetch_with_progress() -> Run: max_interval=max_poll_interval, ) + def _wait_for_run_ingest_raw( + self, + job_id: str, + run_id: str, + *, + timeout: float, + poll_interval: float, + max_poll_interval: float = 15.0, + ) -> Run: + """Poll one run (`GET /jobs/{id}/runs/{run_id}`) until its + `ingest_status` is no longer `pending`, and return it. The run + status is not a shortcut: a run stopped mid-ingest keeps + `ingest_status: pending` until the ingest worker reaches its + next chunk, notices the stop, cuts `stats.total` to the rows + written and marks ingestion done.""" + return poll_until( + fetch=lambda: self._get_run_data(job_id, run_id), + is_done=lambda run: run.ingest_status is not IngestStatus.PENDING, + timeout=timeout, + initial_interval=poll_interval, + max_interval=max_poll_interval, + ) + def _wait_for_ingest_raw( self, job_id: str, @@ -1036,17 +1087,18 @@ def _wait_for_ingest_raw( ) -> Job: """Poll `GET /jobs/{id}` until the latest run's async-carrier ingestion is no longer `pending`. Done means: `ingest_status` - is `done` or absent (the run never ingested asynchronously), - the run is terminal (a mid-ingest stop flips the field to - `done` — SPEC §3.1), or there is no run yet (nothing is - ingesting). Returns the final `Job` snapshot.""" + is `done` or absent (the run never ingested asynchronously), or + there is no run yet (nothing is ingesting). A terminal status + is not enough: a run stopped mid-ingest keeps `pending` until + the ingest worker settles its rows and marks it done. Returns + the final `Job` snapshot.""" def fetch() -> Job: return self._get_job_data(job_id) def is_done(job: Job) -> bool: run = job.latest_run - if run is None or run.status.value in TERMINAL_RUN_STATUSES: + if run is None: return True return run.ingest_status is not IngestStatus.PENDING @@ -1071,6 +1123,7 @@ def _download_dir( progress: bool, max_files: int | None, max_bytes_per_file: int | None, + on_skip: Callable[[TaskResult], None] | None = None, ) -> int: return download_to_dir( self, @@ -1086,6 +1139,7 @@ def _download_dir( max_bytes_per_file=( max_bytes_per_file if max_bytes_per_file is not None else DEFAULT_MAX_BYTES_PER_FILE ), + on_skip=on_skip, ) def _download_memory( @@ -1099,6 +1153,7 @@ def _download_memory( max_count: int | None, max_total_bytes: int | None, max_bytes_per_file: int | None, + on_skip: Callable[[TaskResult], None] | None = None, ) -> list[DownloadedResult]: return download_to_memory( self, @@ -1116,6 +1171,7 @@ def _download_memory( max_bytes_per_file=( max_bytes_per_file if max_bytes_per_file is not None else DEFAULT_MAX_BYTES_PER_FILE ), + on_skip=on_skip, ) diff --git a/src/zenrows/batch/errors.py b/src/zenrows/batch/errors.py index 940a2be..19d11a9 100644 --- a/src/zenrows/batch/errors.py +++ b/src/zenrows/batch/errors.py @@ -8,6 +8,7 @@ import json from dataclasses import dataclass +from enum import Enum from typing import Any import httpx @@ -55,13 +56,117 @@ def from_response(cls, response: httpx.Response) -> "ProblemDetail | None": ) +class BatchErrorCode(str, Enum): + """The stable `code` values the Batch API sends in its problem+json + errors, and in the `error` of a failed task. + + Members are strings, so they compare equal to `BatchAPIError.code`: + + except BatchAPIError as exc: + if exc.code == BatchErrorCode.QUOTA_EXCEEDED: + ... + + The set is open: the API may add codes, and `BatchAPIError.code` + stays a plain `str`, so keep a fallback branch. + """ + + # Generic + INVALID_ARGUMENT = "invalid_argument" + """400: the request failed validation. `extras["invalid_tasks"]` + lists per-task reasons when the problem is in `tasks`.""" + NOT_FOUND = "not_found" + """404: no such resource, or it is not yours.""" + CONFLICT = "conflict" + """409: the request conflicts with the resource's current state.""" + UNAUTHENTICATED = "unauthenticated" + """401: missing or invalid API key.""" + SUBSCRIPTION_INACTIVE = "subscription_inactive" + """401: the key is valid, but the subscription is not active + (past due, unpaid, paused). Fix it in Billing.""" + PAYMENT_REQUIRED = "payment_required" + """402: the account is out of credits.""" + API_KEY_CAP_REACHED = "api_key_cap_reached" + """402: this API key reached one of its credit caps. `detail` says + which cap, and when it resets.""" + PERMISSION_DENIED = "permission_denied" + """403: the account may not use this feature.""" + QUOTA_EXCEEDED = "quota_exceeded" + """429: an account limit was reached: active jobs (3, shared with + Crawl), scheduled jobs (5) or an open job's task cap (100,000). Not + transient; the client does not retry it.""" + INTERNAL = "internal" + """500: an unexpected server error. Also the `code` of a + `BatchAPIError` whose response had no problem body.""" + NOT_IMPLEMENTED = "not_implemented" + """501: the feature is not available.""" + + # Job, run and task lifecycle (409 unless noted) + JOB_NOT_OPEN = "job_not_open" + RUN_NOT_TERMINAL = "run_not_terminal" + """The previous run is still pending or running (for example on a + rerun), or a large submit is still ingesting (on add-tasks).""" + RUN_NOT_STOPPABLE = "run_not_stoppable" + NO_PRIOR_RUN = "no_prior_run" + NO_MATCHING_TASKS = "no_matching_tasks" + """A partial rerun's `status` filter matched no task.""" + RUN_HAS_DESCENDANTS = "run_has_descendants" + LAST_BATCH_RECEIVED = "last_batch_received" + IDEMPOTENCY_KEY_CONFLICT = "idempotency_key_conflict" + """The `Idempotency-Key` was used with a different body, or the + first call with it is still running.""" + TASK_NOT_READY = "task_not_ready" + PARTIAL_RERUN_WINDOW_EXPIRED = "partial_rerun_window_expired" + """A partial rerun (`status=`) more than 10 days after the run it + retries started. A full rerun still works.""" + RESULT_EXPIRED = "result_expired" + """410: the result is past the job's retention window (14 days by + default).""" + + # HMAC key rotation (409) + CANDIDATE_PENDING = "candidate_pending" + NO_CANDIDATE = "no_candidate" + + # Task input validation (400, often per task in `invalid_tasks`) + UNSUPPORTED_SCHEME = "unsupported_scheme" + MALFORMED_URL = "malformed_url" + MISSING_HOST = "missing_host" + URL_TOO_LONG = "url_too_long" + METADATA_TOO_LARGE = "metadata_too_large" + METADATA_KEY_INVALID = "metadata_key_invalid" + INVALID_EXTERNAL_ID = "invalid_external_id" + UNSUPPORTED_METHOD = "unsupported_method" + BODY_NOT_ALLOWED = "body_not_allowed" + BODY_TOO_LARGE = "body_too_large" + BODY_MALFORMED = "body_malformed" + METHOD_PARAM_CONFLICT = "method_param_conflict" + + # CSV file inputs + FILE_INPUT_NOT_FOUND = "file_input_not_found" + """404: no such file input, or its 24 h slot expired.""" + FILE_INPUT_NOT_UPLOADED = "file_input_not_uploaded" + """409: the CSV was never uploaded to the slot.""" + FILE_INPUT_UNSUPPORTED_FOR_JOB_TYPE = "file_input_unsupported_for_job_type" + """409: CSV input works only for closed regular jobs and scheduled + jobs.""" + + # Webhook registration (400) + CALLBACK_URL_MUST_BE_HTTPS = "callback_url_must_be_https" + CALLBACK_URL_UNRESOLVABLE = "callback_url_unresolvable" + WEBHOOK_SIGNING_REQUIRES_ACTIVE_KEY = "webhook_signing_requires_active_key" + + # Task level (in a failed task's `error`) + GATEWAY_UNREACHABLE = "gateway_unreachable" + """The scraping API could not be reached for this task.""" + + class BatchAPIError(Exception): """A non-2xx response from the Batch API. `code` is the RFC 7807 `code` member (e.g. `file_input_not_found`, `idempotency_key_conflict`, `api_key_cap_reached`). Stable; safe to - branch on. `detail` is the human-readable explanation (e.g. which - credit cap was reached and when it resets): display it, don't parse it. + branch on; `BatchErrorCode` names every code the API sends. `detail` + is the human-readable explanation (e.g. which credit cap was reached + and when it resets): display it, don't parse it. """ def __init__(self, status_code: int, problem: ProblemDetail | None, raw: bytes): diff --git a/src/zenrows/batch/models.py b/src/zenrows/batch/models.py index 78d2690..f824eba 100644 --- a/src/zenrows/batch/models.py +++ b/src/zenrows/batch/models.py @@ -1,6 +1,6 @@ # generated by datamodel-codegen: # filename: openapi.yaml -# timestamp: 2026-09-30T14:31:48+00:00 +# timestamp: 2026-10-16T10:33:08+00:00 from __future__ import annotations from zenrows.batch._open_enum import open_enum_missing @@ -124,13 +124,21 @@ class RunStatus(Enum): class TaskStatus(Enum): """ + - `expired`: a successful task whose result is past the job's + retention window (14 days by default). The row has no + `result_url`, and the content endpoint answers + `410 result_expired`. A `status=successful` filter still + returns these rows. + Clients must accept values not listed here. + """ PENDING = "pending" PROCESSING = "processing" SUCCESSFUL = "successful" FAILED = "failed" + EXPIRED = "expired" # Open enum (x-extensible-enum): unknown values parse as UNKNOWN. _missing_ = classmethod(open_enum_missing) @@ -262,7 +270,7 @@ class Status(Enum): class AddTasksRequest(BaseModel): - tasks: Annotated[list[TaskInput], Field(max_length=1000, min_length=1)] + tasks: Annotated[list[TaskInput], Field(max_length=10000, min_length=1)] last_batch: bool | None = False """ Set true on the final batch. Closes the job (status → @@ -286,11 +294,9 @@ class Spend(BaseModel): class TaskSpend(BaseModel): """ - Per-task indicative spend with two roll-ups: `total` - accumulates across every attempt (including retries), - `last_attempt` carries just the most recent gateway call. - On a task that succeeded on its first try the two are - equal; on a retried task they diverge. + Deprecated: result rows never carry a per-task spend, so this + model is never filled. Kept so existing imports keep working. + Read the run-level `stats.spend` instead. """ @@ -321,8 +327,10 @@ class RunStats(BaseModel): Vocabulary (the only keys that appear): - `auth_failed` — credentials or billing - - `blocked` — anti-bot / policy denials - - `bad_target` — target URL is the problem (bad host, 404, 410, too large) + - `blocked`: anti-bot / policy denials, including a bot wall + Extract met on the fetched page + - `bad_target`: target URL is the problem (bad host, 404, 410, + too large, or a page Extract does not cover or found no data on) - `rate_limited` — target throttled the request - `timeout` — the scrape didn't complete in time - `gateway_error` — Zenrows-side transport / 5xx @@ -401,6 +409,7 @@ class Run(BaseModel): job_id: str run_sequence: Annotated[int, Field(ge=1)] status: RunStatus + trigger: RunTrigger | None = None stats: RunStats last_batch_received: bool | None = None """ @@ -904,8 +913,9 @@ class Problem(BaseModel): detail: str | None = None code: str """ - Stable machine-readable error code, e.g. `payment_required` - or `api_key_cap_reached`. Branch on this, not on `detail`. + Stable machine-readable error code, e.g. `payment_required`, + `api_key_cap_reached` or `subscription_inactive`. Branch on + this, not on `detail`. Clients must accept values not listed here. """ @@ -970,7 +980,7 @@ class Export(BaseModel): """ Non-empty only when `status = failed`. Stable strings — e.g. `"results are larger than N gb"` when the combined - results exceed the export byte cap (20 GiB). + results exceed the export byte cap (100 GiB). """ download_url: AnyUrl | None = None @@ -1078,9 +1088,11 @@ class TaskResult(BaseModel): type: ResultType | None = None result_url: str | None = None """ - 24-hour presigned download URL for the result body, or a - `/v1/jobs//runs//tasks//content` URL you can - fetch directly. Empty for non-successful tasks. + Presigned download URL for the result body, valid for 2 + hours; list the results again for a fresh one. Empty for + failed and `expired` tasks. The body is also served by + `GET /v1/jobs/{job_id}/runs/{run_id}/tasks/{task_id}/content`, + which needs your API key. """ error: Problem | None = None @@ -1101,6 +1113,12 @@ class TaskResult(BaseModel): """ spend: TaskSpend | None = None + """ + Deprecated: result rows never carry a per-task spend, so + this is always empty. Read the run-level `stats.spend` + instead. + + """ created_at: AwareDatetime """ When the task was created. @@ -1269,11 +1287,11 @@ class SubmitJobRequest(BaseModel): ignored otherwise. """ - tasks: Annotated[list[TaskInput] | None, Field(max_length=1000, min_length=0)] = ( + tasks: Annotated[list[TaskInput] | None, Field(max_length=100000, min_length=0)] = ( None ) """ - Required for closed jobs (1–1000) unless `file_input_id` + Required for closed jobs (1 to 100,000) unless `file_input_id` is provided. Optional for open jobs — may be empty if the caller will follow up with `addTasks`. Mutually exclusive with `file_input_id`. diff --git a/tests/test_batch_client.py b/tests/test_batch_client.py index 09d1f21..915e487 100644 --- a/tests/test_batch_client.py +++ b/tests/test_batch_client.py @@ -690,7 +690,7 @@ def _ingest_job_json(ingest_status: str | None) -> dict: @respx.mock def test_submit_wait_for_ingest_polls_202_until_done(client: ZenRowsBatchClient): """A 202 submit body carries `ingest_status: pending`; the flag - polls GET /jobs/{id} until it leaves `pending`.""" + polls the submitted run until it leaves `pending`, then reads the job.""" respx.post(f"{BASE_URL}/jobs").mock( return_value=Response( 202, @@ -702,12 +702,16 @@ def test_submit_wait_for_ingest_polls_202_until_done(client: ZenRowsBatchClient) }, ) ) + run_route = respx.get(f"{BASE_URL}/jobs/J/runs/R").mock( + return_value=Response(200, json=_ingest_run_json("done")) + ) get_route = respx.get(f"{BASE_URL}/jobs/J").mock( return_value=Response(200, json=_ingest_job_json("done")) ) job = client.submit_regular(["https://example.com/a"], wait_for_ingest=True) + assert run_route.call_count == 1 assert get_route.call_count == 1 assert job.data.latest_run.ingest_status is IngestStatus.DONE @@ -1178,3 +1182,462 @@ def test_submit_job_post_task_method_body_on_wire(client: ZenRowsBatchClient): assert sent["tasks"][0]["body"] == {"query": "{ products { id } }"} assert "method" not in sent["tasks"][1] assert "body" not in sent["tasks"][1] + + +# ===== expired results, add-tasks idempotency, account limits ===== + + +def _results_page(rows: list[dict]) -> dict: + base = { + "run_id": "R", + "url": "https://example.com", + "created_at": "2026-08-25T12:00:00Z", + "updated_at": "2026-08-25T12:00:05Z", + } + return {"results": [{**base, **r} for r in rows]} + + +def test_task_status_expired_and_run_trigger_parse(): + """`expired` is a known TaskStatus and `Run.trigger` is typed.""" + from zenrows.batch import Run, TaskStatus + from zenrows.batch.models import RunTrigger + + row = _task_result("T1", status="expired") + assert row.status is TaskStatus.EXPIRED + assert row.result_url is None + assert row.spend is None # result rows carry no per-task spend + + run = Run.model_validate( + { + "run_id": "R", + "job_id": "J", + "run_sequence": 2, + "status": "completed", + "trigger": "scheduled", + "stats": {"total": 1, "completed": 1, "successful": 1, "failed": 0}, + "created_at": "2026-08-25T12:00:00Z", + "updated_at": "2026-08-25T12:00:05Z", + } + ) + assert run.trigger is RunTrigger.SCHEDULED + + +@respx.mock +def test_download_to_dir_skips_expired_and_failed_rows( + client: ZenRowsBatchClient, tmp_path, caplog +): + """Rows with no body (expired, failed) are skipped and reported, not + fatal; the rest download.""" + respx.get(f"{BASE_URL}/jobs/J/runs/R/results").mock( + return_value=Response( + 200, + json=_results_page( + [ + { + "task_id": "T1", + "status": "successful", + "type": "html", + "result_url": "https://storage.example.test/T1", + }, + {"task_id": "T2", "status": "expired", "type": "html"}, + {"task_id": "T3", "status": "failed"}, + {"task_id": "T4", "status": "expired"}, + ] + ), + ) + ) + respx.get("https://storage.example.test/T1").mock(return_value=Response(200, content=b"ok")) + + skipped: list[TaskResult] = [] + with caplog.at_level("WARNING", logger="zenrows.batch.download"): + n = client.run("J", "R").download_to_dir(tmp_path, status=None, on_skip=skipped.append) + + assert n == 1 + assert (tmp_path / "T1.html").read_bytes() == b"ok" + assert [r.task_id for r in skipped] == ["T2", "T3", "T4"] + assert "skipped 3 result(s)" in caplog.text + assert "2 expired" in caplog.text and "1 failed" in caplog.text + + +@respx.mock +def test_download_to_memory_skips_expired_rows_concurrently(client: ZenRowsBatchClient): + """The parallel path skips the same rows, and they don't count toward + `max_count`.""" + rows = [ + { + "task_id": f"T{i}", + "status": "successful", + "result_url": f"https://storage.example.test/T{i}", + } + for i in range(3) + ] + [{"task_id": f"E{i}", "status": "expired"} for i in range(5)] + respx.get(f"{BASE_URL}/jobs/J/runs/R/results").mock( + return_value=Response(200, json=_results_page(rows)) + ) + respx.get(url__startswith="https://storage.example.test/").mock( + return_value=Response(200, content=b"x") + ) + + skipped: list[TaskResult] = [] + out = client.run("J", "R").download_to_memory( + concurrency=4, max_count=3, on_skip=skipped.append + ) + + assert sorted(r.task_id for r in out) == ["T0", "T1", "T2"] + assert len(skipped) == 5 + + +@respx.mock +def test_download_successful_row_without_url_still_raises(client: ZenRowsBatchClient, tmp_path): + """A `successful` row with no `result_url` is not a row without a + body, so it is not skipped silently.""" + respx.get(f"{BASE_URL}/jobs/J/runs/R/results").mock( + return_value=Response(200, json=_results_page([{"task_id": "T1", "status": "successful"}])) + ) + with pytest.raises(ValueError, match="no result_url"): + client.run("J", "R").download_to_dir(tmp_path) + + +def test_download_task_expired_raises_clear_error(client: ZenRowsBatchClient): + task = _task_result("T1", status="expired") + with pytest.raises(ValueError, match="has expired"): + client.run("J", "R").download_task_to_memory(task) + + +def _add_tasks_ok() -> Response: + return Response( + 200, json={"accepted_tasks": 1, "job_status": "open", "latest_run": _run_json()} + ) + + +@respx.mock +def test_add_tasks_sends_idempotency_key_and_retries_transient( + client: ZenRowsBatchClient, no_sleep +): + """With a key, add-tasks sends `Idempotency-Key` and a 503 is retried + with the same key.""" + route = respx.post(f"{BASE_URL}/jobs/J/tasks").mock( + side_effect=[Response(503, json={}), _add_tasks_ok()] + ) + + resp = client.job("J").add_tasks( + {"tasks": [{"url": "https://example.com/a"}]}, idempotency_key="add-1" + ) + + assert resp.accepted_tasks == 1 + assert route.call_count == 2 + assert [c.request.headers["Idempotency-Key"] for c in route.calls] == ["add-1", "add-1"] + + +@respx.mock +def test_add_tasks_without_key_is_sent_once(client: ZenRowsBatchClient, no_sleep): + route = respx.post(f"{BASE_URL}/jobs/J/tasks").mock(return_value=Response(503, json={})) + + with pytest.raises(BatchAPIError): + client.job("J").add_tasks({"tasks": [{"url": "https://example.com/a"}]}) + + assert route.call_count == 1 + assert "Idempotency-Key" not in route.calls.last.request.headers + + +@respx.mock +def test_quota_exceeded_429_is_not_retried(client: ZenRowsBatchClient, no_sleep): + """A 429 `quota_exceeded` is an account limit: raised at once, with + the named code.""" + from zenrows.batch import BatchErrorCode + + route = respx.post(f"{BASE_URL}/jobs").mock( + return_value=Response( + 429, + json={ + "type": "https://docs.zenrows.com/api-error-codes#quota_exceeded", + "title": "Quota exceeded", + "status": 429, + "code": "quota_exceeded", + }, + ) + ) + + with pytest.raises(BatchAPIError) as exc: + client.submit_regular(["https://a"], idempotency_key="k1") + + assert route.call_count == 1 + assert no_sleep == [] + assert exc.value.code == BatchErrorCode.QUOTA_EXCEEDED + + +@respx.mock +def test_plain_429_is_still_retried(client: ZenRowsBatchClient, no_sleep): + route = respx.get(f"{BASE_URL}/jobs/J").mock( + side_effect=[Response(429, json={}), Response(200, json=_ingest_job_json("done"))] + ) + client.get_job("J") + assert route.call_count == 2 + + +def test_batch_error_code_names_every_api_code(): + """`BatchErrorCode` covers every code the Batch API sends, and its + members compare equal to the plain-string `BatchAPIError.code`.""" + from zenrows.batch import BatchErrorCode + + expected = { + "invalid_argument", "not_found", "conflict", "unauthenticated", + "payment_required", "permission_denied", "quota_exceeded", + "subscription_inactive", "api_key_cap_reached", "internal", + "job_not_open", "run_not_terminal", "run_not_stoppable", "no_prior_run", + "no_matching_tasks", "run_has_descendants", "last_batch_received", + "idempotency_key_conflict", "task_not_ready", "not_implemented", + "result_expired", "partial_rerun_window_expired", "candidate_pending", + "no_candidate", "unsupported_scheme", "malformed_url", "missing_host", + "url_too_long", "metadata_too_large", "metadata_key_invalid", + "invalid_external_id", "unsupported_method", "body_not_allowed", + "body_too_large", "body_malformed", "method_param_conflict", + "file_input_not_found", "file_input_not_uploaded", + "file_input_unsupported_for_job_type", "callback_url_must_be_https", + "callback_url_unresolvable", "webhook_signing_requires_active_key", + "gateway_unreachable", + } # fmt: skip + assert {c.value for c in BatchErrorCode} == expected + assert BatchErrorCode.RESULT_EXPIRED == "result_expired" + err = BatchAPIError(410, None, b"") + err.code = "result_expired" + assert err.code == BatchErrorCode.RESULT_EXPIRED + + +# ===== waiters on a paused run ===== + + +@pytest.fixture +def no_poll_sleep(monkeypatch): + slept: list[float] = [] + monkeypatch.setattr("zenrows.batch._waiters.time.sleep", slept.append) + return slept + + +@respx.mock +def test_wait_raises_run_paused_instead_of_spinning(client: ZenRowsBatchClient, no_poll_sleep): + """A paused run never progresses: the waiter raises `RunPaused` on the + first poll that sees it, carrying the run.""" + from zenrows.batch import RunPaused, WaiterError + + route = respx.get(f"{BASE_URL}/jobs/J/runs/R").mock( + return_value=Response(200, json=_run_json(pause_state="paused")) + ) + + with pytest.raises(RunPaused) as exc: + client.run("J", "R").wait(timeout=600) + + assert isinstance(exc.value, WaiterError) + assert exc.value.run.run_id == "R" + assert route.call_count == 1 + assert no_poll_sleep == [] + + +@respx.mock +def test_current_run_wait_raises_run_paused(client: ZenRowsBatchClient, no_poll_sleep): + from zenrows.batch import RunPaused + + job = _ingest_job_json("done") + job["latest_run"] = _run_json(pause_state="paused") + respx.get(f"{BASE_URL}/jobs/J").mock(return_value=Response(200, json=job)) + + with pytest.raises(RunPaused): + client.job("J").run.wait() + + +@respx.mock +def test_wait_returns_a_paused_run_that_is_already_terminal( + client: ZenRowsBatchClient, no_poll_sleep +): + """A run stopped while paused is terminal: it is returned, not raised.""" + respx.get(f"{BASE_URL}/jobs/J/runs/R").mock( + return_value=Response(200, json=_run_json(status="stopped", pause_state="paused")) + ) + run = client.run("J", "R").wait() + assert run.data.status.value == "stopped" + + +@respx.mock +def test_wait_while_paused_keeps_polling(client: ZenRowsBatchClient, no_poll_sleep): + route = respx.get(f"{BASE_URL}/jobs/J/runs/R").mock( + side_effect=[ + Response(200, json=_run_json(pause_state="paused")), + Response(200, json=_run_json(pause_state="active")), + Response(200, json=_run_json(status="completed")), + ] + ) + + run = client.wait_for_run("J", run_id="R", wait_while_paused=True) + + assert run.status.value == "completed" + assert route.call_count == 3 + + +# ===== wait_for_ingest on rerun / retry_failed / submit_scheduled ===== + + +def _ingesting_rerun(ingest_status: str) -> dict: + body = _rerun_response("R2", total=100, successful=0) + body["latest_run"]["ingest_status"] = ingest_status + return body + + +def _run_with_ingest(ingest_status: str, *, run_id: str = "R2", status: str = "running") -> dict: + return {**_run_json(run_id=run_id, status=status), "ingest_status": ingest_status} + + +@respx.mock +@pytest.mark.parametrize("call", ["rerun", "retry_failed"]) +def test_rerun_wait_for_ingest_polls_the_returned_run_until_done( + client: ZenRowsBatchClient, no_poll_sleep, call: str +): + respx.post(f"{BASE_URL}/jobs/J/rerun").mock( + return_value=Response(202, json=_ingesting_rerun("pending")) + ) + poll = respx.get(f"{BASE_URL}/jobs/J/runs/R2").mock( + side_effect=[ + Response(200, json=_run_with_ingest("pending")), + Response(200, json=_run_with_ingest("done")), + ] + ) + + run = getattr(client.job("J"), call)(wait_for_ingest=True) + + assert poll.call_count == 2 + assert run.run_id == "R2" + assert run.data.ingest_status is IngestStatus.DONE + + +@respx.mock +def test_rerun_wait_for_ingest_follows_a_replayed_older_run( + client: ZenRowsBatchClient, no_poll_sleep +): + """An idempotent replay can answer with an older run (R2) while a + newer run (R3) is the job's latest: the wait follows R2, not the job.""" + respx.post(f"{BASE_URL}/jobs/J/rerun").mock( + return_value=Response(202, json=_ingesting_rerun("pending")) + ) + job_poll = respx.get(f"{BASE_URL}/jobs/J").mock( + return_value=Response(200, json={**_ingest_job_json("pending")}) + ) + run_poll = respx.get(f"{BASE_URL}/jobs/J/runs/R2").mock( + return_value=Response(200, json=_run_with_ingest("done", status="completed")) + ) + + run = client.job("J").rerun(idempotency_key="k1", wait_for_ingest=True) + + assert job_poll.call_count == 0 + assert run_poll.call_count == 1 + assert run.data.status.value == "completed" + assert run.data.ingest_status is IngestStatus.DONE + + +@respx.mock +def test_rerun_wait_for_ingest_does_not_stop_at_a_stopped_run( + client: ZenRowsBatchClient, no_poll_sleep +): + """A run stopped mid-ingest stays `ingest_status: pending` until the + ingest worker winds down: the wait keeps polling until it is done.""" + respx.post(f"{BASE_URL}/jobs/J/rerun").mock( + return_value=Response(202, json=_ingesting_rerun("pending")) + ) + poll = respx.get(f"{BASE_URL}/jobs/J/runs/R2").mock( + side_effect=[ + Response(200, json=_run_with_ingest("pending", status="stopped")), + Response(200, json=_run_with_ingest("done", status="stopped")), + ] + ) + + run = client.job("J").rerun(wait_for_ingest=True) + + assert poll.call_count == 2 + assert run.data.status.value == "stopped" + assert run.data.ingest_status is IngestStatus.DONE + + +@respx.mock +def test_rerun_without_wait_for_ingest_returns_at_once(client: ZenRowsBatchClient): + respx.post(f"{BASE_URL}/jobs/J/rerun").mock( + return_value=Response(202, json=_ingesting_rerun("pending")) + ) + poll = respx.get(url__startswith=f"{BASE_URL}/jobs/J") + + run = client.job("J").rerun() + + assert poll.call_count == 0 + assert run.data.ingest_status is IngestStatus.PENDING + + +@respx.mock +def test_submit_scheduled_returns_after_one_request(client: ZenRowsBatchClient, no_poll_sleep): + """Creating a scheduled job answers 201 with no run (the first fire + creates it), so there is nothing to poll and no wait option.""" + from zenrows.batch import Rate + + respx.post(f"{BASE_URL}/jobs").mock( + return_value=Response(201, json={"job_id": "J", "status": "closed", "accepted_tasks": 1}) + ) + poll = respx.get(url__startswith=f"{BASE_URL}/jobs/J") + + ref = client.submit_scheduled(Rate(every=6, unit="hour"), ["https://example.com"]) + + assert poll.call_count == 0 + assert ref.job_id == "J" + assert ref.submit_response.latest_run is None + + +# ===== job-level wait_for_ingest ===== + + +@respx.mock +def test_job_wait_for_ingest_does_not_stop_at_a_stopped_run( + client: ZenRowsBatchClient, no_poll_sleep +): + """A run stopped mid-ingest is terminal but still `pending`: the + job-level wait keeps polling until ingestion is done.""" + stopped_pending = _ingest_job_json("pending") + stopped_pending["latest_run"]["status"] = "stopped" + stopped_done = _ingest_job_json("done") + stopped_done["latest_run"]["status"] = "stopped" + route = respx.get(f"{BASE_URL}/jobs/J").mock( + side_effect=[Response(200, json=stopped_pending), Response(200, json=stopped_done)] + ) + + out = client.job("J").wait_for_ingest() + + assert route.call_count == 2 + assert out.data.latest_run.ingest_status is IngestStatus.DONE + + +@respx.mock +def test_submit_wait_for_ingest_polls_the_submitted_run(client: ZenRowsBatchClient, no_poll_sleep): + """A replayed submit can name an older run while a newer one is the + job's latest: the wait polls the submitted run by id.""" + respx.post(f"{BASE_URL}/jobs").mock( + return_value=Response( + 202, + json={ + "job_id": "J", + "status": "closed", + "accepted_tasks": 10000, + "latest_run": _ingest_run_json("pending"), + }, + ) + ) + run_route = respx.get(f"{BASE_URL}/jobs/J/runs/R").mock( + side_effect=[ + Response(200, json=_ingest_run_json("pending", status="stopped")), + Response(200, json=_ingest_run_json("done", status="stopped")), + ] + ) + newer = _ingest_job_json("pending") + newer["latest_run"]["run_id"] = "R3" + job_route = respx.get(f"{BASE_URL}/jobs/J").mock(return_value=Response(200, json=newer)) + + handle = client.submit_regular( + ["https://example.com/a"], idempotency_key="k1", wait_for_ingest=True + ) + + assert run_route.call_count == 2 + assert job_route.call_count == 1 # one read for the handle's data, not a poll + assert handle.submit_response.latest_run.run_id == "R" diff --git a/tests/test_estimate.py b/tests/test_estimate.py index 659b4da..08da134 100644 --- a/tests/test_estimate.py +++ b/tests/test_estimate.py @@ -35,12 +35,14 @@ def test_cost_for_params_rate_card(params, tier, lo, hi): assert tc.exact == (lo == hi) -@pytest.mark.parametrize("value", [True, "true", "True", " TRUE ", 1, "1", "yes", "on"]) +@pytest.mark.parametrize("value", [True, "true", "True", "TRUE"]) def test_truthy_spellings_turn_on_a_flag(value): assert cost_for_params({"js_render": value}).tier is Tier.JS -@pytest.mark.parametrize("value", [False, "false", "0", 0, "", "no"]) +# The API rejects 1, "1", "yes" and "on" for a boolean flag, so they are +# not priced as on. +@pytest.mark.parametrize("value", [False, "false", "0", 0, "", "no", 1, "1", "yes", "on", " TRUE "]) def test_falsy_spellings_keep_base(value): assert cost_for_params({"js_render": value}).tier is Tier.BASE @@ -159,3 +161,9 @@ def test_client_estimate_cost_file_input_is_zero(): client = ZenRowsBatchClient(api_key="test-key") est = client.estimate_cost({"type": "regular", "status": "closed", "file_input_id": "01HKE..."}) assert (est.task_count, est.min, est.max) == (0, 0, 0) + + +@pytest.mark.parametrize("value", ["AUTO", "Auto", " auto"]) +def test_mode_is_auto_only_for_the_exact_value(value): + """The API accepts only `mode: "auto"`, so other spellings price as base.""" + assert cost_for_params({"mode": value}).tier is Tier.BASE diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py new file mode 100644 index 0000000..84e8c31 --- /dev/null +++ b/tests/test_webhooks.py @@ -0,0 +1,132 @@ +"""`verify_webhook_signature` against signatures computed the way the +Batch API signs deliveries: +`t=,v1=.")>,kid=`, keyed +with the 32 bytes the base64 secret decodes to.""" + +import base64 +import hashlib +import hmac + +import pytest + +from zenrows.batch import ( + DEFAULT_WEBHOOK_TOLERANCE_SECONDS, + WEBHOOK_EVENT_ID_HEADER, + WEBHOOK_SIGNATURE_HEADER, + WebhookEvent, + WebhookEventType, + verify_webhook_signature, +) + +# A fixed vector produced by the API's own signing routine, with the +# secret bytes 0x00..0x1f, kid "kid_live" and t=1790000000. +SECRET = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" +BODY = ( + b'{"schema":"v1","event_id":"01JEVENT0000000000000000000","event_type":"run.completed",' + b'"occurred_at":"2026-10-16T12:00:00Z","job_id":"01JJOB00000000000000000000",' + b'"run_id":"01JRUN00000000000000000000","run_sequence":1}' +) +T = 1790000000 +HEADER = ( + "t=1790000000,v1=4f5ee4f729590ac8b2f37163ecd9f88bf3fb500632fe2a3f509cced90d177c5b,kid=kid_live" +) + + +def _sign(secret_b64: str, body: bytes, t: int, kid: str) -> str: + key = base64.b64decode(secret_b64) + digest = hmac.new(key, f"{t}.".encode() + body, hashlib.sha256).hexdigest() + return f"t={t},v1={digest},kid={kid}" + + +def test_fixture_vector_verifies(): + assert verify_webhook_signature(BODY, HEADER, {"kid_live": SECRET}, now=T) + assert verify_webhook_signature(BODY.decode(), HEADER, {"kid_live": SECRET}, now=T + 299) + + +def test_fixture_matches_local_signer(): + """The vector is plain HMAC over `.` with the decoded key.""" + assert _sign(SECRET, BODY, T, "kid_live") == HEADER + + +def test_the_secret_is_base64_decoded_not_used_as_text(): + """Signing with the base64 text itself (a common mistake) never verifies.""" + digest = hmac.new(SECRET.encode(), f"{T}.".encode() + BODY, hashlib.sha256).hexdigest() + header = f"t={T},v1={digest},kid=kid_live" + assert not verify_webhook_signature(BODY, header, {"kid_live": SECRET}, now=T) + + +def test_tampered_body_or_digest_fails(): + secrets = {"kid_live": SECRET} + assert not verify_webhook_signature(BODY + b" ", HEADER, secrets, now=T) + flipped = HEADER.replace("v1=4f", "v1=5f") + assert not verify_webhook_signature(BODY, flipped, secrets, now=T) + + +def test_kid_selects_the_secret_during_rotation(): + new_secret = base64.b64encode(bytes(range(32, 64))).decode() + secrets = {"kid_old": SECRET, "kid_new": new_secret} + assert verify_webhook_signature(BODY, _sign(SECRET, BODY, T, "kid_old"), secrets, now=T) + assert verify_webhook_signature(BODY, _sign(new_secret, BODY, T, "kid_new"), secrets, now=T) + # Signed with the old key but labelled with the new kid: rejected. + assert not verify_webhook_signature(BODY, _sign(SECRET, BODY, T, "kid_new"), secrets, now=T) + # Unknown kid: rejected. + assert not verify_webhook_signature(BODY, HEADER, {"other": SECRET}, now=T) + + +def test_tolerance(): + secrets = {"kid_live": SECRET} + assert DEFAULT_WEBHOOK_TOLERANCE_SECONDS == 300 + assert verify_webhook_signature(BODY, HEADER, secrets, now=T + 300) + assert verify_webhook_signature(BODY, HEADER, secrets, now=T - 300) + assert not verify_webhook_signature(BODY, HEADER, secrets, now=T + 301) + assert not verify_webhook_signature(BODY, HEADER, secrets, now=T - 301) + assert verify_webhook_signature(BODY, HEADER, secrets, tolerance_seconds=60, now=T + 60) + assert not verify_webhook_signature(BODY, HEADER, secrets, tolerance_seconds=60, now=T + 61) + # 0 turns the check off; a negative tolerance rejects everything. + assert verify_webhook_signature(BODY, HEADER, secrets, tolerance_seconds=0, now=T + 10**6) + assert not verify_webhook_signature(BODY, HEADER, secrets, tolerance_seconds=-1, now=T) + # Default `now` is the real clock, far past T. + assert not verify_webhook_signature(BODY, HEADER, secrets) + + +@pytest.mark.parametrize( + "header", + [ + "", + "garbage", + "t=1790000000,kid=kid_live", + "v1=4f5ee4f729590ac8b2f37163ecd9f88bf3fb500632fe2a3f509cced90d177c5b,kid=kid_live", + "t=1790000000,v1=4f5ee4f729590ac8b2f37163ecd9f88bf3fb500632fe2a3f509cced90d177c5b", + "t=abc,v1=4f5ee4f729590ac8b2f37163ecd9f88bf3fb500632fe2a3f509cced90d177c5b,kid=kid_live", + "t=1790000000,v1=zz,kid=kid_live", + "t=1790000000,v1=4f5e,kid=kid_live", + ], +) +def test_malformed_headers_fail(header: str): + assert not verify_webhook_signature(BODY, header, {"kid_live": SECRET}, now=T) + + +@pytest.mark.parametrize( + "secret", + ["", "not base64!!", base64.b64encode(b"short").decode(), SECRET.rstrip("=")], +) +def test_bad_secrets_never_verify(secret: str): + """An empty, non-base64 or wrong-length secret never verifies, even + against a header signed with that same key.""" + header = f"t={T},v1={hmac.new(b'', f'{T}.'.encode() + BODY, hashlib.sha256).hexdigest()},kid=k" + assert not verify_webhook_signature(BODY, header, {"k": secret}, now=T) + + +def test_wrong_argument_types_return_false(): + assert not verify_webhook_signature(BODY, None, {"kid_live": SECRET}, now=T) # type: ignore[arg-type] + assert not verify_webhook_signature(BODY, HEADER, None, now=T) # type: ignore[arg-type] + assert not verify_webhook_signature(123, HEADER, {"kid_live": SECRET}, now=T) # type: ignore[arg-type] + assert not verify_webhook_signature(BODY, HEADER, {"kid_live": SECRET}, now=float("nan")) + + +def test_header_names_and_payload_model(): + assert WEBHOOK_SIGNATURE_HEADER == "X-Signature" + assert WEBHOOK_EVENT_ID_HEADER == "X-ZenRows-Event-Id" + event = WebhookEvent.model_validate_json(BODY) + assert event.event_type is WebhookEventType.RUN_COMPLETED + assert event.run_sequence == 1