scriva.extract

scriva.extract

scriva.extract is the schema-first entry point — give it a file and a pydantic schema, get a typed instance back. It is a thin convenience on top of presets and the underlying Orchestrator.

import scriva
from scriva.schemas import Invoice

invoice: Invoice = scriva.extract("acme-2026-05.pdf", schema=Invoice)
print(invoice.total_amount, invoice.line_items[0].description)

Under the hood extract picks the right preset (and therefore the right Orchestrator recipe + Worker + prompt) for your schema, runs the recipe, validates against the model, and returns the model instance. If you want the full DocumentResult (regions, confidence, page-level views), pass return_="result":

result = scriva.extract("acme.pdf", schema=Invoice, return_="result")
invoice = result.fields                                  # the typed instance
result.fields_with_provenance["total_amount"].region    # which box did it come from

Signature

def extract(
    source: str | Path | bytes | Document | Sequence[str | Path],
    *,
    schema: type[BaseModel] | None = None,
    preset: str | None = None,                # mutually exclusive with schema
    locale: str = "auto",                     # "auto" sniffs from the image
    model: str | None = None,                 # override the default Worker model
    cache: Cache | str | None = ".scriva_cache",
    return_: Literal["fields", "result"] = "fields",
    on_event: Callable[[Event], None] | None = None,
    timeout_s: float | None = None,
    max_cost_usd: float | None = None,
) -> BaseModel | DocumentResult | list[BaseModel]: ...

Pass a sequence as source and you get a list back. See Batch extraction below.

Three ways to call it

1. With a schema

from scriva.schemas import Receipt
receipt = scriva.extract("receipt.jpg", schema=Receipt)

extract infers the splitter and prompt from the schema:

• A “flat” schema (no nested grid types) gets .split.whole_page() and Prompt.structured(schema=...).
• A schema with a line_items: list[...] field gets a table-aware recipe (.split.grid() with whole-page fallback).
• A schema subclass of scriva.schemas.TabularForm gets the full forms recipe (grid split + cell-by-cell recognise + dictionary reconstruct).

2. With a preset

receipt = scriva.extract("receipt.jpg", preset="receipt")
invoice = scriva.extract("invoice.pdf", preset="invoice", locale="en")

A preset is a tuned (Orchestrator recipe, schema) bundle. See Presets for the catalogue. preset= and schema= are mutually exclusive — the preset already pins a schema.

3. With auto-classification

When you don’t know what kind of document it is:

result = scriva.extract("unknown.pdf")          # no schema, no preset
result.kind                                      # -> DocumentKind.INVOICE
result.fields                                    # typed against the matching schema

Without schema or preset, extract first calls scriva.classify_document and routes to the matching preset. Useful for ingest pipelines that accept “whatever the user uploaded.”

You can also classify explicitly:

kind = scriva.classify_document("unknown.pdf")  # -> DocumentKindGuess
if kind.confidence < 0.7:
    queue_for_human_review(kind)
else:
    fields = scriva.extract("unknown.pdf", schema=kind.schema)

classify_document

def classify_document(
    source: str | Path | Document,
    *,
    candidates: Sequence[DocumentKind] | None = None,    # restrict to a subset
    model: str | None = None,
) -> DocumentKindGuess: ...

class DocumentKindGuess(BaseModel):
    kind: DocumentKind                  # enum value
    confidence: float
    schema: type[BaseModel] | None      # the matching schema, if any
    runner_up: DocumentKind | None      # second-best guess

DocumentKind is an enum: INVOICE, RECEIPT, ID_CARD, PASSPORT, BUSINESS_CARD, CONTRACT, BANK_STATEMENT, RESUME, MEDICAL_FORM, PRESCRIPTION, PURCHASE_ORDER, SHIPPING_LABEL, CHEQUE, TABULAR_FORM, UNKNOWN.

Each enum value carries its built-in schema via kind.schema — that’s how extract resolves the routing.

Batch extraction

results: list[Invoice] = scriva.extract(
    ["a.pdf", "b.pdf", "c.pdf"],
    schema=Invoice,
    max_concurrency=4,
)

Or use the explicit scriva.batch when you want more control:

batch = scriva.batch(
    sources=Path("incoming/").glob("*.pdf"),
    schema=Invoice,
    max_concurrency=4,
    on_each=lambda src, invoice: notify_billing(invoice),
    on_error="continue",                     # "raise" | "continue" | "collect"
    progress=True,                           # tqdm bar
    resume_file=".scriva_batch.json",        # idempotent restart
    max_cost_usd=20.0,                       # hard cap; raises BudgetExceeded
)

for src, invoice_or_error in batch:
    ...

The resume_file is a JSONL ledger; restarting with the same path skips sources whose hash + schema already succeeded. That makes the batch idempotent across crashes.

Watch a folder

scriva.watch(
    folder="./incoming",
    to="./extracted",
    schema=Invoice,
    pattern="*.pdf",
    move_on_success=True,                    # consume the input
    move_on_error="./failed",                # quarantine
    on_each=lambda src, dst: print(f"{src} -> {dst}"),
)

watch blocks until interrupted. It uses watchdog for filesystem events and falls back to polling on platforms without inotify (poll_interval_s=2.0). Each new file is processed exactly once thanks to a sidecar ledger.

Output adapters beyond a destination folder:

scriva.watch("./incoming", to=scriva.sinks.s3("s3://bucket/out/"), schema=Invoice)
scriva.watch("./incoming", to=scriva.sinks.webhook("https://api/extract"), schema=Invoice)
scriva.watch("./incoming", to=scriva.sinks.sqlite("results.db"), schema=Invoice)

scriva.sinks is the catalogue — fs, s3, webhook, sqlite, postgres, gsheets. Each takes a DocumentResult (or the serialisation form) and writes it.

Cost and timing controls

invoice = scriva.extract(
    "huge.pdf",
    schema=Invoice,
    timeout_s=120.0,                         # raises TimeoutError
    max_cost_usd=0.50,                       # raises BudgetExceeded mid-run
)

max_cost_usd is enforced by the cost accountant in Reliability › Cost caps; it samples after each Worker call and short-circuits before exceeding the cap.

Eventing

def on_event(ev):
    if ev.kind == "progress":
        update_ui(ev.payload["done"], ev.payload["total"])

invoice = scriva.extract("acme.pdf", schema=Invoice, on_event=on_event)

Same event shape as the lower-level Orchestrator.events(...) — see Architecture › Observability for the table.

When to drop down to Orchestrator

extract covers the 80% case. Reach for an explicit Orchestrator recipe when you need:

• Multi-stage flows like split → human-review → re-OCR.
• Custom steps beyond what a preset exposes.
• Streaming (async for page in recipe.stream(doc)).
• Pause/resume with sidecar layouts.

Every preset is itself a recipe — scriva.presets.invoice.recipe() returns the underlying Orchestrator so you can .replace() a step and keep the rest.

recipe = scriva.presets.invoice.recipe()
recipe = recipe.replace("recognize", Worker.anthropic("claude-opus-4-7"))
result = recipe("acme.pdf")

What to read next

• Schemas — the built-in pydantic models.
• Presets — pre-tuned recipes per document kind.
• CLI — scriva extract, scriva watch, scriva batch from the shell.
• Reliability — retries, timeouts, cost caps, redaction.