PIGS-1096 - Add Optimize API benchmark sample

README.md

@@ -102,6 +102,7 @@ This repository demonstrates complete business workflows through working code ex
 - **redact_by_keyword.py** - Finds and permanently redacts specific keywords across documents
 - **bulk_password_protect.py** - Adds password protection to multiple PDFs in batch
 - **prepare_pdf_for_distribution.py** - Marketing workflow that converts Word to PDF, compresses files, and removes metadata for external sharing
+- **optimize_benchmark.py** - Evaluation workflow that runs a folder of PDFs through the Optimize API and reports per-file size reduction in CSVs and a shareable, self-contained HTML report
 
 ### eSignature Example
 - **employee_policy_onboarding.py** - HR workflow that reads employee data from CSV, creates signature envelopes with policy documents, sends for signing, tracks status, and downloads signed documents organized by employee

samples/python/README.md

@@ -148,6 +148,7 @@ api/
 - `batch_process.py` - Batch convert documents
 - `bulk_password_protect.py` - Password protect multiple PDFs
 - `prepare_pdf_for_distribution.py` - Prepare PDFs for external distribution (convert, compress, remove metadata)
+- `optimize_benchmark.py` - Benchmark the Optimize API on a folder of PDFs, with CSVs and a self-contained HTML report
 
 ### Sign API Tools (eSignature)
 - `employee_policy_onboarding.py` - Complete HR workflow: send policy documents to employees for signature
@@ -199,6 +200,22 @@ uv run python batch_process.py ./input ./output pdf "*.docx"
 task batch INPUT_DIR=./input OUTPUT_DIR=./output FORMAT=pdf PATTERN='*.docx'
 ```
 
+### Benchmark PDF Compression
+```bash
+# Run every PDF in a folder through the Optimize API (default profile: minimal-file-size)
+uv run python optimize_benchmark.py ./pdfs ./output
+
+# Benchmark several profiles side by side
+uv run python optimize_benchmark.py ./pdfs ./output -p minimal-file-size -p web
+
+# Or use Task command
+task optimize-benchmark INPUT_DIR=./pdfs OUTPUT_DIR=./output
+```
+
+The run writes the optimized PDFs, `results.csv`, `summary.csv` and a
+self-contained `report.html` (headline numbers, a size-reduction distribution
+chart and a filterable per-file table) that opens in your browser when done.
+
 ## Using the API Client
 
 ### Platform API Client

samples/python/Taskfile.yml

@@ -48,6 +48,13 @@ tasks:
     requires:
       vars: [INPUT, OUTPUT, KEYWORDS]
 
+  optimize-benchmark:
+    desc: "Benchmark the Optimize API on a folder of PDFs, with an HTML report (e.g., task optimize-benchmark INPUT_DIR=./pdfs OUTPUT_DIR=./output)"
+    cmds:
+      - uv run python optimize_benchmark.py "{{.INPUT_DIR}}" "{{.OUTPUT_DIR}}"
+    requires:
+      vars: [INPUT_DIR, OUTPUT_DIR]
+
   install:
     desc: "Install Python dependencies using uv"
     cmds:

samples/python/api/__init__.py

@@ -1,7 +1,7 @@
 """API clients for Nitro Platform integrations."""
 
 from .base_client import BaseOAuthClient
-from .platform_api import PlatformAPIClient
+from .platform_api import JobFailedError, PlatformAPIClient
 from .sign_api import SignAPIClient
 
-__all__ = ["BaseOAuthClient", "PlatformAPIClient", "SignAPIClient"]
+__all__ = ["BaseOAuthClient", "JobFailedError", "PlatformAPIClient", "SignAPIClient"]

samples/python/api/platform_api.py

@@ -5,16 +5,66 @@
 
 import json
 import mimetypes
+import uuid
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 import httpx
 
 from .base_client import BaseOAuthClient
 
 if TYPE_CHECKING:
+    from collections.abc import Iterator
     from pathlib import Path
 
+# How long to wait for an asynchronous job to finish, in seconds.
+JOB_TIMEOUT_SECONDS = 900
+HTTP_OK = 200
+HTTP_ACCEPTED = 202
+
+
+class JobFailedError(RuntimeError):
+    """An asynchronous Platform API job failed.
+
+    Carries the detail needed to diagnose the failure: the HTTP status, the
+    error message returned by the API, and the request ID to quote to support.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        status_code: int | None = None,
+        error_type: str | None = None,
+        request_id: str | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+        self.error_type = error_type
+        self.request_id = request_id
+
+
+def _str_or_none(value: object) -> str | None:
+    """Return the value if it is a string, otherwise None."""
+    return value if isinstance(value, str) else None
+
+
+def _problem_detail(body: str) -> tuple[str | None, str | None]:
+    """Pull (error type, human message) out of a Platform API error body."""
+    try:
+        payload: object = json.loads(body)
+    except ValueError:
+        return None, None
+    if not isinstance(payload, dict):
+        return None, None
+    data = cast("dict[str, object]", payload)
+    error = data.get("error")
+    if isinstance(error, dict):
+        nested = cast("dict[str, object]", error)
+        return _str_or_none(nested.get("type")), _str_or_none(nested.get("title"))
+    return _str_or_none(data.get("type")), _str_or_none(data.get("title"))
+
 
 @dataclass
 class PlatformAPIClient(BaseOAuthClient):
@@ -80,6 +130,153 @@ def _request_bytes(
         download_response.raise_for_status()
         return download_response.content
 
+    def _iter_job_events(self, status_url: str, request_id: str) -> Iterator[dict[str, Any]]:
+        """Yield job events from the server-sent-events status stream."""
+        headers = {
+            "Authorization": f"Bearer {self._get_token()}",
+            "Accept": "text/event-stream",
+            "X-Analytics-Session-Id": request_id,
+        }
+        timeout = httpx.Timeout(30.0, read=float(JOB_TIMEOUT_SECONDS))
+        with self._client.stream("GET", status_url, headers=headers, timeout=timeout) as response:
+            if response.status_code not in {200, 202}:
+                response.read()
+                error_type, title = _problem_detail(response.text)
+                raise JobFailedError(
+                    title or f"Job status request failed with HTTP {response.status_code}",
+                    status_code=response.status_code,
+                    error_type=error_type,
+                    request_id=request_id,
+                )
+            data_lines: list[str] = []
+            for raw_line in response.iter_lines():
+                line = raw_line.rstrip("\r")
+                if line:
+                    if line.startswith("data:"):
+                        data_lines.append(line[5:].lstrip())
+                    continue
+                if data_lines:
+                    payload = "\n".join(data_lines).strip()
+                    data_lines = []
+                    if payload:
+                        yield cast("dict[str, Any]", json.loads(payload))
+            if data_lines:
+                payload = "\n".join(data_lines).strip()
+                if payload:
+                    yield cast("dict[str, Any]", json.loads(payload))
+
+    def _await_job(self, status_url: str, request_id: str) -> tuple[bool, str]:
+        """Follow a job to completion. Returns (failed, result location)."""
+        for event in self._iter_job_events(status_url, request_id):
+            status = event.get("status")
+            if status == "running":
+                continue
+            if status in {"completed", "failed"}:
+                return status == "failed", str(event["location"])
+        raise JobFailedError(
+            "The job status stream closed before the job finished.",
+            request_id=request_id,
+        )
+
+    def _request_async_bytes(
+        self,
+        endpoint: Literal["conversions", "extractions", "transformations"],
+        method: str,
+        file_path: Path,
+        params: dict[str, Any] | None = None,
+    ) -> bytes:
+        """Run an operation as an asynchronous job and return the resulting bytes.
+
+        Unlike the synchronous helpers, this submits the work with
+        ``Prefer: respond-async`` and then follows the job to completion, so
+        documents large enough to exceed the synchronous request window are
+        processed successfully instead of timing out on the client.
+
+        Raises:
+            JobFailedError: if the submission, the job, or the download fails.
+        """
+        request_id = str(uuid.uuid4())
+        headers = {
+            "Authorization": f"Bearer {self._get_token()}",
+            "Prefer": "respond-async",
+            "X-Analytics-Session-Id": request_id,
+        }
+        mime_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
+        files = {"file": (file_path.name, file_path.read_bytes(), mime_type)}
+        data = {"method": method, "params": json.dumps(params or {})}
+
+        submit = self._client.post(
+            f"{self._settings.platform_base_url}/{endpoint}",
+            headers=headers,
+            files=files,
+            data=data,
+        )
+        if submit.status_code != HTTP_ACCEPTED:
+            error_type, title = _problem_detail(submit.text)
+            raise JobFailedError(
+                title or f"Job submission failed with HTTP {submit.status_code}",
+                status_code=submit.status_code,
+                error_type=error_type,
+                request_id=request_id,
+            )
+
+        status_url = submit.headers.get("Location")
+        if not status_url:
+            raise JobFailedError(
+                "The job was accepted but no Location header was returned.",
+                status_code=submit.status_code,
+                request_id=request_id,
+            )
+
+        failed, result_url = self._await_job(status_url, request_id)
+        auth = {
+            "Authorization": f"Bearer {self._get_token()}",
+            "X-Analytics-Session-Id": request_id,
+        }
+
+        if failed:
+            error = self._client.get(result_url, headers=auth)
+            error_type, title = _problem_detail(error.text)
+            raise JobFailedError(
+                title or "The job failed.",
+                status_code=error.status_code,
+                error_type=error_type,
+                request_id=request_id,
+            )
+
+        # Without this Accept header the result endpoint returns the job's
+        # JSON representation rather than the output file itself.
+        result = self._client.get(
+            result_url, headers={**auth, "Accept": "application/octet-stream"}
+        )
+        if result.status_code != HTTP_OK:
+            error_type, title = _problem_detail(result.text)
+            raise JobFailedError(
+                title or f"Result download failed with HTTP {result.status_code}",
+                status_code=result.status_code,
+                error_type=error_type,
+                request_id=request_id,
+            )
+        return result.content
+
+    def optimize(self, file_path: Path, profile: str = "minimal-file-size") -> bytes:
+        """Optimize (compress) a PDF using an optimization profile.
+
+        Profiles are ``minimal-file-size``, ``web``, ``print``, ``archive`` and
+        ``mixed-raster-content``. This runs as an asynchronous job, so it works
+        for large documents as well as small ones.
+
+        Args:
+            file_path: The PDF to optimize.
+            profile: The optimization profile to apply.
+
+        Returns:
+            The optimized PDF as bytes.
+        """
+        return self._request_async_bytes(
+            "transformations", "optimize", file_path, {"profile": profile}
+        )
+
     def convert(self, file_path: Path, to_format: str) -> bytes:
         """Convert document to specified format."""
         return self._request_bytes("conversions", "convert", file_path, {"to": to_format})

samples/python/benchmark/__init__.py

@@ -0,0 +1,15 @@
+"""Optimize API benchmark: operations, and CSV/HTML reporting."""
+
+from __future__ import annotations
+
+from .operations import OperationResult, run_optimize
+from .report import summarise, write_csv, write_html, write_summary_csv
+
+__all__ = [
+    "OperationResult",
+    "run_optimize",
+    "summarise",
+    "write_csv",
+    "write_html",
+    "write_summary_csv",
+]