velrim 0.0.2

Python SDK for the Velrim document-extraction API.

velrim

Python SDK for the Velrim document-extraction API.

Structured extraction against a JSON Schema you supply — pass a Pydantic model and get a validated instance of it back, with a per-field state (present / null / missing) and a source anchor (page + bounding box) you can audit.

• Runtime dependency: pydantic>=2 only. The HTTP transport is the standard library's urllib behind an injectable Transport seam — no other runtime dependency.
• Requires Python 3.9+.

Install

pip install velrim
# or
uv add velrim

Quickstart

from pydantic import BaseModel
from velrim import Client, Document


class Invoice(BaseModel):
    invoice_number: str
    total: float


with Client(api_key="...") as v:            # context manager; reads VELRIM_API_KEY if omitted
    r = v.extract(document=Document.from_path("invoice.pdf"), schema=Invoice)
    inv: Invoice = r.parsed                  # already model_validate()'d, fully typed
    if r.fields["/total"].state != "present":
        ...                                  # branch on the field's anchor / confidence

r.data is the raw extracted object (a dict); r.fields maps an RFC-6901 JSON Pointer ("/total", "/line_items/0/sku") to a per-leaf ResponseField carrying its state (present / null / missing), an optional confidence score the calibrator emits, and an optional anchor (page + bounding box). r.meta carries the request metadata. r.parsed is set only when you pass a Pydantic model as the schema.

Documents

A document input is always explicit — a bare str is rejected because it is ambiguous between a filesystem path and a pre-staged R2 key:

from velrim import Document

Document.from_path("invoice.pdf")     # read a file and inline it as base64
Document.from_bytes(pdf_bytes)        # inline raw bytes
Document.from_r2_key("staging/acc/uuid")   # reference a pre-staged R2 object

extract(...) also accepts bytes or os.PathLike directly; it never accepts a bare str.

Schemas

from velrim import to_json_schema

to_json_schema(Invoice)                 # default mode="validation"

to_json_schema defaults to mode="validation": the schema you send must match what Model.model_validate(result.data) accepts on the way back. Nested models become $defs + $ref (Draft 2020-12). For a discriminated union (Field(discriminator="kind")), the schema carries a oneOf plus a discriminator block, and the SDK surfaces it into options.hints for you.

Discriminated unions and reverse narrowing — read this before relying on result.parsed

result.data is the raw extracted object exactly as Velrim returned it, and result.fields carries the per-field state/anchor for every leaf. These are the source of truth.

When you pass a Pydantic model with a discriminated union (Field(discriminator="kind") — recommended over a callable Discriminator/Tag, which historically emits a bare anyOf with no discriminator block, pydantic #7491/#8628), to_json_schema() emits oneOf + a discriminator block and the SDK surfaces it into options.hints. On the way back, result.parsed = Model.model_validate(result.data) narrows the dict to the selected branch: with Pydantic's default extra="ignore", keys belonging only to a sibling branch — or any key outside the chosen shape — are silently dropped. So result.parsed is faithful to the narrowed shape, not to the original document bytes. A non-discriminated (smart) union can also silently rebind to a different branch. If byte-fidelity matters, keep result.data (the raw dict) — it is never narrowed. Treat result.parsed as the typed, narrowed convenience view.

Difference from the TypeScript SDK: the TypeScript sibling rejects the extraction-unsafe Zod subset loudly (a throw). Pydantic always emits a schema, so there is no equivalent throw; this asymmetry is intentional and the lossiness above is documented instead. (Pydantic also emits oneOf for discriminated unions where Zod emits anyOf — both carry the discriminator the extractor needs.)

Jobs (async / batch)

job = v.jobs.create(document=Document.from_path("invoice.pdf"), schema=Invoice)
status = v.jobs.get(job.job_id)                 # JobRunning | JobFailed | JobSucceeded
result = v.jobs.poll(job.job_id, schema=Invoice)  # blocks until terminal; raises on failure/timeout

poll loops get() on the configured interval until the job succeeds (returns an ExtractResult), raises the mapped error on failure, or raises TimeoutError at the deadline.

Large documents

plan = v.uploads.create(content_length=len(big_pdf), content_type="application/pdf")
uploaded = []
offset = 0
for part in plan.parts:
    data = big_pdf[offset : offset + part.size]
    uploaded.append(v.uploads.upload_part(plan, part_number=part.part_number, data=data))
    offset += part.size
completed = v.uploads.complete(plan, parts=uploaded)
r = v.extract(document=Document.from_r2_key(completed.r2_key), schema=Invoice)

Errors

Every non-2xx response raises a typed subclass of APIError (itself a VelrimError): one per ErrorCode — InvalidSchemaError, DocumentTooLargeError, UnsupportedDocumentError, InvalidAPIKeyError, InsufficientBalanceError, IdempotencyKeyConflictError, ExtractionFailedError, RateLimitedError, ProviderUnavailableError, NotFoundError, and InternalError (also the fallback for any unknown code). Each carries status_code, code, message, and request_id; InsufficientBalanceError.top_up_url and RateLimitedError.retry_after are populated when present. Transport failures raise APIConnectionError / APITimeoutError.

from velrim import APIError, RateLimitedError

try:
    v.extract(document=Document.from_path("invoice.pdf"), schema=Invoice)
except RateLimitedError as e:
    wait = e.retry_after
except APIError as e:
    print(e.status_code, e.code, e.message, e.request_id)

Webhooks

verify_webhook recomputes the HMAC-SHA256 over the raw request body, compares it in constant time, and rejects a stale timestamp. Pass the raw request body (never a re-serialized dict — reordered keys break the signature):

from velrim import verify_webhook, WebhookVerificationError

try:
    event = verify_webhook(raw_body, request.headers["X-Velrim-Signature"], webhook_secret)
except WebhookVerificationError:
    return 400
# event.type is "job.completed" or "job.failed"; event.result_url is set on completed only.

To get a boolean instead of an exception, wrap it:

def is_valid(body, header, secret) -> bool:
    try:
        verify_webhook(body, header, secret)
        return True
    except WebhookVerificationError:
        return False

License

Apache-2.0.