Metadata-Version: 2.4
Name: docutray
Version: 0.3.0
Summary: Python library for the DocuTray API
Keywords: docutray,ocr,document processing,data extraction,api
Author: Roberto Arce
Author-email: Roberto Arce <robertoarcee@gmail.com>
License-Expression: MIT
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Dist: httpx>=0.23.0,<1
Requires-Dist: pydantic>=2.0,<3
Requires-Dist: typing-extensions>=4.5
Requires-Dist: griffe>=1.0 ; extra == 'docs'
Requires-Python: >=3.10
Project-URL: Homepage, https://docutray.com
Project-URL: Documentation, https://docs.docutray.com
Project-URL: Repository, https://github.com/docutray/docutray-python
Project-URL: Changelog, https://github.com/docutray/docutray-python/blob/main/CHANGELOG.md
Project-URL: Bug Tracker, https://github.com/docutray/docutray-python/issues
Provides-Extra: docs
Description-Content-Type: text/markdown

# DocuTray Python Library

[![PyPI version](https://img.shields.io/pypi/v/docutray.svg)](https://pypi.org/project/docutray/)
[![Python versions](https://img.shields.io/pypi/pyversions/docutray.svg)](https://pypi.org/project/docutray/)
[![License](https://img.shields.io/pypi/l/docutray.svg)](https://github.com/docutray/docutray-python/blob/main/LICENSE)

The official Python library for the [DocuTray API](https://docutray.com), providing access to document processing capabilities including OCR, document identification, data extraction, and knowledge bases.

## Documentation

Full API documentation is available at [docs.docutray.com](https://docs.docutray.com).

## Installation

```bash
pip install docutray
```

For API reference generation tools (maintainers only):

```bash
pip install docutray[docs]
```

### Requirements

- Python 3.10+

## Quick Start

### Synchronous Usage

```python
from pathlib import Path
from docutray import Client

client = Client(api_key="your-api-key")

# Convert a document
result = client.convert.run(
    file=Path("invoice.pdf"),
    document_type_code="invoice"
)
print(result.data)

client.close()
```

### Asynchronous Usage

```python
import asyncio
from pathlib import Path
from docutray import AsyncClient

async def main():
    async with AsyncClient(api_key="your-api-key") as client:
        result = await client.convert.run(
            file=Path("invoice.pdf"),
            document_type_code="invoice"
        )
        print(result.data)

asyncio.run(main())
```

## Configuration

### API Key

Set your API key via constructor argument or environment variable:

```python
# Via constructor
client = Client(api_key="your-api-key")

# Via environment variable
# export DOCUTRAY_API_KEY="your-api-key"
client = Client()  # Reads from DOCUTRAY_API_KEY
```

### Base URL

Override the default API endpoint:

```python
client = Client(
    api_key="your-api-key",
    base_url="https://custom-api.example.com"
)
```

### Timeout

Configure request timeouts (in seconds):

```python
import httpx

# Simple timeout (applies to all operations)
client = Client(api_key="your-api-key", timeout=30.0)

# Granular timeout control
client = Client(
    api_key="your-api-key",
    timeout=httpx.Timeout(
        connect=5.0,
        read=60.0,
        write=60.0,
        pool=10.0
    )
)
```

### Retries

Configure automatic retry behavior:

```python
# Default: 2 retries with exponential backoff
client = Client(api_key="your-api-key")

# Custom retry count
client = Client(api_key="your-api-key", max_retries=5)

# Disable retries
client = Client(api_key="your-api-key", max_retries=0)
```

## Error Handling

The SDK provides a comprehensive exception hierarchy:

```
DocuTrayError (base)
├── APIConnectionError (network errors)
│   └── APITimeoutError (request timeout)
└── APIError (HTTP errors)
    ├── BadRequestError (400)
    │   └── DSLValidationError (invalid DSL rules, with line/column diagnostics)
    ├── AuthenticationError (401)
    ├── PermissionDeniedError (403)
    ├── NotFoundError (404)
    ├── ConflictError (409)
    ├── UnprocessableEntityError (422)
    ├── RateLimitError (429)
    └── InternalServerError (5xx)
```

### Catching Errors

```python
from pathlib import Path
from docutray import Client
from docutray import (
    DocuTrayError,
    APIConnectionError,
    APIError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
)

client = Client(api_key="your-api-key")

try:
    result = client.convert.run(
        file=Path("document.pdf"),
        document_type_code="invoice"
    )
except AuthenticationError as e:
    print(f"Invalid API key: {e.message}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except NotFoundError as e:
    print(f"Resource not found: {e.message}")
except APIError as e:
    print(f"API error {e.status_code}: {e.message}")
    print(f"Request ID: {e.request_id}")
except APIConnectionError as e:
    print(f"Connection failed: {e.message}")
except DocuTrayError as e:
    print(f"SDK error: {e.message}")
```

### Rate Limit Handling

```python
import time
from pathlib import Path
from docutray import Client, RateLimitError

client = Client(api_key="your-api-key")

try:
    result = client.convert.run(file=Path("document.pdf"), document_type_code="invoice")
except RateLimitError as e:
    if e.retry_after:
        print(f"Rate limited. Waiting {e.retry_after} seconds...")
        time.sleep(e.retry_after)
        # Retry the request
```

## Resources

### Convert

Convert documents to structured data using OCR and AI extraction.

```python
from pathlib import Path

# Synchronous conversion (waits for result)
result = client.convert.run(
    file=Path("invoice.pdf"),
    document_type_code="invoice"
)
print(result.data)

# From bytes
with open("invoice.pdf", "rb") as f:
    result = client.convert.run(
        file=f.read(),
        document_type_code="invoice"
    )

# From URL
result = client.convert.run(
    url="https://example.com/invoice.pdf",
    document_type_code="invoice"
)

# Asynchronous conversion (returns immediately)
status = client.convert.run_async(
    file=Path("large_document.pdf"),
    document_type_code="invoice"
)
print(f"Conversion ID: {status.conversion_id}")

# Poll for completion
final_status = status.wait()
if final_status.is_success():
    print(final_status.data)
```

#### Validation results

Successful conversions include `validation`: the result of checking `data` against the
document type's JSON Schema and [DSL validation rules](https://docs.docutray.com/docs/guides/validation).
It is informational and never changes the conversion status.

```python
from docutray import ValidationOutcome, ValidationSystemFailure

result = client.convert.run(file=Path("invoice.pdf"), document_type_code="invoice")

if isinstance(result.validation, ValidationOutcome):
    print("Valid:", result.validation.valid)
    for issue in result.validation.errors + result.validation.warnings:
        print(f"[{issue.severity}] {issue.field or '<root>'}: {issue.message} ({issue.source})")
elif isinstance(result.validation, ValidationSystemFailure):
    print("Validation could not be completed")
elif result.validation is None:
    print("Validation skipped (no schema and no rules)")
else:
    print("Unrecognized validation shape:", result.validation)  # plain dict
```

The same `validation` field is available on `ConversionStatus` (async conversions).
When the document type has bounding boxes enabled, `result.field_locations` contains
the page and box of each extracted field.

### Identify

Automatically identify document types.

```python
# document_type_code_options is required: the candidate types (at least one)
result = client.identify.run(
    file=Path("unknown_document.pdf"),
    document_type_code_options=["acme_invoice", "acme_receipt"],
)

print(f"Identified as: {result.document_type.name}")
print(f"Confidence: {result.document_type.confidence:.2%}")

# View alternatives
for alt in result.alternatives:
    print(f"  Alternative: {alt.name} ({alt.confidence:.2%})")

# Attach your own metadata
result = client.identify.run(
    file=Path("unknown_document.pdf"),
    document_type_code_options=["acme_invoice", "acme_receipt"],
    document_metadata={"customer_id": "12345"},
)
```

### Document Types

List, create, update, and manage document type definitions.

Model fields keep the API's camelCase names (`codeType`, `jsonSchema`, `isDraft`),
while method arguments are snake_case (`code_type=`, `json_schema=`, `is_draft=`).

`get()`, `update()` and `validate()` take the document type's internal `id` (a cuid
such as `"cm0example0000000000000id"`), not its `codeType`; passing a code returns 404.
Use `get_by_code()` to fetch a type by its code. The API prefixes the codes of your
organization's types with its slug (`code_type="invoice"` is stored as `acme_invoice`),
so pass the full `codeType` there and wherever a code is expected, such as
`document_type_code=` in `convert`.

```python
# List all document types
page = client.document_types.list()
for doc_type in page.data:
    print(f"{doc_type.codeType}: {doc_type.name}")

# Search document types (matches name, codeType or description)
page = client.document_types.list(search="invoice")

# Get a document type by its full codeType (raises NotFoundError if missing)
doc_type = client.document_types.get_by_code("acme_invoice")
type_id = doc_type.id
print(f"Schema: {doc_type.jsonSchema}")

# Create a custom document type
doc_type = client.document_types.create(
    name="Purchase Order",
    code_type="purchase_order",  # stored as "<org-slug>_purchase_order"
    description="Purchase orders with line items",
    json_schema={
        "type": "object",
        "properties": {
            "po_number": {"type": "string"},
            "total": {"type": "number"},
        },
    },
    conversion_mode="json",
)
print(f"Created: {doc_type.id}")

# Update a document type
doc_type = client.document_types.update(
    type_id,
    name="Updated Invoice",
    is_draft=False,
)

# Validate data against a document type (JSON Schema + saved DSL rules)
validation = client.document_types.validate(
    type_id,
    {"invoice_number": "INV-001", "total": 100.00}
)
if validation.is_valid():
    print("Data is valid!")
else:
    for issue in validation.issues:
        print(f"[{issue.severity}] {issue.field}: {issue.message}")
```

#### DSL validation rules

Document types can carry [DSL validation rules](https://docs.docutray.com/docs/guides/validation)
that run after extraction, in addition to the JSON Schema.

```python
from docutray import DSLValidationError

type_id = client.document_types.get_by_code("acme_invoice").id

rules = """
# The total must be the subtotal plus tax
define expected sum field.subtotal field.tax
field.total equalsWithPrecision(calculated.expected, 2)
field.currency in("CLP", "USD")
"""

# Try rules without saving them (dry run): they replace the saved rules for this call only
result = client.document_types.validate(
    type_id,
    {"subtotal": 1000, "tax": 190, "total": 1290, "currency": "CLP"},
    dsl_rules=rules,
)
print(result.valid, [issue.message for issue in result.issues])

# Save rules on the document type (also accepted by create())
try:
    doc_type = client.document_types.update(type_id, dsl_rules=rules)
    print(doc_type.dslRules)
except DSLValidationError as e:
    for d in e.diagnostics:
        print(f"line {d.line}, column {d.column}: {d.message}")

# Omitting dsl_rules keeps the saved rules; None or "" clears them
client.document_types.update(type_id, dsl_rules=None)
```

### Steps

Execute predefined document processing workflows.

```python
# Start async step execution
status = client.steps.run_async(
    step_id="step_invoice_extraction",
    file=Path("invoice.pdf")
)

# Wait for completion with progress callback
def on_progress(s):
    print(f"Status: {s.status}")

final = status.wait(on_status=on_progress)
print(final.data)
```

### Knowledge Bases

Manage document collections with semantic search capabilities.

```python
# List knowledge bases
for kb in client.knowledge_bases.list().auto_paging_iter():
    print(f"{kb.name}: {kb.document_count} documents")

# Create a knowledge base
kb = client.knowledge_bases.create(
    name="Product Documentation",
    description="Technical documentation for products",
    schema={
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "content": {"type": "string"},
            "category": {"type": "string"}
        }
    }
)

# Add documents
doc = client.knowledge_bases.documents(kb.id).create(
    document_id="getting-started",  # your own unique ID, required
    content={
        "title": "Getting Started",
        "content": "Welcome to our product...",
        "category": "guides"
    },
    metadata={"source": "manual"}
)

# Semantic search
results = client.knowledge_bases.search(
    kb.id,
    query="how to configure authentication",
    limit=5
)
for item in results.data:
    print(f"{item.similarity:.2%}: {item.document.content['title']}")

# Update a document
# get(), update() and delete() take the documentId given on create, not doc.id
client.knowledge_bases.documents(kb.id).update(
    doc.documentId,
    content={"title": "Updated Title", "content": "..."}
)

# Filter documents whose documentId contains a value (case-insensitive);
# use get(document_id) for an exact match
page = client.knowledge_bases.documents(kb.id).list(document_id="getting")

# Delete a document
client.knowledge_bases.documents(kb.id).delete(doc.documentId)

# Re-sync, forcing every embedding to be regenerated. sync() is not retried by
# default (each attempt is billed); pass max_retries= to opt in
client.knowledge_bases.sync(kb.id, force=True)

# Delete knowledge base
client.knowledge_bases.delete(kb.id)
```

### Webhooks

Parse webhook deliveries with the payload models. The event type is in the
`X-Docutray-Event` header.

```python
from docutray import ConversionWebhookPayload, ValidationOutcome
from docutray.types import IdentificationWebhookPayload, StepWebhookPayload

payload = ConversionWebhookPayload.model_validate_json(request_body)
if payload.status == "SUCCESS" and isinstance(payload.validation, ValidationOutcome):
    print(payload.validation.valid)
```

`StepWebhookPayload.validation` (and `StepExecutionStatus.validation`) keep the summarized
`{errors: {count, messages}, warnings: {count, messages}}` format.

## Pagination

Resources that return lists support pagination:

```python
# Get the first page
page = client.document_types.list(limit=10)
print(f"Page {page.page} of {page.total_pages}")

# Iterate through all pages manually
for page in client.document_types.list().iter_pages():
    for doc_type in page.data:
        print(doc_type.name)

# Auto-iterate through all items
for doc_type in client.document_types.list().auto_paging_iter():
    print(doc_type.name)

# Async pagination (inside an async function)
# async for doc_type in (await client.document_types.list()).auto_paging_iter_async():
#     print(doc_type.name)
```

## Raw Response Access

Access raw HTTP response data for debugging:

```python
from pathlib import Path
from docutray import Client

client = Client(api_key="your-api-key")

response = client.convert.with_raw_response.run(
    file=Path("invoice.pdf"),
    document_type_code="invoice"
)

print(f"Status: {response.status_code}")
print(f"Headers: {response.headers}")
print(f"Request ID: {response.headers.get('x-request-id')}")

# Parse the response body
result = response.parse()
print(result.data)
```

## Async Operations

For long-running operations, use async methods with polling:

```python
from pathlib import Path
from docutray import Client

client = Client(api_key="your-api-key")

# Start async conversion
status = client.convert.run_async(
    file=Path("large_document.pdf"),
    document_type_code="invoice"
)

# Poll with progress callback
def on_status(s):
    print(f"Status: {s.status}, Progress: {s.progress or 'N/A'}")

final = status.wait(
    on_status=on_status,
    poll_interval=2.0,  # seconds between polls
    timeout=300.0       # maximum wait time
)

if final.is_success():
    print("Conversion complete!")
    print(final.data)
elif final.is_failed():
    print(f"Conversion failed: {final.error}")
```

## Type Safety

The SDK uses Pydantic models for all responses, providing full type safety:

```python
from pathlib import Path
from docutray import Client
from docutray.types import ConversionResult, DocumentType

client = Client(api_key="your-api-key")

# Type hints work with your IDE
result: ConversionResult = client.convert.run(
    file=Path("invoice.pdf"),
    document_type_code="invoice"
)

# Access typed attributes
print(result.conversion_id)  # str
print(result.data)           # dict[str, Any]
print(result.status)         # str
```

## Contributing

We welcome contributions! Here's how to get started:

### Development Setup

```bash
# Clone the repository
git clone https://github.com/docutray/docutray-python.git
cd docutray-python

# Install dependencies with uv
uv sync

# Run tests
uv run pytest

# Run tests with coverage
uv run pytest --cov=src/docutray

# Type checking
uv run mypy src

# Linting
uv run ruff check src

# Format code
uv run ruff format src
```

### Running Tests

```bash
# All tests
uv run pytest

# Specific test file
uv run pytest tests/test_client.py

# Specific test
uv run pytest tests/test_client.py::test_client_initialization
```

## Support

- [Documentation](https://docs.docutray.com)
- [API Reference](https://docs.docutray.com/api)
- [Issue Tracker](https://github.com/docutray/docutray-python/issues)

## License

MIT
