diff --git a/README.md b/README.md index 6fd860b..9b3d461 100644 --- a/README.md +++ b/README.md @@ -268,7 +268,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. Add `extract_fields` next to `extract` to narrow the +fields returned. ### Upload URLs from a CSV @@ -310,11 +311,12 @@ 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, `run.stats.spend` (`{credits, cost}`) shows the +spend the API recorded for it. It is indicative, not billing-grade: your +account's usage and invoices are the authoritative figures. Successful +requests are charged, and so are target `404` / `410` responses, which +show up as failed tasks. Result rows carry no per-task spend: use the +run-level rollup. ### Retry only the failed tasks @@ -382,9 +384,16 @@ 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-side zip has a size cap per run. For larger runs, or one +> file per task, use `download_to_dir`, which fetches bodies client-side. +> It is not bound by the export cap, only by its own configurable limits +> (`max_files`, `max_bytes_per_file`), and is slower (tune with +> `concurrency=`). Rows with no body to fetch (failed tasks, rows not +> finished yet, and results past the 14-day visibility window, which come +> back as `expired`) are skipped and counted in a warning log record. A +> `successful` row that comes back without a `result_url` is fetched once +> more; if its link is still missing, the download raises +> `ResultURLUnavailable` instead of dropping it. **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), @@ -438,8 +447,11 @@ 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. +endpoints); it defaults to unsigned. You get one event per run that +completes or fails on its own; stopping a run sends nothing. Delivery is +best-effort: a failed delivery is retried with backoff for a bounded +window, so an event can arrive more than once, and is not guaranteed. +Dedupe on the `X-ZenRows-Event-Id` header. ```python client.submit_regular( @@ -448,6 +460,20 @@ client.submit_regular( ) ``` +Each delivery is one POST when a run finishes (`run.completed` or +`run.failed`), carrying the run's ids and stats, not the results. On a +signed delivery, check the `X-Signature` header with `verify_webhook`. +Pass the raw request body and the secret exactly as the API showed it +(base64); the helper decodes it, picks the key by `kid`, and rejects +deliveries older than five minutes: + +```python +from zenrows.batch import verify_webhook + +if not verify_webhook(request.body, request.headers["X-Signature"], {kid: secret_b64}): + return 401 +``` + > No webhook receiver? Just poll for completion with `job.run.wait()` / > `run.wait()`. diff --git a/docs/batch-client-reference.md b/docs/batch-client-reference.md index 4f4ced6..8bf4afd 100644 --- a/docs/batch-client-reference.md +++ b/docs/batch-client-reference.md @@ -120,7 +120,7 @@ def submit_regular(urls: list[str | TaskInputDict] | None = None, Submit a one-shot scraping job (closed, all tasks upfront). Pass `urls` as either bare strings or inline dicts carrying -per-task `external_id` / `metadata` / `params`: +per-task `external_id` / `metadata` / `zenrows_params`: client.submit_regular(["https://a", "https://b"]) client.submit_regular([ @@ -417,8 +417,7 @@ until `timeout`. Pass `failure_statuses={"failed"}` to have the waiter raise `WaiterError` on it instead. `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`). +It is off by default. @@ -527,8 +526,8 @@ carrying the just-issued `export_id`; call `.load()` or block on 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. +error="results are larger than N gb") if the pre-zip total +exceeds the server's export size cap. @@ -564,7 +563,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,12 +590,13 @@ 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-side export has a size cap per run. 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=`). +fetches each body client-side, bound only by its own configurable +limits (`max_files`, `max_bytes_per_file`), at the cost of being +slower (one body at a time, tunable via `concurrency=`). # Job, run & export handles @@ -748,9 +748,10 @@ 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 the export size per run; for larger runs use +``download_to_dir``, bound only by its own configurable limits +(``max_files``, ``max_bytes_per_file``) and slower: one body at a +time, tunable via ``concurrency=``. @@ -798,7 +799,7 @@ 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. when the run is over the size cap). Call ``load()`` / ``wait()`` for an :class:`ExportHandle` with data. @@ -1028,9 +1029,10 @@ 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 the export size per run; for larger runs use +``download_to_dir``, bound only by its own configurable limits +(``max_files``, ``max_bytes_per_file``) and slower: one body at a +time, tunable via ``concurrency=``. @@ -1235,10 +1237,18 @@ 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 to the open initial run. + +Pass ``idempotency_key`` to make the call safe to retry: a repeat +with the same key and body replays the first answer instead of +adding the tasks twice, and the transport then retries this POST +on transient errors. Without a key every call appends its tasks +and a failed call is not retried automatically. @@ -1449,6 +1459,17 @@ Results are NOT guaranteed to be in iteration order in either bulk flavour. The iteration of `iter_results` itself stays single-threaded so we never page faster than we drain. +Rows with no body to download carry no `result_url`: failed tasks, +`expired` rows whose result is past the visibility window, and rows not +finished yet (`pending` / `processing`). The bulk helpers skip those +instead of stopping part-way, and log one WARNING on the +`zenrows.batch.download` logger with the counts per status. A +`successful` row without a `result_url` is different: its body exists +but the API could not sign a link for it. The helpers fetch that page +once more, and if the link is still missing they raise +`ResultURLUnavailable` rather than drop a result silently. The +single-task helpers still raise `ValueError` for a row with no URL. + `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). @@ -1476,6 +1497,22 @@ One row's worth of downloaded content held in memory. 500 MiB + + +## ResultURLUnavailable Objects + +```python +class ResultURLUnavailable(ValueError) +``` + +Raised by a bulk download when `successful` rows still have no +`result_url` after their results page was fetched a second time. +The bodies exist; the API could not sign download links for them +just now. Retry later, or fetch them one by one. + +Subclasses `ValueError`, which bulk downloads raised for a row with +no `result_url` before, so existing `except ValueError` still works. + ## DownloadLimitExceeded Objects @@ -1506,7 +1543,11 @@ def download_to_dir( max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE) -> int ``` -Stream every task's body into `target_dir`. Returns the count. +Stream every task's body into `target_dir`. Returns the number of +files written. Rows with no body (failed, `expired`, or not finished) +are skipped and counted in a WARNING log record; a `successful` row +still missing its `result_url` after one refetch raises +`ResultURLUnavailable`. `concurrency` parallelises the body-fetch + write. With the default 1, behaviour is exactly serial. With N>1, up to N bodies @@ -1562,6 +1603,9 @@ def download_to_memory( ``` Load every (matching) task body into a list and return it. +Rows with no body (failed, `expired`, or not finished) are skipped +and counted in a WARNING log record; a `successful` row still missing +its `result_url` after one refetch raises `ResultURLUnavailable`. Three independent caps that all raise `DownloadLimitExceeded`: - `max_count` — how many rows total @@ -1830,6 +1874,53 @@ 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 an incoming Batch webhook delivery. + +A signed delivery carries an `X-Signature` header shaped +`t=,v1=,kid=`. The digest is +HMAC-SHA256 over the bytes `.`, keyed with the +signing secret. The API shows that secret once, base64-encoded, when you +create or rotate a key; this helper decodes it for you. + + from zenrows.batch import verify_webhook + + ok = verify_webhook( + request.body, # raw bytes, before any JSON parsing + request.headers["X-Signature"], + {"kid_123": "q3Jk...base64...="}, # every key you may still receive + ) + +Keep the previous key in `secrets` while a rotation is in progress: +deliveries name the key they were signed with in `kid`. + + + +#### verify\_webhook + +```python +def verify_webhook(raw_body: bytes | str, + signature_header: str, + secrets: Mapping[str, str], + *, + tolerance_seconds: float = DEFAULT_TOLERANCE_SECONDS, + now: float | None = None) -> bool +``` + +Return True when `signature_header` is a valid signature of `raw_body`. + +`raw_body` must be the exact bytes received (a `str` is encoded as +UTF-8). `secrets` maps key id (`kid`) to the base64 secret the API +returned; a secret that doesn't decode to 32 bytes (including an +empty one) never verifies. The header's `kid` picks the secret; an unknown `kid`, a +malformed header or secret, a wrong digest, or a timestamp more than +`tolerance_seconds` away from `now` (default: the current time) all +return False. Pass `tolerance_seconds=0` to skip the timestamp check; +a negative tolerance rejects every delivery. It never raises on bad +input, including arguments of the wrong type: a delivery it can't +read is rejected. + # Errors RFC 7807 Problem JSON → friendly Python exceptions. @@ -1985,6 +2076,11 @@ Clients must accept values not listed here. class TaskStatus(Enum) ``` +- `expired`: the task's result is past the visibility window + (14 days by default). The row has no `result_url` and + `/content` answers `410 result_expired`. A `successful` + filter still returns these rows. + Clients must accept values not listed here. @@ -2061,7 +2157,9 @@ assigned `task_id`. #### url -Must be http(s). Other schemes rejected at submit. +Must be http(s). Other schemes rejected at submit. The API +limits the URL to 2,048 bytes once UTF-8 encoded +(`url_too_long`). @@ -2144,11 +2242,11 @@ cost?" indicator, not for reconciliation. class TaskSpend(BaseModel) ``` -Per-task indicative spend with two roll-ups: `total` +Deprecated: result rows never carry per-task spend, so this +model is never populated. Kept so existing imports keep +working. 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. +`last_attempt` carries just the most recent attempt. @@ -2834,7 +2932,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 (a per-environment setting). @@ -2949,9 +3047,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. Empty for non-successful and `expired` tasks. The +body is also served by the separate +`GET /v1/jobs/{job_id}/runs/{run_id}/tasks/{task_id}/content` +endpoint, which needs your API key. @@ -2972,6 +3072,13 @@ 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 per-task spend, so this +is always empty. Read the run-level `stats.spend` instead. + #### created\_at @@ -3153,7 +3260,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..222f55e 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 creating 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,8 @@ 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 (a per-environment + setting). responses: '202': description: Export accepted; the zip is produced asynchronously. @@ -1078,9 +1090,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. Re-sending the same key 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 live 24 h. schema: { type: string, maxLength: 128 } ResultStatus: name: status @@ -1138,6 +1152,17 @@ components: content: application/problem+json: schema: { $ref: '#/components/schemas/Problem' } + QuotaExceeded: + description: | + Account limit reached (`quota_exceeded`): too many active jobs + (3 by default), too many scheduled jobs (5 by default), or an + open job's task cap (100,000). A job stays active while it is + open, and after it is closed until its latest run finishes; + deleting it frees the slot at once. Wait for a run to finish, + close an idle open job, or delete one. + content: + application/problem+json: + schema: { $ref: '#/components/schemas/Problem' } NotImplemented: description: Feature is reserved but not yet implemented. content: @@ -1236,8 +1261,14 @@ 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`: the task's result is past the visibility window + (14 days by default). The row has no `result_url` and + `/content` answers `410 result_expired`. A `successful` + filter still returns these rows. + + Clients must accept values not listed here. ResultType: type: string @@ -1310,7 +1341,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: @@ -1341,7 +1372,10 @@ components: type: string format: uri maxLength: 2048 - description: Must be http(s). Other schemes rejected at submit. + description: | + Must be http(s). Other schemes rejected at submit. The API + limits the URL to 2,048 bytes once UTF-8 encoded + (`url_too_long`). metadata: { $ref: '#/components/schemas/Metadata' } method: type: string @@ -1398,12 +1432,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 +1502,7 @@ components: tasks: type: array minItems: 1 - maxItems: 1000 + maxItems: 10000 items: { $ref: '#/components/schemas/TaskInput' } last_batch: type: boolean @@ -1524,12 +1558,13 @@ components: TaskSpend: type: object + deprecated: true description: | - Per-task indicative spend with two roll-ups: `total` + Deprecated: result rows never carry per-task spend, so this + model is never populated. Kept so existing imports keep + working. 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. + `last_attempt` carries just the most recent attempt. required: [total, last_attempt] properties: total: { $ref: '#/components/schemas/Spend' } @@ -1581,6 +1616,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 +1886,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. Empty for non-successful and `expired` tasks. The + body is also served by the separate + `GET /v1/jobs/{job_id}/runs/{run_id}/tasks/{task_id}/content` + endpoint, which needs your API key. error: $ref: '#/components/schemas/Problem' description: | @@ -1868,7 +1906,11 @@ 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 per-task spend, so this + is always empty. Read the run-level `stats.spend` instead. created_at: type: string format: date-time @@ -2284,7 +2326,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 (a per-environment setting). download_url: type: string format: uri diff --git a/examples/08_download_all_results.py b/examples/08_download_all_results.py index 26cfb79..0cc4465 100644 --- a/examples/08_download_all_results.py +++ b/examples/08_download_all_results.py @@ -6,11 +6,13 @@ 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 zip has a 12h TTL and a server-side size cap per run. An oversize +run fails the export (`"results are larger than N gb"`) and raises +`BatchAPIError`. For larger runs (or one file per task), use the +client-side download `download_to_dir` (example 02). It is not bound by +the export cap, only by its own configurable limits (`max_files`, +`max_bytes_per_file`), and 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 +42,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 export size cap. print(f"export failed: {e}") diff --git a/scripts/build_reference.py b/scripts/build_reference.py index 6a2d38b..d121b3c 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..eda3366 100644 --- a/src/zenrows/batch/__init__.py +++ b/src/zenrows/batch/__init__.py @@ -15,7 +15,7 @@ webhooks — poll or get notified; there's no callback to wire up. """ -from zenrows.batch._download import DownloadedResult, DownloadLimitExceeded +from zenrows.batch._download import DownloadedResult, DownloadLimitExceeded, ResultURLUnavailable from zenrows.batch._estimate import ( CostEstimate, CostLine, @@ -74,6 +74,7 @@ TestWebhookResponse, WebhookConfig, ) +from zenrows.batch.webhooks import verify_webhook __all__ = [ "TERMINAL_RUN_STATUSES", @@ -106,6 +107,7 @@ "Monthly", "ProblemDetail", "Rate", + "ResultURLUnavailable", "Run", "RunHandle", "RunRef", @@ -129,4 +131,5 @@ "WebhookDict", "Weekly", "ZenRowsBatchClient", + "verify_webhook", ] diff --git a/src/zenrows/batch/_download.py b/src/zenrows/batch/_download.py index b06ff50..39de30f 100644 --- a/src/zenrows/batch/_download.py +++ b/src/zenrows/batch/_download.py @@ -25,6 +25,17 @@ flavour. The iteration of `iter_results` itself stays single-threaded so we never page faster than we drain. +Rows with no body to download carry no `result_url`: failed tasks, +`expired` rows whose result is past the visibility window, and rows not +finished yet (`pending` / `processing`). The bulk helpers skip those +instead of stopping part-way, and log one WARNING on the +`zenrows.batch.download` logger with the counts per status. A +`successful` row without a `result_url` is different: its body exists +but the API could not sign a link for it. The helpers fetch that page +once more, and if the link is still missing they raise +`ResultURLUnavailable` rather than drop a result silently. The +single-task helpers still raise `ValueError` for a row with no URL. + `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). @@ -32,6 +43,7 @@ import os import threading +from collections import Counter from collections.abc import Callable, Iterable, Iterator from concurrent.futures import ThreadPoolExecutor, as_completed from contextlib import contextmanager, nullcontext @@ -41,9 +53,17 @@ import httpx +from zenrows.batch._logging import logger from zenrows.batch.errors import BatchAPIError from zenrows.batch.models import TaskResult, TaskStatus +_log = logger("download") + + +class _ResultsPage(Protocol): + results: list[TaskResult] + next_cursor: str | None + class _BodyFetcher(Protocol): """The slice of `ZenRowsBatchClient` the helpers need — just result @@ -54,13 +74,14 @@ class _BodyFetcher(Protocol): the whole client. """ - def _iter_results_raw( + def list_results( self, job_id: str, *, run_id: str | None = ..., status: str | None = ..., - ) -> Iterator[TaskResult]: ... + cursor: str | None = ..., + ) -> _ResultsPage: ... @dataclass(slots=True) @@ -81,6 +102,24 @@ class DownloadedResult: DEFAULT_MAX_TOTAL_BYTES_IN_MEMORY = 500 * 1024 * 1024 # 500 MiB +class ResultURLUnavailable(ValueError): + """Raised by a bulk download when `successful` rows still have no + `result_url` after their results page was fetched a second time. + The bodies exist; the API could not sign download links for them + just now. Retry later, or fetch them one by one. + + Subclasses `ValueError`, which bulk downloads raised for a row with + no `result_url` before, so existing `except ValueError` still works.""" + + def __init__(self, task_ids: list[str]): + self.task_ids = task_ids + shown = ", ".join(task_ids[:5]) + (", ..." if len(task_ids) > 5 else "") + super().__init__( + f"download: {len(task_ids)} successful result(s) have no result_url " + f"after a retry: {shown}" + ) + + class DownloadLimitExceeded(RuntimeError): """Raised when a download exceeds a configured cap.""" @@ -108,7 +147,11 @@ def download_to_dir( max_files: int = DEFAULT_MAX_FILES, max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE, ) -> int: - """Stream every task's body into `target_dir`. Returns the count. + """Stream every task's body into `target_dir`. Returns the number of + files written. Rows with no body (failed, `expired`, or not finished) + are skipped and counted in a WARNING log record; a `successful` row + still missing its `result_url` after one refetch raises + `ResultURLUnavailable`. `concurrency` parallelises the body-fetch + write. With the default 1, behaviour is exactly serial. With N>1, up to N bodies @@ -285,6 +328,9 @@ def download_to_memory( max_bytes_per_file: int = DEFAULT_MAX_BYTES_PER_FILE, ) -> list[DownloadedResult]: """Load every (matching) task body into a list and return it. + Rows with no body (failed, `expired`, or not finished) are skipped + and counted in a WARNING log record; a `successful` row still missing + its `result_url` after one refetch raises `ResultURLUnavailable`. Three independent caps that all raise `DownloadLimitExceeded`: - `max_count` — how many rows total @@ -353,27 +399,77 @@ def _run_bulk( 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) - - with _maybe_progress(progress, progress_description) as advance: - if concurrency <= 1: - written = 0 + skipped: Counter[str] = Counter() # status -> rows with no body to fetch + + def downloadable() -> Iterator[TaskResult]: + cursor: str | None = None + while True: + page = client.list_results(job_id, run_id=run_id, status=status, cursor=cursor) + rows = list(page.results) + if any(_unsigned_success(r) for r in rows): + # The body exists but the link could not be signed on + # this read: ask for the same page once more. + again = { + r.task_id: r + for r in client.list_results( + job_id, run_id=run_id, status=status, cursor=cursor + ).results + } + rows = [again.get(r.task_id, r) if _unsigned_success(r) else r for r in rows] + missing = [r.task_id for r in rows if _unsigned_success(r)] + if missing: + raise ResultURLUnavailable(missing) for row in rows: - if written >= max_rows: - raise DownloadLimitExceeded(max_rows_limit_name, max_rows, written + 1) - per_row(row) - written += 1 - advance(1) - return written - - return _run_parallel( - rows, - per_row=per_row, - concurrency=concurrency, - max_rows=max_rows, - max_rows_limit_name=max_rows_limit_name, - advance=advance, - ) + if row.result_url: + yield row + else: + # Failed, `expired`, or not finished yet: no body. + skipped[_status_name(row)] += 1 + cursor = page.next_cursor + if not cursor: + return + + rows = downloadable() + try: + with _maybe_progress(progress, progress_description) as advance: + if concurrency <= 1: + written = 0 + for row in rows: + if written >= max_rows: + raise DownloadLimitExceeded(max_rows_limit_name, max_rows, written + 1) + per_row(row) + written += 1 + advance(1) + return written + + return _run_parallel( + rows, + per_row=per_row, + concurrency=concurrency, + max_rows=max_rows, + max_rows_limit_name=max_rows_limit_name, + advance=advance, + ) + finally: + total = sum(skipped.values()) + if total: + detail = ", ".join(f"{n} {name}" for name, n in sorted(skipped.items())) + _log.warning( + "skipped %d result(s) with no downloadable body (%s)", + total, + detail, + extra={"job_id": job_id, "run_id": run_id, "skipped": dict(skipped)}, + ) + + +def _status_name(row: TaskResult) -> str: + return getattr(row.status, "value", None) or str(row.status) + + +def _unsigned_success(row: TaskResult) -> bool: + """A `successful` row with no `result_url`: the body exists but the + API returned no signed link for it.""" + return row.status is TaskStatus.SUCCESSFUL and not row.result_url def _run_parallel( @@ -464,6 +560,7 @@ def _maybe_progress(enabled: bool, description: str): "DEFAULT_MAX_TOTAL_BYTES_IN_MEMORY", "DownloadLimitExceeded", "DownloadedResult", + "ResultURLUnavailable", "download_to_dir", "download_to_memory", ] diff --git a/src/zenrows/batch/_estimate.py b/src/zenrows/batch/_estimate.py index 9deb559..ae003dd 100644 --- a/src/zenrows/batch/_estimate.py +++ b/src/zenrows/batch/_estimate.py @@ -51,9 +51,10 @@ 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 lowercases boolean +# params and accepts only "true" / "false" (anything else, such as 1, +# "1", "yes" or "on", is rejected with 400), so only JSON true or a +# string equal to "true" ignoring case counts as "on". # 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,14 +171,13 @@ 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.""" + """Read a boolean scraper param the way the API does: JSON `True`, + or a string equal to `"true"` ignoring case. Every other value is + off (the API rejects non-boolean spellings at submit).""" 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 diff --git a/src/zenrows/batch/_resources.py b/src/zenrows/batch/_resources.py index 00f852d..1a7830b 100644 --- a/src/zenrows/batch/_resources.py +++ b/src/zenrows/batch/_resources.py @@ -226,9 +226,10 @@ 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 the export size per run; for larger runs use + ``download_to_dir``, bound only by its own configurable limits + (``max_files``, ``max_bytes_per_file``) and slower: one body at a + time, tunable via ``concurrency=``.""" return self._client.download_all_results( self.job_id, self.run_id, @@ -279,7 +280,7 @@ 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. when the run is over the size cap). Call ``load()`` / ``wait()`` for an :class:`ExportHandle` with data. """ @@ -594,9 +595,10 @@ 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 the export size per run; for larger runs use + ``download_to_dir``, bound only by its own configurable limits + (``max_files``, ``max_bytes_per_file``) and slower: one body at a + time, tunable via ``concurrency=``.""" return self._client.download_all_results( self.job_id, self._current_run_id(), @@ -786,9 +788,20 @@ def retry_failed( # ----- 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 to the open initial run. + + Pass ``idempotency_key`` to make the call safe to retry: a repeat + with the same key and body replays the first answer instead of + adding the tasks twice, and the transport then retries this POST + on transient errors. Without a key every call appends its tasks + and a failed call is not retried automatically.""" + return self._client._post_tasks(self.job_id, body, idempotency_key=idempotency_key) # ----- runs (list) ----- diff --git a/src/zenrows/batch/client.py b/src/zenrows/batch/client.py index af09969..b64bc3a 100644 --- a/src/zenrows/batch/client.py +++ b/src/zenrows/batch/client.py @@ -253,7 +253,7 @@ def submit_regular( """Submit a one-shot scraping job (closed, all tasks upfront). Pass `urls` as either bare strings or inline dicts carrying - per-task `external_id` / `metadata` / `params`: + per-task `external_id` / `metadata` / `zenrows_params`: client.submit_regular(["https://a", "https://b"]) client.submit_regular([ @@ -583,8 +583,7 @@ def wait_for_run( waiter raise `WaiterError` on it instead. `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`). + It is off by default. """ return self._wait_for_run_raw( job_id, @@ -721,8 +720,8 @@ def start_results_export(self, job_id: str, run_id: str) -> ExportRef: 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. + error="results are larger than N gb") if the pre-zip total + exceeds the server's export size cap. """ resp = self._post_export_start(job_id, run_id) return ExportRef(self, job_id, run_id, resp.export_id, start_response=resp) @@ -750,7 +749,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 +834,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,12 +936,13 @@ 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-side export has a size cap per run. 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=`). + fetches each body client-side, bound only by its own configurable + limits (`max_files`, `max_bytes_per_file`), at the cost of being + slower (one body at a time, tunable via `concurrency=`). """ export = self.start_results_export(job_id, run_id) final = self.wait_for_export( diff --git a/src/zenrows/batch/models.py b/src/zenrows/batch/models.py index 78d2690..26037a8 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-15T10:36:33+00:00 from __future__ import annotations from zenrows.batch._open_enum import open_enum_missing @@ -124,13 +124,20 @@ class RunStatus(Enum): class TaskStatus(Enum): """ + - `expired`: the task's result is past the visibility window + (14 days by default). The row has no `result_url` and + `/content` answers `410 result_expired`. A `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) @@ -214,7 +221,10 @@ class TaskInput(BaseModel): """ url: AnyUrl """ - Must be http(s). Other schemes rejected at submit. + Must be http(s). Other schemes rejected at submit. The API + limits the URL to 2,048 bytes once UTF-8 encoded + (`url_too_long`). + """ metadata: Annotated[dict[str, str] | None, Field(max_length=20)] = None method: Method | None = "GET" @@ -262,7 +272,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 +296,11 @@ class Spend(BaseModel): class TaskSpend(BaseModel): """ - Per-task indicative spend with two roll-ups: `total` + Deprecated: result rows never carry per-task spend, so this + model is never populated. Kept so existing imports keep + working. 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. + `last_attempt` carries just the most recent attempt. """ @@ -401,6 +411,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 """ @@ -970,7 +981,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 (a per-environment setting). """ download_url: AnyUrl | None = None @@ -1078,9 +1089,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. Empty for non-successful and `expired` tasks. The + body is also served by the separate + `GET /v1/jobs/{job_id}/runs/{run_id}/tasks/{task_id}/content` + endpoint, which needs your API key. """ error: Problem | None = None @@ -1101,6 +1114,11 @@ class TaskResult(BaseModel): """ spend: TaskSpend | None = None + """ + Deprecated: result rows never carry 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/src/zenrows/batch/webhooks.py b/src/zenrows/batch/webhooks.py new file mode 100644 index 0000000..96e38ff --- /dev/null +++ b/src/zenrows/batch/webhooks.py @@ -0,0 +1,117 @@ +"""Verify the signature on an incoming Batch webhook delivery. + +A signed delivery carries an `X-Signature` header shaped +`t=,v1=,kid=`. The digest is +HMAC-SHA256 over the bytes `.`, keyed with the +signing secret. The API shows that secret once, base64-encoded, when you +create or rotate a key; this helper decodes it for you. + + from zenrows.batch import verify_webhook + + ok = verify_webhook( + request.body, # raw bytes, before any JSON parsing + request.headers["X-Signature"], + {"kid_123": "q3Jk...base64...="}, # every key you may still receive + ) + +Keep the previous key in `secrets` while a rotation is in progress: +deliveries name the key they were signed with in `kid`. +""" + +import base64 +import binascii +import hashlib +import hmac +import math +import re +import time +from collections.abc import Mapping + +DEFAULT_TOLERANCE_SECONDS = 300 + +# 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 (ASCII digits, bounded so arithmetic can't +# overflow) and a SHA-256 hex digest. Anything else is malformed. +_TIMESTAMP = re.compile(r"[0-9]{1,12}", re.ASCII) +_DIGEST = re.compile(r"[0-9a-fA-F]{64}", re.ASCII) + + +def verify_webhook( + raw_body: bytes | str, + signature_header: str, + secrets: Mapping[str, str], + *, + tolerance_seconds: float = DEFAULT_TOLERANCE_SECONDS, + now: float | None = None, +) -> bool: + """Return True when `signature_header` is a valid signature of `raw_body`. + + `raw_body` must be the exact bytes received (a `str` is encoded as + UTF-8). `secrets` maps key id (`kid`) to the base64 secret the API + returned; a secret that doesn't decode to 32 bytes (including an + empty one) never verifies. The header's `kid` picks the secret; an unknown `kid`, a + malformed header or secret, a wrong digest, or a timestamp more than + `tolerance_seconds` away from `now` (default: the current time) all + return False. Pass `tolerance_seconds=0` to skip the timestamp check; + a negative tolerance rejects every delivery. It never raises on bad + input, including arguments of the wrong type: a delivery it can't + read is rejected. + """ + 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 t is None or digest is None or kid is None: + 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: + # `now` and the timestamp are both validated finite numbers here. + 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. Ints too large + for a float count as not finite rather than raising.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return False + try: + return math.isfinite(value) + except OverflowError: + return False + + +__all__ = ["DEFAULT_TOLERANCE_SECONDS", "verify_webhook"] diff --git a/tests/test_batch_client.py b/tests/test_batch_client.py index 09d1f21..2a5ec48 100644 --- a/tests/test_batch_client.py +++ b/tests/test_batch_client.py @@ -1178,3 +1178,236 @@ 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] + + +# ===== rows with no body, and fields the server sends verbatim ===== + + +def _row_json(task_id: str, status: str, result_url: str | None = None) -> dict: + data = { + "task_id": task_id, + "run_id": "R", + "url": "https://example.com/" + task_id, + "status": status, + "created_at": "2026-08-25T12:00:00Z", + "updated_at": "2026-08-25T12:00:05Z", + } + if result_url: + data["result_url"] = result_url + data["type"] = "html" + return data + + +@pytest.mark.parametrize("concurrency", [1, 4]) +@respx.mock +def test_bulk_downloads_skip_rows_without_a_body( + client: ZenRowsBatchClient, tmp_path, caplog, concurrency +): + """An `expired` row (past the visibility window) or a failed row has + no `result_url`: bulk downloads skip it, count it, and keep going.""" + url = "https://storage.example.test/bodies/T1.html?sig=abc" + respx.get(url).mock(return_value=Response(200, content=b"ok")) + respx.get(f"{BASE_URL}/jobs/J/runs/R/results").mock( + return_value=Response( + 200, + json={ + "results": [ + _row_json("T1", "successful", url), + _row_json("T2", "expired"), + _row_json("T3", "failed"), + ] + }, + ) + ) + run = client.run("J", "R") + + with caplog.at_level("WARNING", logger="zenrows.batch.download"): + written = run.download_to_dir(tmp_path, status=None, concurrency=concurrency) + assert written == 1 + assert [p.name for p in tmp_path.iterdir()] == ["T1.html"] + assert "skipped 2 result(s)" in caplog.text + + caplog.clear() + with caplog.at_level("WARNING", logger="zenrows.batch.download"): + bodies = run.download_to_memory(status=None, concurrency=concurrency) + assert [b.task_id for b in bodies] == ["T1"] + assert "skipped 2 result(s)" in caplog.text + + +@respx.mock +def test_bulk_download_skip_message_counts_each_status( + client: ZenRowsBatchClient, tmp_path, caplog +): + respx.get(f"{BASE_URL}/jobs/J/runs/R/results").mock( + return_value=Response( + 200, + json={ + "results": [ + _row_json("T1", "expired"), + _row_json("T2", "failed"), + _row_json("T3", "pending"), + _row_json("T4", "processing"), + ] + }, + ) + ) + with caplog.at_level("WARNING", logger="zenrows.batch.download"): + assert client.run("J", "R").download_to_dir(tmp_path, status=None) == 0 + assert "skipped 4 result(s)" in caplog.text + assert "1 expired, 1 failed, 1 pending, 1 processing" in caplog.text + + +@respx.mock +def test_successful_row_without_url_is_refetched_once(client: ZenRowsBatchClient, tmp_path): + """A successful row with no `result_url` means the link wasn't signed + on that read: the page is fetched again and the row is downloaded.""" + url = "https://storage.example.test/bodies/T1.html?sig=abc" + respx.get(url).mock(return_value=Response(200, content=b"ok")) + page = respx.get(f"{BASE_URL}/jobs/J/runs/R/results").mock( + side_effect=[ + Response(200, json={"results": [_row_json("T1", "successful")]}), + Response(200, json={"results": [_row_json("T1", "successful", url)]}), + ] + ) + assert client.run("J", "R").download_to_dir(tmp_path) == 1 + assert page.call_count == 2 + assert (tmp_path / "T1.html").read_bytes() == b"ok" + + +@respx.mock +def test_successful_row_still_without_url_raises(client: ZenRowsBatchClient, tmp_path): + """If the second read still has no link, the download fails loudly + instead of dropping a successful result.""" + from zenrows.batch import ResultURLUnavailable + + page = respx.get(f"{BASE_URL}/jobs/J/runs/R/results").mock( + return_value=Response(200, json={"results": [_row_json("T1", "successful")]}) + ) + with pytest.raises(ResultURLUnavailable) as exc: + client.run("J", "R").download_to_memory() + assert exc.value.task_ids == ["T1"] + assert page.call_count == 2 + + +def test_task_result_parses_expired(): + from zenrows.batch import TaskStatus + + row = TaskResult.model_validate(_row_json("T1", "expired")) + assert row.status is TaskStatus.EXPIRED + assert row.result_url is None + assert row.url.host == "example.com" # still an AnyUrl, as before + + +def test_run_carries_trigger(): + from zenrows.batch import Run + from zenrows.batch.models import RunTrigger + + run = Run.model_validate( + { + "run_id": "R", + "job_id": "J", + "run_sequence": 1, + "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_add_tasks_sends_idempotency_key_and_retries(client: ZenRowsBatchClient, no_sleep): + """With a key, add_tasks is safe to replay: the header is sent and a + transient 503 is retried. Without one, it is sent once.""" + route = respx.post(f"{BASE_URL}/jobs/J/tasks").mock( + side_effect=[ + Response(503, json={"code": "unavailable"}), + Response( + 200, + json={ + "accepted_tasks": 1, + "job_status": "open", + "latest_run": { + "run_id": "R", + "job_id": "J", + "run_sequence": 1, + "status": "pending", + "stats": {"total": 1, "completed": 0, "successful": 0, "failed": 0}, + "created_at": "2026-08-25T12:00:00Z", + "updated_at": "2026-08-25T12:00:05Z", + }, + }, + ), + ] + ) + client.job("J").add_tasks( + {"tasks": [{"url": "https://example.com/a"}]}, idempotency_key="add-1" + ) + assert route.call_count == 2 + assert all(c.request.headers["Idempotency-Key"] == "add-1" for c in route.calls) + + +@respx.mock +def test_add_tasks_without_a_key_is_sent_once(client: ZenRowsBatchClient, no_sleep): + """No key means the POST may not be replayed: a 503 surfaces at once.""" + route = respx.post(f"{BASE_URL}/jobs/J/tasks").mock( + return_value=Response(503, json={"code": "unavailable"}) + ) + 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 + + +def test_result_url_unavailable_is_still_a_value_error(): + """Bulk downloads raised ValueError for a URL-less row before; the new + exception keeps `except ValueError` working.""" + from zenrows.batch import ResultURLUnavailable + + assert issubclass(ResultURLUnavailable, ValueError) + err = ResultURLUnavailable([f"T{i}" for i in range(7)]) + assert err.task_ids[0] == "T0" + assert "7 successful result(s)" in str(err) and str(err).endswith(", ...") + + +def test_deprecated_task_spend_still_imports_and_parses(): + import warnings + + from zenrows.batch.models import TaskSpend + + spend = {"credits": 1, "cost": 0.1} + row = TaskResult.model_validate( + {**_row_json("T1", "successful"), "spend": {"total": spend, "last_attempt": spend}} + ) + with warnings.catch_warnings(): + # Deprecated in the docs only: reading it must not warn, so + # projects that turn warnings into errors keep working. + warnings.simplefilter("error") + assert isinstance(row.spend, TaskSpend) + assert row.spend.total.credits == 1 + assert TaskResult.model_validate(_row_json("T2", "successful")).spend is None + + +@respx.mock +def test_bulk_download_cap_still_applies_after_skipping(client: ZenRowsBatchClient, tmp_path): + from zenrows.batch import DownloadLimitExceeded + + url = "https://storage.example.test/bodies/T{}.html?sig=abc" + for i in (1, 2): + respx.get(url.format(i)).mock(return_value=Response(200, content=b"x")) + respx.get(f"{BASE_URL}/jobs/J/runs/R/results").mock( + return_value=Response( + 200, + json={ + "results": [ + _row_json("T0", "failed"), + _row_json("T1", "successful", url.format(1)), + _row_json("T2", "successful", url.format(2)), + ] + }, + ) + ) + with pytest.raises(DownloadLimitExceeded): + client.run("J", "R").download_to_dir(tmp_path, status=None, max_files=1) diff --git a/tests/test_estimate.py b/tests/test_estimate.py index 659b4da..faafd8b 100644 --- a/tests/test_estimate.py +++ b/tests/test_estimate.py @@ -35,13 +35,18 @@ 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"]) -def test_truthy_spellings_turn_on_a_flag(value): +@pytest.mark.parametrize("value", [True, "true", "True", "TRUE"]) +def test_true_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"]) -def test_falsy_spellings_keep_base(value): +@pytest.mark.parametrize( + "value", + # The API accepts only true / "true" (any case) as on; 1, "1", "yes", + # "on" and padded strings are rejected at submit, so they never cost more. + [False, "false", "0", 0, "", "no", 1, "1", "yes", "on", " TRUE "], +) +def test_other_spellings_keep_base(value): assert cost_for_params({"js_render": value}).tier is Tier.BASE diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py new file mode 100644 index 0000000..10feac1 --- /dev/null +++ b/tests/test_webhooks.py @@ -0,0 +1,147 @@ +"""verify_webhook: signatures built the way the Batch API signs them.""" + +import base64 +import hashlib +import hmac +import os + +from zenrows.batch import verify_webhook + +RAW = os.urandom(32) +SECRET_B64 = base64.b64encode(RAW).decode() +BODY = b'{"schema":1,"event_type":"run.completed","job_id":"j1","run_id":"r1"}' +T = 1_760_000_000 + + +def sign(body: bytes = BODY, t: int = T, key: bytes = RAW, kid: str = "k1") -> str: + digest = hmac.new(key, f"{t}.".encode() + body, hashlib.sha256).hexdigest() + return f"t={t},v1={digest},kid={kid}" + + +def test_accepts_a_genuine_delivery(): + assert verify_webhook(BODY, sign(), {"k1": SECRET_B64}, now=T) + + +def test_accepts_str_body_and_spaces_in_header(): + header = sign().replace(",", ", ") + assert verify_webhook(BODY.decode(), header, {"k1": SECRET_B64}, now=T) + + +def test_rejects_a_tampered_body(): + assert not verify_webhook(BODY + b" ", sign(), {"k1": SECRET_B64}, now=T) + + +def test_rejects_the_undecoded_secret_used_as_key(): + # The mistake this helper exists to prevent: HMAC over the base64 text. + header = sign(key=SECRET_B64.encode()) + assert not verify_webhook(BODY, header, {"k1": SECRET_B64}, now=T) + + +def test_picks_the_secret_by_kid_during_rotation(): + old = base64.b64encode(os.urandom(32)).decode() + secrets = {"k0": old, "k1": SECRET_B64} + assert verify_webhook(BODY, sign(kid="k1"), secrets, now=T) + assert not verify_webhook(BODY, sign(kid="k0"), secrets, now=T) + + +def test_rejects_unknown_kid_and_malformed_input(): + assert not verify_webhook(BODY, sign(kid="nope"), {"k1": SECRET_B64}, now=T) + assert not verify_webhook(BODY, "garbage", {"k1": SECRET_B64}, now=T) + assert not verify_webhook(BODY, "", {"k1": SECRET_B64}, now=T) + assert not verify_webhook(BODY, sign(), {"k1": "not base64!"}, now=T) + assert not verify_webhook(BODY, sign().replace(f"t={T}", "t=abc"), {"k1": SECRET_B64}, now=T) + + +def test_timestamp_tolerance(): + assert verify_webhook(BODY, sign(), {"k1": SECRET_B64}, now=T + 300) + assert not verify_webhook(BODY, sign(), {"k1": SECRET_B64}, now=T + 301) + assert not verify_webhook(BODY, sign(), {"k1": SECRET_B64}, now=T - 301) + assert verify_webhook(BODY, sign(), {"k1": SECRET_B64}, now=T + 10_000, tolerance_seconds=0) + + +def test_malformed_headers_return_false_instead_of_raising(): + """Bad input from the network must be rejected, never crash the receiver.""" + s = {"k1": SECRET_B64} + good = sign().split(",")[1] # v1= + cases = [ + "t=1,v1=é,kid=k1", # non-ASCII digest + f"t={chr(0x661)},{good},kid=k1", # Arabic-Indic digit: int() would accept it + f"t={'9' * 400},{good},kid=k1", # huge timestamp + f"t=-5,{good},kid=k1", + f"t={T},v1=xyz,kid=k1", # not hex + f"t={T},v1={'a' * 63},kid=k1", # wrong length + f"t={T},{good}", # no kid + f"t=,{good},kid=k1", + ] + for header in cases: + assert verify_webhook(BODY, header, s, now=1.5) is False, header + assert verify_webhook(BODY, header, s, now=float(T)) is False, header + + +def test_unusual_now_values_do_not_raise(): + s = {"k1": SECRET_B64} + assert verify_webhook(BODY, sign(), s, now=float("inf")) is False + assert verify_webhook(BODY, sign(), s, now=float("nan")) is False + assert verify_webhook(BODY, sign(), s, now=1e308) is False + + +def test_uppercase_hex_digest_is_accepted(): + t, v1, kid = sign().split(",") + header = ",".join([t, v1.upper().replace("V1=", "v1="), kid]) + assert verify_webhook(BODY, header, {"k1": SECRET_B64}, now=T) + + +def test_negative_or_bad_tolerance_rejects(): + """Only 0 disables the age check; a negative value must not.""" + s = {"k1": SECRET_B64} + assert verify_webhook(BODY, sign(), s, now=T, tolerance_seconds=-1) is False + assert verify_webhook(BODY, sign(), s, now=T + 10_000, tolerance_seconds=-1) is False + assert verify_webhook(BODY, sign(), s, now=T, tolerance_seconds=float("nan")) is False + assert verify_webhook(BODY, sign(), s, now=T + 10, tolerance_seconds=30.5) + + +def test_wrong_typed_arguments_return_false(): + s = {"k1": SECRET_B64} + header = sign() + assert verify_webhook(BODY, header.encode(), s, now=T) is False # type: ignore[arg-type] + assert verify_webhook(BODY, 123, s, now=T) is False # type: ignore[arg-type] + assert verify_webhook(BODY, None, s, now=T) is False # type: ignore[arg-type] + assert verify_webhook(BODY, header, None, now=T) is False # type: ignore[arg-type] + assert verify_webhook(BODY, header, {"k1": 42}, now=T) is False # type: ignore[dict-item] + assert verify_webhook(None, header, s, now=T) is False # type: ignore[arg-type] + assert verify_webhook(BODY, header, s, now="soon") is False # type: ignore[arg-type] + assert verify_webhook(bytearray(BODY), header, s, now=T) + + +def test_huge_tolerance_returns_false_instead_of_raising(): + assert ( + verify_webhook(BODY, sign(), {"k1": SECRET_B64}, now=T, tolerance_seconds=10**1000) is False + ) + + +def test_bad_now_is_rejected_even_when_the_age_check_is_off(): + s = {"k1": SECRET_B64} + for bad in (float("nan"), float("inf"), float("-inf"), 10**1000, True, "soon"): + assert verify_webhook(BODY, sign(), s, now=bad, tolerance_seconds=0) is False, bad + assert verify_webhook(BODY, sign(), s, now=T, tolerance_seconds=0) + + +def test_empty_or_wrong_length_secret_never_verifies(): + """With an empty key anyone can compute a valid HMAC: refuse it.""" + forged = sign(key=b"") + assert verify_webhook(BODY, forged, {"k1": ""}, now=T) is False + short = base64.b64encode(os.urandom(16)).decode() + assert verify_webhook(BODY, sign(key=base64.b64decode(short)), {"k1": short}, now=T) is False + long = base64.b64encode(os.urandom(64)).decode() + assert verify_webhook(BODY, sign(key=base64.b64decode(long)), {"k1": long}, now=T) is False + + +def test_unreadable_secrets_or_body_return_false(): + class Exploding(dict): + def get(self, key, default=None): + raise RuntimeError("store unavailable") + + assert verify_webhook(BODY, sign(), Exploding(), now=T) is False + # A str body with a lone surrogate can't be UTF-8 encoded. + assert verify_webhook("\ud800", sign(), {"k1": SECRET_B64}, now=T) is False + assert verify_webhook(memoryview(BODY), sign(), {"k1": SECRET_B64}, now=T)