Cookbook: Batch processing

Cookbook: batch & watch ingest

The two shapes for “process many files”:

Batch — you have a list now, run it to completion.
Watch — files keep arriving, process each as it lands.

Both are convenience layers around an Orchestrator recipe. You build the recipe once and hand it to scriva.batch(...) or scriva.watch(...).

Batch

import scriva
from pathlib import Path
from scriva import Orchestrator, Worker

recipe = (
    Orchestrator()
    .deskew()
    .split.boxes(detector="invoice")
    .recognize(Worker.openai("gpt-4o").cache(".scriva_cache"))
    .reconstruct.kv()
    .export.json("out/{stem}.json")
)

results = scriva.batch(
    sources=Path("incoming/").glob("*.pdf"),
    recipe=recipe,
    max_concurrency=8,
    progress=True,
)
for src, result in results:
    save_to_erp(src, result)

scriva.batch returns an iterable of (source, result_or_exception). By default it yields as each source finishes (out-of-order); pass ordered=True to preserve input order.

Error handling modes

scriva.batch(sources, recipe=recipe, on_error="raise")        # default outside try
scriva.batch(sources, recipe=recipe, on_error="continue")     # yield (src, Exception)
scriva.batch(sources, recipe=recipe, on_error="collect")      # finish all, then raise BatchErrors

"continue" is the right shape for overnight runs: every source either succeeds or yields an exception you can ledger and move on.

Resume on restart

scriva.batch(
    sources=Path("incoming/").glob("*.pdf"),
    recipe=recipe,
    resume_file=".scriva_batch.json",        # JSONL ledger
)

The ledger keys each source by hash(source) + recipe_id. Re-running with the same resume_file skips already-completed sources. force=True ignores the ledger and reprocesses everything.

To clear: delete the file.

Budget

scriva.batch(sources, recipe=recipe, max_cost_usd=20.0)

Hard cap across the whole batch. When the cap is hit, in-flight sources finish; new ones short-circuit. The result iterator continues to yield; failed entries carry a BudgetExceeded exception.

Per-source caps are independent:

scriva.batch(
    sources, recipe=recipe,
    max_cost_usd=20.0,
    per_source_max_cost_usd=0.50,
)

Cancellation

import asyncio

async def main():
    batch = scriva.batch(sources, recipe=recipe, return_="iterator")
    task = asyncio.create_task(consume(batch))
    await asyncio.sleep(60)
    batch.cancel()
    await task                           # consume returns whatever finished

Per-source side effects

scriva.batch(
    sources,
    recipe=recipe,
    on_each=lambda src, result: requests.post(
        "https://api/extract",
        json=result.to_dict(),
    ),
)

on_each runs once per success in the batch’s executor. Exceptions in on_each are wrapped as SinkError and routed through on_error.

Watch

scriva.watch(
    folder="./incoming",
    to="./extracted",
    recipe=recipe,
    pattern="*.pdf",
)

watch blocks until interrupted. It uses watchdog for inotify events (falls back to polling on platforms without it), processes each new file exactly once, and writes results to to.

Output sinks

The to argument accepts:

scriva.watch("./in", to="./out", recipe=recipe)                                       # filesystem
scriva.watch("./in", to=scriva.sinks.s3("s3://bucket/out/"), recipe=recipe)           # S3
scriva.watch("./in", to=scriva.sinks.webhook("https://api/extract"), recipe=recipe)   # HTTP POST
scriva.watch("./in", to=scriva.sinks.sqlite("results.db"), recipe=recipe)             # SQLite
scriva.watch("./in", to=scriva.sinks.postgres("postgres://...", table="invoices"), recipe=recipe)
scriva.watch("./in", to=scriva.sinks.gsheets("<sheet-id>", tab="invoices"), recipe=recipe)

scriva.sinks is the catalogue. Each sink accepts a DocumentResult and serialises it.

For sinks that need a typed record, point at the recipe’s reconstruct output:

scriva.watch(
    "./in",
    to=scriva.sinks.postgres("postgres://...", table="invoices", from_field="fields"),
    recipe=recipe,
)

Source disposition

scriva.watch(
    "./in",
    to="./out",
    recipe=recipe,
    move_on_success="./in/.done",         # consume input
    move_on_error="./in/.failed",         # quarantine
)

By default watch leaves the source in place and uses a sidecar .scriva_watch.ledger.jsonl to track what has been processed. move_on_success is preferable for “ingest and forget” pipelines.

Graceful drain

SIGTERM (or KeyboardInterrupt) starts a drain: no new files are accepted, in-flight runs finish, the ledger is flushed, the process exits 0.

import signal, asyncio

async def main():
    watcher = scriva.watch("./in", to="./out", recipe=recipe, blocking=False)
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGTERM, watcher.drain)
    await watcher.run()

blocking=False returns a Watcher you can drive yourself.

Concurrency

scriva.watch("./in", to="./out", recipe=recipe, workers=4)

workers is the per-folder concurrency cap. Combine with recipe.options(page_concurrency=N) for nested control on multi-page sources.

Multi-recipe ingest

When the folder mixes document kinds, route by classifying first:

invoice_recipe   = scriva.presets.invoice.recipe()
receipt_recipe   = scriva.presets.receipt.recipe()
statement_recipe = scriva.presets.bank_statement.recipe()

def route(source: Path):
    kind = scriva.classify_document(source).kind
    return {
        "invoice":        invoice_recipe,
        "receipt":        receipt_recipe,
        "bank_statement": statement_recipe,
    }[kind]

scriva.watch("./in", to="./out", recipe=route, pattern="*.pdf")

recipe= accepts a callable that takes a source and returns the recipe to run. Internally watch runs classify_document once per source and dispatches.

For batch:

scriva.batch(sources, recipe=route)

Same shape.

Putting it together — overnight ingest

scriva.watch(
    folder="/var/data/invoices/in",
    to=scriva.sinks.postgres("postgres://app@db/billing", table="invoices_extracted"),
    recipe=invoice_recipe,
    pattern="*.pdf",
    workers=4,
    move_on_success="/var/data/invoices/done",
    move_on_error="/var/data/invoices/failed",
    max_cost_usd=0.50,                    # per file
    resume_file="/var/data/invoices/.scriva.ledger.jsonl",
    on_error="continue",
)

Wire it into a systemd service that handles restarts; the ledger + move_on_* make the whole thing idempotent.