gravi-vision 0.1.4

Python client for Gravitate's Vision API - BOL and delivery receipt document scanning

Gravi-Vision Python Client

A Python client library for Gravitate's Vision API, which extracts structured data from BOL (Bill of Lading) and delivery receipt images using computer vision.

For more information, visit gravitate.energy

Features

• Synchronous scanning - Upload files and wait for results
• Asynchronous scanning - Upload files with callback for background processing
• Stateless callback handling - Kubernetes-friendly architecture
• Type-safe - Full type hints and Pydantic models
• Easy integration - Works with FastAPI for callback endpoints

Installation

pip install gravi-vision

Quick Start

Synchronous Scanning

from gravi_vision import GraviVisionClient

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

# Scan a BOL image
result = client.scan_bol(
    image_path="path/to/bol.jpg",
    reference_id="order-123"
)

print(result.items)  # List of line items
print(result.reference_id)  # "order-123"

Asynchronous Scanning with Callbacks

For long-running scans, use asynchronous mode where the server processes the image in the background and sends results to your callback endpoint.

Step 1: Set up a FastAPI callback handler

from fastapi import FastAPI
from gravi_vision import GraviVisionClient, EBolRequest

app = FastAPI()
client = GraviVisionClient(
    api_key="your-api-key",
    base_url="https://vision-dev.gravitate.energy",
    callback_url="https://your-app.example.com/callbacks",
    callback_auth_token="secret-token"
)

# Define your callback handler
def handle_bol_result(request: EBolRequest) -> None:
    """Process the scanned BOL data when it comes back from the server."""
    print(f"Received scan result for request_id: {request.request_id}")
    print(f"Reference ID: {request.reference_id}")
    print(f"Items: {request.items}")

    # Store result in database, update order status, etc.
    # use reference_id to correlate with your business data
    order_id = request.reference_id
    # ... your logic here

# Mount the callback router at your desired path
callback_router = client.get_callback_router(
    handler=handle_bol_result,
    path_prefix="/api/v1"
)
app.include_router(callback_router)

Step 2: Upload image for processing

# Initiate async scan - returns immediately with request_id
response = client.scan_bol_async(
    image_path="path/to/bol.jpg",
    reference_id="order-123"  # Use your business identifier for correlation
)

print(response["request_id"])    # Unique scan request ID
print(response["reference_id"])  # "order-123"

# The server processes the image in the background
# When done, it POSTs the result to your callback endpoint
# Your handle_bol_result function will be called automatically

Architecture

Callback Design

The callback system is stateless and Kubernetes-friendly:

1. Client generates a request_id for each scan
2. Client calls scan_bol_async() which returns immediately
3. Server processes image in background
4. Server POSTs result to {callback_url}/gravi-vision-callbacks/{request_id}
5. Your app's CallbackRouter validates the request and calls your handler
6. Handler uses reference_id to correlate result with your business data

Key benefit: No state stored on the client. Works seamlessly with:

• Serverless deployments
• Kubernetes auto-scaling
• Multiple app replicas
• Load balancers

API Reference

GraviVisionClient

Main client class for interacting with the Gravi-Vision API.

Initialization

client = GraviVisionClient(
    api_key: str,                          # API key for authentication
    base_url: str = "https://vision-dev.gravitate.energy",  # API server URL
    callback_url: str | None = None,       # Your app's callback URL
    callback_auth_token: str | None = None,   # Bearer token for callback auth
    default_callback_handler: Callable | None = None  # Default handler function
)

Methods

health_check() -> dict Check if the API is healthy.

health = client.health_check()
print(health["status"])  # "ok"

scan_bol(bol_files: str | Path | BinaryIO | tuple[BinaryIO, str] | list, reference_id: str | None = None) -> EBolRequest Synchronously scan a BOL image and return results immediately.

# From file path
result = client.scan_bol(
    bol_files="path/to/bol.pdf",
    reference_id="order-123"
)

# From file-like object (BytesIO, UploadFile, etc.)
from io import BytesIO
file_bytes = b"..."
result = client.scan_bol(
    bol_files=BytesIO(file_bytes),
    reference_id="order-123"
)

# From BytesIO with custom filename
result = client.scan_bol(
    bol_files=(BytesIO(file_bytes), "invoice_12345.pdf"),
    reference_id="order-123"
)

# Multiple files mixed
result = client.scan_bol(
    bol_files=[
        "file1.pdf",
        BytesIO(data1),
        (BytesIO(data2), "custom_name.pdf")
    ],
    reference_id="order-123"
)
print(result.bol_number)  # Extracted BOL number

scan_bol_async(bol_files: str | Path | BinaryIO | tuple[BinaryIO, str] | list, reference_id: str | None = None) -> dict Asynchronously scan a BOL image. Returns immediately with request_id.

# From file path
response = client.scan_bol_async(
    bol_files="path/to/bol.pdf",
    reference_id="order-123"
)

# From BytesIO with custom filename (no disk write!)
from io import BytesIO
file_bytes = get_from_database()  # Never touches disk
response = client.scan_bol_async(
    bol_files=(BytesIO(file_bytes), "invoice_12345.pdf"),
    reference_id="order-123"
)
# Server will process in background and POST result to your callback

get_callback_router(handler: Callable, path_prefix: str = "") -> APIRouter Get a FastAPI router for receiving callbacks.

async def my_handler(data: EBolRequest):
    print(f"Got result for {data.reference_id}")

router = client.get_callback_router(
    handler=my_handler,
    path_prefix="/api/v1"
)
app.include_router(router)

Models

EBolRequest Scanned BOL data with line items.

class EBolRequest(BaseModel):
    request_id: str | None = None    # Unique scan request ID
    reference_id: str | None = None  # Your business identifier (e.g., order_id)
    items: list[EBolDetailRequest]   # Line items scanned from the document

EBolDetailRequest A single line item from a BOL.

class EBolDetailRequest(BaseModel):
    item_code: str | None = None
    description: str | None = None
    quantity: int | None = None
    unit: str | None = None
    weight: float | None = None
    # ... other fields

VisionResponse Generic API response wrapper.

class VisionResponse(BaseModel):
    success: bool
    data: EBolRequest | None = None
    error: str | None = None

Context Manager Usage

Use the client as a context manager for automatic resource cleanup:

async with GraviVisionClient(api_key="key") as client:
    result = await client.scan_bol("image.jpg")

Error Handling

from httpx import HTTPError

try:
    result = client.scan_bol("bol.jpg")
except HTTPError as e:
    print(f"API error: {e}")

Configuration

Configure via environment variables or parameters:

import os
from gravi_vision import GraviVisionClient

client = GraviVisionClient(
    api_key=os.getenv("GRAVI_API_KEY"),
    base_url=os.getenv("GRAVI_API_URL", "http://localhost:8000"),
    callback_url=os.getenv("GRAVI_CALLBACK_URL"),
    callback_auth_token=os.getenv("GRAVI_CALLBACK_TOKEN")
)

Development

Running Tests

pytest

Type Checking

mypy .

Formatting and Linting

ruff check --fix .
ruff format .

License

Proprietary - See gravitate.energy for licensing information.

Support

For issues, feature requests, or inquiries about the Vision API, visit:

• GitHub Issues
• Gravitate Energy