Smole
Blog

Batch Document Processing: Process Hundreds of Files via API

Process large volumes of documents at scale using a REST API. Batch extract data from invoices, contracts, forms, and reports with parallel processing and error handling.

Smole Team
  • batch
  • api
  • automation
  • scale

Single-document extraction is useful. But the real value unlocks when you can process hundreds or thousands of documents through the same pipeline: a quarter's worth of invoices, a folder of contracts, an archive of forms.

This guide covers how to set up batch document processing with Smole's API, including parallel execution, error handling, and result collection.

Architecture of a Batch Pipeline#

Text
Document folder
    │
    ├── file_001.pdf
    ├── file_002.pdf
    ├── ...
    └── file_500.pdf
         │
         ▼
    Upload (parallel)
         │
         ▼
    Pipeline processing
         │
         ▼
    Result collection
         │
         ▼
    Output (JSON, CSV, database)

Each document is processed independently. A failure in one doesn't affect the others.

Python Batch Processing#

Basic Sequential Processing#

Start simple: process one file at a time:

Python
import requests
import time
import json
from pathlib import Path
 
API_BASE = "https://api.smole.tech/api"
headers = {"Authorization": f"Bearer {API_KEY}"}
 
def extract(file_path, schema_id, timeout=120):
    with open(file_path, "rb") as f:
        resp = requests.post(
            f"{API_BASE}/pipeline/file",
            headers=headers,
            files={"file": f},
            data={"schemaId": schema_id}
        )
    resp.raise_for_status()
    pipeline_id = resp.json()["id"]
 
    start = time.time()
    while time.time() - start < timeout:
        resp = requests.get(f"{API_BASE}/pipeline/{pipeline_id}", headers=headers)
        data = resp.json()
        if data["status"] == "completed":
            return data["extraction"]["data"]
        if data["status"] == "failed":
            raise Exception(f"Failed: {data.get('error')}")
        time.sleep(2)
 
    raise TimeoutError("Extraction timed out")
 
# Process all PDFs in a folder
results = []
for pdf in sorted(Path("./documents").glob("*.pdf")):
    try:
        data = extract(str(pdf), schema_id)
        results.append({"file": pdf.name, "status": "ok", "data": data})
        print(f"  OK: {pdf.name}")
    except Exception as e:
        results.append({"file": pdf.name, "status": "error", "error": str(e)})
        print(f"  FAIL: {pdf.name}: {e}")
 
with open("results.json", "w") as f:
    json.dump(results, f, indent=2)

Parallel Processing with ThreadPoolExecutor#

Process multiple documents simultaneously:

Python
from concurrent.futures import ThreadPoolExecutor, as_completed
 
def process_batch(folder, schema_id, max_workers=5):
    files = sorted(Path(folder).glob("*.pdf"))
    results = []
 
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = {
            pool.submit(extract, str(f), schema_id): f.name
            for f in files
        }
 
        for future in as_completed(futures):
            filename = futures[future]
            try:
                data = future.result()
                results.append({"file": filename, "status": "ok", "data": data})
                print(f"  OK: {filename}")
            except Exception as e:
                results.append({"file": filename, "status": "error", "error": str(e)})
                print(f"  FAIL: {filename}: {e}")
 
    return results

Recommended concurrency: Start with 5 workers and increase based on your rate limits.

With Retries#

Add resilience for transient failures:

Python
def extract_with_retry(file_path, schema_id, max_retries=3):
    for attempt in range(1, max_retries + 1):
        try:
            return extract(file_path, schema_id)
        except Exception as e:
            if attempt == max_retries:
                raise
            wait = 2 ** attempt
            print(f"    Retry {attempt}/{max_retries} in {wait}s...")
            time.sleep(wait)

JavaScript Batch Processing#

JavaScript
import fs from "node:fs";
import path from "node:path";
 
async function processBatch(directory, schemaId, concurrency = 5) {
  const files = fs.readdirSync(directory).filter(f => f.endsWith(".pdf"));
  const results = [];
 
  for (let i = 0; i < files.length; i += concurrency) {
    const chunk = files.slice(i, i + concurrency);
    const settled = await Promise.allSettled(
      chunk.map(async (file) => {
        const data = await extract(path.join(directory, file), schemaId);
        return { file, data };
      })
    );
 
    for (const result of settled) {
      if (result.status === "fulfilled") {
        results.push({ ...result.value, status: "ok" });
      } else {
        results.push({ file: "unknown", status: "error", error: result.reason.message });
      }
    }
  }
 
  return results;
}

Output Formats#

Save as JSON#

Python
with open("results.json", "w") as f:
    json.dump(results, f, indent=2, ensure_ascii=False)

Save as CSV#

Flatten the data for spreadsheet use:

Python
import csv
 
def results_to_csv(results, output_path, row_key="line_items"):
    rows = []
    for r in results:
        if r["status"] != "ok":
            continue
        data = r["data"]
        for item in data.get(row_key, []):
            row = {"source_file": r["file"]}
            row.update(item)
            rows.append(row)
 
    if rows:
        with open(output_path, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=rows[0].keys())
            writer.writeheader()
            writer.writerows(rows)

Write to Database#

Python
def save_batch_to_db(results, conn):
    cur = conn.cursor()
    for r in results:
        if r["status"] != "ok":
            continue
        data = r["data"]
        cur.execute(
            "INSERT INTO documents (filename, vendor, date, total) VALUES (%s, %s, %s, %s)",
            (r["file"], data.get("vendor_name"), data.get("date"), data.get("total"))
        )
    conn.commit()

Monitoring and Reporting#

Track progress and generate a summary:

Python
def print_summary(results):
    ok = sum(1 for r in results if r["status"] == "ok")
    failed = sum(1 for r in results if r["status"] == "error")
    total = len(results)
 
    print(f"\n{'='*40}")
    print(f"Batch complete: {total} documents")
    print(f"  Succeeded: {ok}")
    print(f"  Failed:    {failed}")
    print(f"  Success rate: {ok/total*100:.1f}%")
 
    if failed > 0:
        print(f"\nFailed files:")
        for r in results:
            if r["status"] == "error":
                print(f"  - {r['file']}: {r['error']}")

Handling Mixed Document Types#

If your batch contains different document types, use separate schemas:

Python
schemas = {
    "invoice": "schema_id_invoices",
    "receipt": "schema_id_receipts",
    "contract": "schema_id_contracts",
}
 
def classify_and_extract(file_path, schemas):
    """Route documents to the right schema based on filename or folder."""
    name = Path(file_path).stem.lower()
 
    if "invoice" in name or "inv" in name:
        schema_id = schemas["invoice"]
    elif "receipt" in name:
        schema_id = schemas["receipt"]
    elif "contract" in name or "agreement" in name:
        schema_id = schemas["contract"]
    else:
        schema_id = schemas["invoice"]  # default
 
    return extract(file_path, schema_id)

Performance Tips#

  1. Start with 5 concurrent workers and adjust based on response times and rate limits
  2. Process large batches in off-peak hours if possible
  3. Use retries with exponential backoff: transient failures are normal at scale
  4. Save intermediate results: Write results to disk periodically so a crash doesn't lose everything
  5. Log failures for reprocessing: Keep a list of failed files to retry separately

Try It Now#

Start with a single document in the Playground to validate your schema, then scale up with the batch processing code above.

For API details and rate limits, see the documentation.

Related articles

View all posts