Batch Processing - Pulse API

Overview

The batch endpoints let you run any step of the Pulse pipeline across many documents at once. Each batch call is fully asynchronous — it returns immediately with an ID to poll, and orchestrates parallel workers behind the scenes. /batch/schema, /batch/tables, and /batch/split always return a batch_job_id, polled via GET /job/batch_job_id for real-time progress including per-item completion status and individual child job IDs. /batch/extract returns one of two different ID fields depending on which request shape you use — see below.

Batch endpoints mirror the individual pipeline steps. Each child call goes through the exact same code path as calling the individual endpoint directly — batch is orchestration, not a separate implementation.

POST /batch/extract accepts two request shapes at the same path, and Pulse picks the right one automatically based on your request body:

Source and destination — input/output/extract_options/workers. Returns batch_job_id; poll via GET /job/{'{'}id{'}'}.
URL list — a top-level urls array. Backed by a durable, distributed queue built for very large batches (up to ~1,000,000 documents). Returns batch_id; poll via GET /batch/{'{'}id{'}'} for O(1) status, with pagination, cancellation, and a downloadable result manifest.

/batch/schema, /batch/tables, and /batch/split accept a batch_job_id from either shape via batch_extract_id.

Pipeline

Batch endpoints can be chained together, just like their single-document counterparts: Each step takes the output of a previous step as input, either via a batch_extract_id / batch_split_id that references the parent batch job, or via an explicit list of individual IDs.

Workers

Workers process items in parallel. You can control concurrency with the workers parameter on every batch endpoint.

Parameter	Type	Default	Max	Description
`workers`	integer	4	10	Number of parallel workers

Batch Extract

Enumerate files from an input source and extract content from each one. POST /batch/extract accepts either request shape below at the same path. A top-level urls array selects URL-list processing; input and output select source/destination processing.

See Extract for details on extract_options (pages, figure processing, extensions, etc.).

Source and destination request

POST /batch/extract

Field	Type	Required	Description
`input`	object	Yes	Source of files to process (see Input Sources)
`output`	object	Yes	Where to save extraction results (see Output Destinations)
`extract_options`	object	No	Options forwarded to each `/extract` call
`workers`	integer	No	Parallel workers (default: 4, max: 10)

Multipart file uploads (-F "file=@...") are also accepted at this same path instead of a JSON body — see Input Sources.

Source and destination response (202)

Field	Type	Description
`batch_job_id`	string	Job ID for polling
`status`	string	`"processing"`
`total_files`	integer	Number of files that will be processed

The batch_job_id from a batch extract can be used in two ways:

Batch Schema — Pass it as batch_extract_id to POST /batch/schema to apply the same schema to each document independently (one result per document).
Multi-Extraction — Pass it as extraction_id to POST /schema to combine all documents and apply the schema to the composite (one merged result). See Multi-Extraction Mode.

URL-list request

POST /batch/extract Built for very large batches (up to ~1,000,000 documents) via a durable, distributed queue rather than in-process workers. URL-only — no S3 prefix, inline base64, or multipart upload input.

Field	Type	Required	Description
`urls`	array of strings	Yes	Non-empty list of document URLs to download and process
`priority`	string	No	Accepted and stored on the batch; not yet used for queue routing (default: `"normal"`)
(any other field)	—	No	Passed through as extract options to every child, same as `extract_options` above

There is no separate output field for this shape — results are managed by Pulse and retrieved via the result manifest (see URL-list batch status below), not written to a caller-specified destination.

URL-list response (202)

Field	Type	Description
`batch_id`	string	Batch ID — poll via `GET /batch/{'{'}batch_id{'}'}`, not `GET /job/{'{'}id{'}'}`
`status`	string	`"processing"`
`total_files`	integer	Number of URLs that will be processed

Example — S3 Source

from pulse import Pulse
from pulse.types.batch_input_source import BatchInputSource
from pulse.types.batch_output_destination import BatchOutputDestination

client = Pulse(api_key="YOUR_API_KEY")

resp = client.batch.extract(
    input=BatchInputSource(s_3_prefix="s3://my-bucket/documents/"),
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/extractions/"),
    workers=10,
)

print(f"Job: {resp.batch_job_id}, files: {resp.total_files}")

import { PulseClient } from "pulse-ts-sdk";

const client = new PulseClient({ apiKey: "YOUR_API_KEY" });

const resp = await client.batch.extract({
    input: { s3_prefix: "s3://my-bucket/documents/" },
    output: { s3_prefix: "s3://my-bucket/results/extractions/" },
    workers: 10,
});

console.log(`Job: ${resp.batch_job_id}, files: ${resp.total_files}`);

curl -X POST https://api.runpulse.com/batch/extract \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "input": { "s3_prefix": "s3://my-bucket/documents/" },
    "output": { "s3_prefix": "s3://my-bucket/results/extractions/" },
    "workers": 10
  }'

Example — Upload Files from Folder

import base64
import os
from pulse.types import BatchFileUpload

# Upload and extract all files from a local folder
folder = "documents/"
files = []
for name in os.listdir(folder):
    with open(os.path.join(folder, name), "rb") as f:
        files.append(BatchFileUpload(
            filename=name,
            content=base64.b64encode(f.read()).decode(),
        ))

resp = client.batch.extract(
    input=BatchInputSource(files=files),
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/"),
)

import * as fs from "fs";
import * as path from "path";

// Upload and extract all files from a local folder
const folder = "documents/";
const files = fs.readdirSync(folder).map(name => ({
    filename: name,
    content: fs.readFileSync(path.join(folder, name)).toString("base64"),
}));

const resp = await client.batch.extract({
    input: { files },
    output: { s3Prefix: "s3://my-bucket/results/" },
});

# Upload files directly as multipart form data
curl -X POST https://api.runpulse.com/batch/extract \
  -H "x-api-key: YOUR_API_KEY" \
  -F "file=@documents/report.pdf" \
  -F "file=@documents/data.xlsx" \
  -F 'output={"s3_prefix":"s3://my-bucket/results/"}'

Example — URL list (large batches)

A top-level urls array selects URL-list processing automatically. Shown here as curl; check the SDK changelog for native urls= support in your language.

curl -X POST https://api.runpulse.com/batch/extract \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "urls": [
      "https://example.com/report-1.pdf",
      "https://example.com/report-2.pdf"
    ],
    "priority": "normal"
  }'

# Response (202):
# { "batch_id": "b2232707-3975-4e12-b815-b907fe0f653d", "status": "processing", "total_files": 2 }

Batch Schema

Apply the same schema to each document independently, producing one result per extraction. Supports two modes, inferred from input:

Single mode — Provide extraction_ids or batch_extract_id with schema_config
Split mode — Provide split_ids or batch_split_id with split_schema_config

Batch Schema vs Multi-Extraction — these solve different problems:

Batch Schema (POST /batch/schema) applies the same schema to each document separately — you get N results for N documents.
Multi-Extraction (POST /schema with a batch extract ID or extraction_ids) combines all documents into one composite and applies the schema once — you get a single result with data merged from all sources.

If you need to pull data that spans across multiple files (e.g., loss data in one file + exposure data in another), use Multi-Extraction instead.

See Schema for details on schema_config, split_schema_config, and the difference between single and split modes.

Request — `POST /batch/schema`

Field	Type	Required	Description
`output`	object	Yes	Where to save schema results
`batch_extract_id`	string	XOR	ID of a prior batch extract run (single mode)
`extraction_ids`	array	XOR	Explicit list of extraction IDs (single mode)
`batch_split_id`	string	XOR	ID of a prior batch split run (split mode)
`split_ids`	array	XOR	Explicit list of split IDs (split mode)
`schema_config`	object	Conditional	Schema configuration for single mode
`split_schema_config`	object	Conditional	Per-topic schema configurations for split mode
`workers`	integer	No	Parallel workers (default: 4, max: 10)

Response (202)

Field	Type	Description
`batch_job_id`	string	Job ID for polling
`status`	string	`"processing"`
`total_extractions`	integer	Number of extractions to process (single mode)
`total_splits`	integer	Number of splits to process (split mode)

Example — Single Mode

from pulse.types.schema_config import SchemaConfig

resp = client.batch.schema(
    batch_extract_id="<batch_extract_job_id>",
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/schemas/"),
    schema_config=SchemaConfig(
        input_schema={
            "type": "object",
            "properties": {
                "total_amount": {"type": "number"},
                "vendor_name": {"type": "string"},
            },
        },
        schema_prompt="Extract invoice details",
    ),
    workers=10,
)

const resp = await client.batch.schema({
    batchExtractId: "<batch_extract_job_id>",
    output: { s3_prefix: "s3://my-bucket/results/schemas/" },
    schemaConfig: {
        inputSchema: {
            type: "object",
            properties: {
                total_amount: { type: "number" },
                vendor_name: { type: "string" },
            },
        },
        schemaPrompt: "Extract invoice details",
    },
    workers: 10,
});

curl -X POST https://api.runpulse.com/batch/schema \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "batch_extract_id": "<batch_extract_job_id>",
    "output": { "s3_prefix": "s3://my-bucket/results/schemas/" },
    "schema_config": {
      "input_schema": {
        "type": "object",
        "properties": {
          "total_amount": {"type": "number"},
          "vendor_name": {"type": "string"}
        }
      },
      "schema_prompt": "Extract invoice details"
    },
    "workers": 10
  }'

Example — Split Mode

from pulse.types.topic_schema_config import TopicSchemaConfig

resp = client.batch.schema(
    batch_split_id="<batch_split_job_id>",
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/schema-splits/"),
    split_schema_config={
        "financial_statements": TopicSchemaConfig(
            schema_={"type": "object", "properties": {"revenue": {"type": "number"}}},
            schema_prompt="Extract financial data",
        ),
        "risk_factors": TopicSchemaConfig(
            schema_={"type": "object", "properties": {"risks": {"type": "array", "items": {"type": "string"}}}},
        ),
    },
    workers=10,
)

curl -X POST https://api.runpulse.com/batch/schema \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "batch_split_id": "<batch_split_job_id>",
    "output": { "s3_prefix": "s3://my-bucket/results/schema-splits/" },
    "split_schema_config": {
      "financial_statements": {
        "schema": {"type": "object", "properties": {"revenue": {"type": "number"}}},
        "schema_prompt": "Extract financial data"
      },
      "risk_factors": {
        "schema": {"type": "object", "properties": {"risks": {"type": "array", "items": {"type": "string"}}}}
      }
    },
    "workers": 10
  }'

Batch Tables

Extract tables from multiple existing extractions.

See Tables for details on tables_config (merge, table format, etc.).

Request — `POST /batch/tables`

Field	Type	Required	Description
`output`	object	Yes	Where to save table results
`batch_extract_id`	string	XOR	ID of a prior batch extract run
`extraction_ids`	array	XOR	Explicit list of extraction IDs
`tables_config`	object	No	Table extraction configuration
`workers`	integer	No	Parallel workers (default: 4, max: 10)

Response (202)

Field	Type	Description
`batch_job_id`	string	Job ID for polling
`status`	string	`"processing"`
`total_extractions`	integer	Number of extractions to process

Example

resp = client.batch.tables(
    batch_extract_id="<batch_extract_job_id>",
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/tables/"),
    tables_config=TablesConfig(merge=True, table_format="html"),
    workers=10,
)

curl -X POST https://api.runpulse.com/batch/tables \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "batch_extract_id": "<batch_extract_job_id>",
    "output": { "s3_prefix": "s3://my-bucket/results/tables/" },
    "tables_config": { "merge": true, "table_format": "html" },
    "workers": 10
  }'

Batch Split

Split multiple extractions into topics.

See Split for details on split_config (topic definitions with names and descriptions).

Request — `POST /batch/split`

Field	Type	Required	Description
`output`	object	Yes	Where to save split results
`split_config`	object	Yes	Split configuration with topic definitions
`batch_extract_id`	string	XOR	ID of a prior batch extract run
`extraction_ids`	array	XOR	Explicit list of extraction IDs
`workers`	integer	No	Parallel workers (default: 4, max: 10)

Response (202)

Field	Type	Description
`batch_job_id`	string	Job ID for polling
`status`	string	`"processing"`
`total_extractions`	integer	Number of extractions to process

Example

from pulse.types.split_config import SplitConfig
from pulse.types.topic_definition import TopicDefinition

resp = client.batch.split(
    batch_extract_id="<batch_extract_job_id>",
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/splits/"),
    split_config=SplitConfig(
        split_input=[
            TopicDefinition(name="financial_statements", description="Balance sheets, income statements"),
            TopicDefinition(name="risk_factors", description="Risk factors and forward-looking statements"),
        ],
    ),
    workers=10,
)

curl -X POST https://api.runpulse.com/batch/split \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "batch_extract_id": "<batch_extract_job_id>",
    "output": { "s3_prefix": "s3://my-bucket/results/splits/" },
    "split_config": {
      "split_input": [
        {"name": "financial_statements", "description": "Balance sheets, income statements"},
        {"name": "risk_factors", "description": "Risk factors and forward-looking statements"}
      ]
    },
    "workers": 10
  }'

Input and Output

Input Sources

Batch Extract accepts one of the following input sources via JSON body:

Source	Field	Description
S3 prefix	`input.s3_prefix`	S3 URI prefix (e.g. `s3://my-bucket/documents/`). All supported files under this prefix are processed.
URL list	`input.file_urls`	Explicit list of file URLs to download and process.
Inline file uploads	`input.files`	Array of `{ filename, content }` objects where `content` is base64-encoded file bytes.

Alternatively, you can upload files directly via multipart form data instead of a JSON body:

curl -X POST https://api.runpulse.com/batch/extract \
  -H "x-api-key: YOUR_API_KEY" \
  -F "file=@report.pdf" \
  -F "file=@data.xlsx" \
  -F 'output={"s3_prefix":"s3://my-bucket/results/"}'

All other batch endpoints reference prior results via IDs rather than raw files.

Output Destinations

Every batch endpoint writes results to an output destination. You can specify one or both:

Destination	Field	Example
S3 prefix	`s3_prefix`	`s3://my-bucket/results/`
Local directory	`local_path`	`/data/results/`

Source/destination batch status

Use this flow for a source/destination request that returned batch_job_id. For a URL-list request, use URL-list batch status instead. GET /job/{'{'}id{'}'} does not accept a batch_id.

Poll GET /job/batch_job_id to monitor a batch job. The response includes a result object with structured progress:

{
  "status": "processing",
  "result": {
    "progress": {
      "total": 10,
      "completed": 7,
      "failed": 1
    },
    "jobs": {
      "completed": [
        { "job_id": "abc-123", "file": "report.pdf" }
      ],
      "failed": [
        { "job_id": "def-456", "file": "corrupt.pdf", "error": "..." }
      ],
      "processing": ["ghi-789"]
    }
  }
}

Each child job_id can be polled individually for detailed results.

Polling Example

import time

job_id = resp.batch_job_id

while True:
    job = client.jobs.get_job(job_id=job_id)

    if job.status in ("completed", "failed", "canceled"):
        print(f"Final status: {job.status}")
        break

    if job.result and "progress" in job.result:
        p = job.result["progress"]
        print(f"{p['completed']}/{p['total']} completed, {p['failed']} failed")

    time.sleep(5)

let job = await client.jobs.getJob(resp.batchJobId);

while (job.status !== "completed" && job.status !== "failed" && job.status !== "canceled") {
    await new Promise(r => setTimeout(r, 5000));
    job = await client.jobs.getJob(resp.batchJobId);

    if (job.result?.progress) {
        const p = job.result.progress;
        console.log(`${p.completed}/${p.total} completed, ${p.failed} failed`);
    }
}

console.log(`Final status: ${job.status}`);

URL-list batch status

For batches created with a URL-list request, status is O(1) regardless of batch size — no need to page through per-item results just to check overall progress.

`GET /batch/{'{'}batch_id{'}'}`

Field	Type	Description
`batch_id`	string	The batch ID
`status`	string	`"processing"`, `"completed"`, `"failed"`, or `"canceled"`
`priority`	string	As submitted (default `"normal"`)
`created_at`	string	ISO timestamp
`counts`	object	`total`, `pending`, `processing`, `succeeded`, `failed`, `canceled`
`jobs_url`	string	Path to the paginated per-document drill-down, see below
`result_manifest_url`	string \| null	Downloadable NDJSON manifest of every child’s outcome; built lazily on the first poll after the batch reaches a terminal status, so this stays `null` for a few polls after completion

curl

curl https://api.runpulse.com/batch/b2232707-3975-4e12-b815-b907fe0f653d \
  -H "x-api-key: YOUR_API_KEY"

# {
#   "batch_id": "b2232707-3975-4e12-b815-b907fe0f653d",
#   "status": "processing",
#   "priority": "normal",
#   "created_at": "2026-08-04T12:00:00Z",
#   "counts": { "total": 1000, "pending": 400, "processing": 80, "succeeded": 510, "failed": 10, "canceled": 0 },
#   "jobs_url": "/batch/b2232707-3975-4e12-b815-b907fe0f653d/jobs",
#   "result_manifest_url": null
# }

`GET /batch/{'{'}batch_id{'}'}/jobs?status=&cursor=`

Paginated per-document drill-down (100 per page). status optionally filters to one of pending, processing, completed, failed, canceled. Keep following next_cursor until it’s null.

Field	Type	Description
`jobs`	array	`{ job_id, status, result, error }` per document on this page
`next_cursor`	string \| null	Pass back as `?cursor=` for the next page; `null` when exhausted

curl

curl "https://api.runpulse.com/batch/b2232707-3975-4e12-b815-b907fe0f653d/jobs?status=failed" \
  -H "x-api-key: YOUR_API_KEY"

Cancellation

Use DELETE /job/{'{'}batch_job_id{'}'} for source/destination batches. Use POST /batch/{'{'}batch_id{'}'}/cancel for URL-list batches.

Cancel a batch job with DELETE /job/batch_job_id. This cascades to all child jobs that are still pending or processing.

`POST /batch/{'{'}batch_id{'}'}/cancel` (URL list)

Pending children are canceled outright; children already processing run to natural completion rather than being force-killed.

curl

curl -X POST https://api.runpulse.com/batch/b2232707-3975-4e12-b815-b907fe0f653d/cancel \
  -H "x-api-key: YOUR_API_KEY"

# { "batch_id": "b2232707-3975-4e12-b815-b907fe0f653d", "status": "canceled" }

Full Pipeline Example

Process a folder of SEC filings: extract all files, apply a schema, extract tables, split by topic, and apply per-topic schemas.

from pulse import Pulse
from pulse.types.batch_input_source import BatchInputSource
from pulse.types.batch_output_destination import BatchOutputDestination
from pulse.types.schema_config import SchemaConfig
from pulse.types.split_config import SplitConfig
from pulse.types.tables_config import TablesConfig
from pulse.types.topic_definition import TopicDefinition
from pulse.types.topic_schema_config import TopicSchemaConfig
import time

client = Pulse(api_key="YOUR_API_KEY")

def wait_for_job(job_id: str) -> dict:
    while True:
        job = client.jobs.get_job(job_id=job_id)
        if job.status in ("completed", "failed", "canceled"):
            return {"status": job.status, "result": job.result}
        time.sleep(5)

# Step 1: Batch Extract
extract = client.batch.extract(
    input=BatchInputSource(s_3_prefix="s3://my-bucket/10-K/"),
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/extractions/"),
    workers=10,
)
extract_result = wait_for_job(extract.batch_job_id)

# Step 2: Batch Schema
schema = client.batch.schema(
    batch_extract_id=extract.batch_job_id,
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/schemas/"),
    schema_config=SchemaConfig(
        input_schema={"type": "object", "properties": {"revenue": {"type": "number"}}},
    ),
    workers=10,
)
wait_for_job(schema.batch_job_id)

# Step 3: Batch Tables
tables = client.batch.tables(
    batch_extract_id=extract.batch_job_id,
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/tables/"),
    tables_config=TablesConfig(merge=True, table_format="html"),
    workers=10,
)
wait_for_job(tables.batch_job_id)

# Step 4: Batch Split
split = client.batch.split(
    batch_extract_id=extract.batch_job_id,
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/splits/"),
    split_config=SplitConfig(split_input=[
        TopicDefinition(name="financials", description="Financial statements"),
        TopicDefinition(name="risk_factors", description="Risk disclosures"),
    ]),
    workers=10,
)
wait_for_job(split.batch_job_id)

# Step 5: Batch Schema (split mode)
split_schema = client.batch.schema(
    batch_split_id=split.batch_job_id,
    output=BatchOutputDestination(s_3_prefix="s3://my-bucket/results/schema-splits/"),
    split_schema_config={
        "financials": TopicSchemaConfig(
            schema_={"type": "object", "properties": {"revenue": {"type": "number"}}},
        ),
        "risk_factors": TopicSchemaConfig(
            schema_={"type": "object", "properties": {"risks": {"type": "array", "items": {"type": "string"}}}},
        ),
    },
    workers=10,
)
wait_for_job(split_schema.batch_job_id)

Extract

Individual file extraction — config options apply to Batch Extract

Schema

Single/split schema extraction — config options apply to Batch Schema

Tables

Table extraction — config options apply to Batch Tables

Split

Topic splitting — config options apply to Batch Split

Poll Job

Poll batch job progress

Cancel Job

Cancel a batch job and all child jobs

​Overview

​Pipeline

​Workers

​Batch Extract

​Source and destination request

​Source and destination response (202)

​URL-list request

​URL-list response (202)

​Example — S3 Source

​Example — Upload Files from Folder

​Example — URL list (large batches)

​Batch Schema

​Request — POST /batch/schema

​Response (202)

​Example — Single Mode

​Example — Split Mode

​Batch Tables

​Request — POST /batch/tables

​Response (202)

​Example

​Batch Split

​Request — POST /batch/split

​Response (202)

​Example

​Input and Output

​Input Sources

​Output Destinations

​Source/destination batch status

​Polling Example

​URL-list batch status

​GET /batch/{'{'}batch_id{'}'}

​GET /batch/{'{'}batch_id{'}'}/jobs?status=&cursor=

​Cancellation

​POST /batch/{'{'}batch_id{'}'}/cancel (URL list)

​Full Pipeline Example

​Related Endpoints

Extract

Schema

Tables

Split

Poll Job

Cancel Job

Overview

Pipeline

Workers

Batch Extract

Source and destination request

Source and destination response (202)

URL-list request

URL-list response (202)

Example — S3 Source

Example — Upload Files from Folder

Example — URL list (large batches)

Batch Schema

Request — `POST /batch/schema`

Response (202)

Example — Single Mode

Example — Split Mode

Batch Tables

Request — `POST /batch/tables`

Response (202)

Example

Batch Split

Request — `POST /batch/split`

Response (202)

Example

Input and Output

Input Sources

Output Destinations

Source/destination batch status

Polling Example

URL-list batch status

`GET /batch/{'{'}batch_id{'}'}`

`GET /batch/{'{'}batch_id{'}'}/jobs?status=&cursor=`

Cancellation

`POST /batch/{'{'}batch_id{'}'}/cancel` (URL list)

Full Pipeline Example

Related Endpoints