Screenshot Client Batch

Complete screenshot service client and batch orchestration package. Provides both a low-level API client (ScreenshotServiceClient) and high-level coordinators for batch processing of screenshot jobs via the Azure-hosted screenshot service.

Built on top of the auto-generated screenshot-client-core, this package adds:

• Polling with configurable timeout/interval (using infra_core.polling)
• Environment-based configuration (from_env())
• Custom exception hierarchy for better error handling
• Batch coordination with chunking, parallel execution, and progress tracking
• CLI tool for processing JSONL files

Installation

pip install screenshot-client-batch

Requirements

• Python 3.11+
• Access to a screenshot service endpoint (Azure Function)

Environment Variables

Quick Start: API Client

from screenshot_batch import ScreenshotServiceClient

# Create client from environment variables
client = ScreenshotServiceClient.from_env()  # Uses SCREENSHOT_FUNC_BASE_URL, SCREENSHOT_FUNC_API_KEY

# Start a batch
handle = client.start_batch(
    "my-batch",
    {"jobs": [{"job_id": "1", "url": "https://example.com"}]}
)

# Poll until completion
result = client.poll_run(handle.batch_id, handle.job_id, timeout=600)
print(f"Status: {result['status']}")

# Get results
manifest = client.get_result(handle.batch_id, handle.job_id)

Batch Orchestration

For single batch runs with automatic polling and optional artifact download:

from pathlib import Path

from screenshot import ScreenshotOptions, CaptureOptions
from screenshot_batch import ScreenshotBatchCoordinator, ScreenshotJobSpec

coordinator = ScreenshotBatchCoordinator(
    base_url="https://site-screenshot.azurewebsites.net",
    batch_id="marketing-refresh",
    store_dir=Path("data/site-screens"),
    download_results=True,
)

jobs = [
    ScreenshotJobSpec.from_url(
        "https://example.com",
        options=ScreenshotOptions(
            capture=CaptureOptions(enabled=True, max_pages=3, depth=1),
        ),
    ),
]

result = coordinator.run_batch_sync(jobs)
print(result.status, result.job_completed)

Chunked Runs

For large collections (e.g., 4,000 URLs), use ScreenshotMultiRunCoordinator to fan out multiple API calls while keeping results grouped in per-run folders:

from screenshot_batch import (
    ScreenshotMultiRunCoordinator,
    ScreenshotJobSpec,
)

runner = ScreenshotMultiRunCoordinator(
    base_url="https://site-screenshot.azurewebsites.net",
    batch_id="yc-companies",
    store_dir=Path("data/site-screens"),
    download_results=True,
    upload_results=True,  # Optional: upload to Azure Storage
)

specs = [
    ScreenshotJobSpec.from_url("https://example.com", options=my_options),
    # ... more specs
]

summaries = runner.run_jobs(specs, chunk_size=4, max_parallel_runs=4)
for summary in summaries:
    print(summary.chunk_index, summary.result.status, summary.result.job_failed)

CLI

Install the package and run the CLI to process a JSONL file in chunks:

screenshot-client-batch input.jsonl my-batch-id \
  --base-url https://site-screenshot.azurewebsites.net \
  --chunk-size 4 \
  --max-parallel-runs 32 \
  --per-run-concurrency 4 \
  --store-dir data/screenshots \
  --download-results \
  --upload-results \
  --state-path data/state.json \
  --metadata '{"source":"my-project"}' \
  --output data/results.jsonl

CLI Options

Input Format

The input JSONL file should contain one JSON object per line:

{"url": "https://example.com", "job_id": "site-1", "metadata": {"category": "tech"}}
{"url": "https://another.com", "job_id": "site-2"}

Output Format

The output JSONL mirrors the input with added status fields:

{"url": "https://example.com", "job_id": "site-1", "status": "succeeded", "blob_url": "https://..."}
{"url": "https://another.com", "job_id": "site-2", "status": "failed", "errors": ["timeout"]}

Exception Handling

The package provides a custom exception hierarchy:

from screenshot_batch import (
    ScreenshotServiceError,  # Base exception
    ScreenshotConfigError,   # Configuration/environment errors
    ScreenshotAPIError,      # API communication errors
    ScreenshotTimeoutError,  # Polling timeout
)

try:
    result = client.poll_run(batch_id, job_id, timeout=60)
except ScreenshotTimeoutError:
    print("Batch did not complete in time")
except ScreenshotAPIError as e:
    print(f"API error: {e}")

Development

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Type checking
mypy src/screenshot_batch

# Linting
ruff check src/ tests/
ruff format src/ tests/

License

MIT