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 staged upload 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_upload_key("staging/acc/uuid")  # reference a staged upload by its upload_key

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)
job.request_id                                  # correlation id; also on the result meta + webhook
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 velrim.TimeoutError at the deadline (measured against the wall clock; it subclasses the builtin TimeoutError too, so existing except TimeoutError blocks keep working).

Retries and idempotency

Failed requests are retried automatically: 2 additional attempts by default (max_retries client option; 0 disables), on connection errors, transport timeouts, and HTTP 408 / 429 / 5xx, with exponential backoff (0.5s, 1s, ... capped at 8s). A Retry-After header is honored when it is ≤ 60 seconds. 409 is never retried — on this API a 409 means the idempotency key was reused with different parameters (or the first request is still running), a deterministic answer that a re-send cannot change; this deliberately differs from SDK generators that retry 409s.

Retrying the two money-adjacent POSTs (/v1/extract, /v1/jobs) is safe because the SDK attaches an Idempotency-Key header (a fresh UUID per logical request, or your own via idempotency_key=...) and reuses the same key across every retry of that request — the server replays the first stored success instead of executing (and charging) twice.

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_upload_key(completed.upload_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

Velrim signs deliveries with the Standard Webhooks scheme: webhook-id, webhook-timestamp, and webhook-signature headers, an HMAC-SHA256 over {id}.{timestamp}.{body}, and a whsec_... secret. verify_webhook recomputes the HMAC over the raw request body, compares it in constant time against every v1 entry in the (space-delimited) signature list, and rejects a stale timestamp (±300s by default). Pass the raw request body (never a re-serialized dict — reordered keys break the signature) and the delivery's headers whole (lookup is case-insensitive):

from velrim import verify_webhook, WebhookVerificationError

try:
    event = verify_webhook(raw_body, request.headers, webhook_secret)
except WebhookVerificationError as e:
    e.reason  # "malformed" | "stale" | "secret" | "signature" | "body"
    return 400
# event.type is "job.completed" or "job.failed"; event.result_url is set on completed only.

Use the webhook-id header as your dedupe key: redeliveries of the same event reuse it. To get a boolean instead of an exception, wrap it:

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

License

MIT.