proofpudding 0.1.3

Python SDK for the Pudding API - Document processing and Q&A

Pudding SDK

A Python SDK for the Pudding API - Document processing and question answering.

Installation

pip install pudding-sdk

Requirements

• Python 3.10+
• httpx
• pydantic >= 2.0

Quick Start

from pudding import PuddingClient
from pudding.exceptions import NotFoundError, ValidationError

# Initialize client
client = PuddingClient(access_token="pk_your_api_key")

# Check API health
health = client.health.check()
print(f"API Status: {health.status}")

# Upload a document
doc = client.documents.upload(file_path="./contract.pdf")
print(f"Uploaded: {doc.id}")

# Ask a question about the document
job = client.jobs.create(
    document_id=doc.id,
    question="What is the effective date of this contract?"
)

if job.success:
    print(f"Answer: {job.result.answer}")
    print(f"Confidence: {job.result.confidence}")
    for citation in job.result.citations:
        print(f"  - Page {citation.page}: {citation.quote}")
else:
    print(f"Failed: {job.error}")

# List all documents with pagination
docs = client.documents.list(skip=0, limit=10)
print(f"Total documents: {docs.total}")
for doc in docs.items:
    print(f"  - {doc.filename} ({doc.size_bytes} bytes)")

# List jobs for a specific document
jobs = client.jobs.list(document_id=doc.id)

# Delete a document (also deletes associated jobs)
try:
    deleted = client.documents.delete(document_id=doc.id)
    print(f"Deleted: {deleted.filename}")
except NotFoundError:
    print("Document not found")

Async Support

The SDK provides both synchronous and asynchronous clients:

from pudding import PuddingClient, AsyncPuddingClient

# Synchronous client
client = PuddingClient(access_token="pk_your_api_key")
docs = client.documents.list()

# Asynchronous client
async_client = AsyncPuddingClient(access_token="pk_your_api_key")
docs = await async_client.documents.list()

Context Manager Support

Both clients support context managers for proper resource cleanup:

# Synchronous
with PuddingClient(access_token="pk_your_api_key") as client:
    docs = client.documents.list()

# Asynchronous
async with AsyncPuddingClient(access_token="pk_your_api_key") as client:
    docs = await client.documents.list()

Configuration

Initialization Options

client = PuddingClient(
    access_token="pk_your_api_key",  # Required: Your API key
    timeout=60.0,                     # Optional: Request timeout in seconds (default: 1800 for jobs)
    max_retries=3,                    # Optional: Max retries for transient errors (default: 3)
)

Custom Timeout for Jobs

Job creation is a blocking operation that waits for document processing. The default timeout is 1800 seconds (30 minutes):

# Use a custom timeout for the entire client
client = PuddingClient(access_token="...", timeout=3600.0)  # 1 hour

API Reference

Health Checks

# Check API health
health = client.health.check()
print(health.status)      # "healthy"
print(health.version)     # API version
print(health.environment) # Environment name

# Check readiness (includes database status)
ready = client.health.ready()
print(ready.status)    # "ready" or "not_ready"
print(ready.database)  # "connected" or error message

Documents

# Upload a document from file path
doc = client.documents.upload(file_path="/path/to/document.pdf")

# Upload a document from bytes
with open("document.pdf", "rb") as f:
    doc = client.documents.upload(file=f.read(), filename="document.pdf")

# List documents with pagination
doc_list = client.documents.list(skip=0, limit=20)
print(f"Total: {doc_list.total}")
for doc in doc_list.items:
    print(f"{doc.id}: {doc.filename}")

# Delete a document
deleted_doc = client.documents.delete(document_id="uuid-string")

Jobs

# Create a job (process document with a question)
job = client.jobs.create(
    document_id="uuid-string",
    question="What is the total revenue mentioned in this document?"
)

# Access job results
if job.success:
    print(job.result.answer)
    print(job.result.confidence)  # "high", "medium", "low", or "not_found"
    for citation in job.result.citations:
        print(f"Page {citation.page}: {citation.quote}")

# List all jobs
job_list = client.jobs.list(skip=0, limit=20)

# List jobs for a specific document
job_list = client.jobs.list(document_id="uuid-string")

Exception Handling

The SDK provides a comprehensive exception hierarchy:

from pudding.exceptions import (
    PuddingError,        # Base exception for all SDK errors
    AuthenticationError, # 401: Invalid or missing API key
    NotFoundError,       # 404: Resource not found
    ValidationError,     # 400: Invalid request
    RateLimitError,      # 429: Too many requests
    ServerError,         # 500: Internal server error
    GatewayError,        # 502: Upstream service error
    TimeoutError,        # 504: Request timed out
)

try:
    doc = client.documents.upload(file_path="./document.txt")
except ValidationError as e:
    print(f"Validation failed: {e.message}")  # "Only PDF files are allowed"
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")
except PuddingError as e:
    print(f"API error ({e.status_code}): {e.message}")

Logging

The SDK uses Python's standard logging module. Configure logging level as needed:

import logging

# Enable debug logging for the SDK
logging.getLogger("pudding").setLevel(logging.DEBUG)

Development

Setup

# Clone the repository
git clone https://github.com/pudding-ai/pudding-sdk.git
cd pudding-sdk

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

Running Tests

pytest

Running Tests with Coverage

pytest --cov=pudding --cov-report=html

Type Checking

mypy src/pudding

Linting

ruff check src/pudding tests

License

MIT License - see LICENSE for details.