Batch Sending

Send multiple invoices efficiently using GoRoute's bulk operations API at /api/v1/batches.

Batching is a two-step operation: you create a batch, then process it. Creating a batch stores the invoices and returns immediately; processing sends them. Because processing runs in the background, you poll the batch or subscribe to webhooks to learn when it finishes.

Create a Batch

import requests

BASE_URL = "https://app.goroute.ai/peppol-api"
HEADERS = {
    "X-API-Key": "your_api_key",
    "Content-Type": "application/json"
}

def create_batch(invoices: list[dict], name: str = None) -> dict:
    """Create a batch of invoices for bulk processing."""

    response = requests.post(
        f"{BASE_URL}/api/v1/batches",
        headers=HEADERS,
        json={
            "name": name,
            "invoices": invoices,
            "batch_type": "send"
        }
    )
    response.raise_for_status()
    return response.json()


# Example: create a batch of 3 invoices
batch = create_batch(
    [
        {"number": "INV-001", "...": "invoice fields"},
        {"number": "INV-002", "...": "invoice fields"},
        {"number": "INV-003", "...": "invoice fields"},
    ],
    name="January invoices"
)

print(f"Batch {batch['id']} created with {batch['progress']['total']} invoices")

Each entry in invoices is an invoice object using the same fields as a single-invoice send. The number field is used as the batch item's invoice number; if it is omitted, items are numbered Item-1, Item-2, and so on.

Batch Types

`batch_type`	Behaviour
`send`	Validate and send all invoices
`validate`	Validate only — a dry run, nothing is sent
`schedule`	Validate now, send at `scheduled_at`

For schedule, pass scheduled_at as an ISO 8601 timestamp.

Batch Response

POST /api/v1/batches returns 201 with the batch:

{
    "id": "3f2a1b4c-5d6e-4f70-8a91-2b3c4d5e6f70",
    "org_id": "9a8b7c6d-5e4f-4a3b-2c1d-0e9f8a7b6c5d",
    "name": "January invoices",
    "description": null,
    "batch_type": "send",
    "status": "pending",
    "progress": {
        "total": 3,
        "pending": 3,
        "processing": 0,
        "success": 0,
        "failed": 0,
        "skipped": 0,
        "percent": 0.0
    },
    "scheduled_at": null,
    "created_at": "2026-08-03T09:15:00",
    "started_at": null,
    "completed_at": null,
    "error_message": null
}

Create a Batch from CSV

Upload a CSV file instead of a JSON array:

def create_batch_from_csv(path: str, name: str = None) -> dict:
    """Create a batch by uploading a CSV file."""

    with open(path, "rb") as handle:
        response = requests.post(
            f"{BASE_URL}/api/v1/batches/csv",
            headers={"X-API-Key": "your_api_key"},
            files={"file": handle},
            params={"name": name, "batch_type": "send"}
        )

    response.raise_for_status()
    return response.json()

Validate a Batch

Validate every invoice in a batch without sending anything:

response = requests.post(
    f"{BASE_URL}/api/v1/batches/{batch_id}/validate",
    headers={"X-API-Key": "your_api_key"}
)

batch = response.json()
print(f"Status: {batch['status']}")

Validating moves the batch to validating and then to ready once it has finished.

Process a Batch

Send the invoices:

response = requests.post(
    f"{BASE_URL}/api/v1/batches/{batch_id}/process",
    headers={"X-API-Key": "your_api_key"},
    params={"skip_invalid": True}
)

batch = response.json()

skip_invalid defaults to true, which sends the valid invoices and skips the invalid ones. Set it to false to require that every invoice is valid.

Track Batch Progress

def get_batch(batch_id: str) -> dict:
    """Get the current state of a batch."""

    response = requests.get(
        f"{BASE_URL}/api/v1/batches/{batch_id}",
        headers={"X-API-Key": "your_api_key"}
    )
    response.raise_for_status()
    return response.json()


batch = get_batch(batch_id)
progress = batch["progress"]

print(f"Batch {batch['id']} — {batch['status']}")
print(f"  Succeeded: {progress['success']}")
print(f"  Failed:    {progress['failed']}")
print(f"  Skipped:   {progress['skipped']}")
print(f"  Pending:   {progress['pending']}")
print(f"  Complete:  {progress['percent']}%")

Batch Statuses

Status	Meaning
`pending`	Created, waiting to start
`validating`	Validating invoices
`ready`	Validated, ready to send
`processing`	Sending invoices
`completed`	All invoices processed
`partial`	Some invoices failed
`failed`	Batch failed entirely
`cancelled`	Cancelled by a user

Polling for Completion

import time

TERMINAL = {"completed", "partial", "failed", "cancelled"}

def wait_for_batch(batch_id: str, timeout: int = 300) -> dict:
    """Poll a batch until it reaches a terminal status."""
    start = time.time()

    while time.time() - start < timeout:
        batch = get_batch(batch_id)

        if batch["status"] in TERMINAL:
            return batch

        time.sleep(5)

    raise TimeoutError("Batch did not complete in time")

Inspect Individual Items

To see per-invoice outcomes, list the batch items:

def list_batch_items(batch_id: str, page: int = 1, page_size: int = 50) -> dict:
    response = requests.get(
        f"{BASE_URL}/api/v1/batches/{batch_id}/items",
        headers={"X-API-Key": "your_api_key"},
        params={"page": page, "page_size": page_size}
    )
    response.raise_for_status()
    return response.json()


for item in list_batch_items(batch_id)["items"]:
    if item["status"] == "success":
        print(f"✅ {item['invoice_number']}: {item['transaction_id']}")
    elif item["error"]:
        print(f"❌ {item['invoice_number']}: {item['error']['message']}")

Each item carries sequence, status, invoice_number, validation, transaction_id, error and timestamps. Item statuses are pending, validating, valid, invalid, processing, success, failed and skipped.

Retry Failed Items

Reset the failed items in a batch so they can be processed again:

response = requests.post(
    f"{BASE_URL}/api/v1/batches/{batch_id}/retry",
    headers={"X-API-Key": "your_api_key"}
)

print(response.json()["message"])  # e.g. "Reset 2 failed items for retry"

Cancel a Batch

Cancel a batch that is still pending or ready:

requests.post(
    f"{BASE_URL}/api/v1/batches/{batch_id}/cancel",
    headers={"X-API-Key": "your_api_key"}
)

List Batches

response = requests.get(
    f"{BASE_URL}/api/v1/batches",
    headers={"X-API-Key": "your_api_key"},
    params={"page": 1, "page_size": 20, "status": "completed"}
)

listing = response.json()
print(f"{listing['total']} batches across {listing['total_pages']} pages")

Batch Limits

Limit	Value
Invoices per batch	1 minimum, 1000 maximum
Batch items per page	100 maximum

Requests are subject to your plan's standard API rate limits.

Chunking Larger Volumes

For more than 1000 invoices, split the work across several batches:

def create_batches_in_chunks(invoices: list[dict], chunk_size: int = 1000) -> list[dict]:
    """Split a large invoice list into batches of at most 1000."""
    batches = []

    for i in range(0, len(invoices), chunk_size):
        chunk = invoices[i:i + chunk_size]
        batches.append(create_batch(chunk, name=f"Chunk {i // chunk_size + 1}"))

    return batches

Webhooks for Batch Events

Rather than polling, subscribe to batch webhook events:

BATCH_EVENTS = [
    "batch.created",
    "batch.validated",
    "batch.started",
    "batch.completed",
    "batch.partial",
    "batch.failed",
    "batch.cancelled"
]


@app.post("/webhooks/goroute")
async def handle_webhook(request: Request):
    event = await request.json()

    if event["type"] == "batch.completed":
        data = event["data"]

        print(f"Batch {data['batch_id']} ({data['name']}) finished")
        print(f"  {data['success_count']}/{data['total_items']} succeeded")
        print(f"  Success rate: {data['success_rate']}%")

        if data["failed_count"] > 0:
            handle_failed_items(data["batch_id"])

    return {"status": "ok"}

The batch.completed payload carries batch_id, name, total_items, success_count, failed_count and success_rate. Call GET /api/v1/batches/{batch_id}/items for per-invoice detail.

Batch events are a premium-tier feature.

Complete Example

import requests
import time
from typing import Optional

BASE_URL = "https://app.goroute.ai/peppol-api"
TERMINAL = {"completed", "partial", "failed", "cancelled"}


class BatchSender:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {
            "X-API-Key": api_key,
            "Content-Type": "application/json"
        }

    def create(self, invoices: list[dict], name: Optional[str] = None) -> dict:
        response = requests.post(
            f"{BASE_URL}/api/v1/batches",
            headers=self.headers,
            json={"name": name, "invoices": invoices, "batch_type": "send"}
        )
        response.raise_for_status()
        return response.json()

    def process(self, batch_id: str, skip_invalid: bool = True) -> dict:
        response = requests.post(
            f"{BASE_URL}/api/v1/batches/{batch_id}/process",
            headers={"X-API-Key": self.api_key},
            params={"skip_invalid": skip_invalid}
        )
        response.raise_for_status()
        return response.json()

    def get(self, batch_id: str) -> dict:
        response = requests.get(
            f"{BASE_URL}/api/v1/batches/{batch_id}",
            headers={"X-API-Key": self.api_key}
        )
        response.raise_for_status()
        return response.json()

    def items(self, batch_id: str) -> list[dict]:
        response = requests.get(
            f"{BASE_URL}/api/v1/batches/{batch_id}/items",
            headers={"X-API-Key": self.api_key},
            params={"page": 1, "page_size": 100}
        )
        response.raise_for_status()
        return response.json()["items"]

    def wait(self, batch_id: str, timeout: int = 300) -> dict:
        start = time.time()

        while time.time() - start < timeout:
            batch = self.get(batch_id)

            if batch["status"] in TERMINAL:
                return batch

            print(f"  Waiting... {batch['progress']['percent']}% complete")
            time.sleep(5)

        raise TimeoutError("Batch did not complete in time")

    def send_all(self, invoices: list[dict], chunk_size: int = 1000) -> list[dict]:
        """Create, process and await a batch for every chunk of invoices."""
        results = []

        for i in range(0, len(invoices), chunk_size):
            chunk = invoices[i:i + chunk_size]
            print(f"Batch {i // chunk_size + 1}: {len(chunk)} invoices")

            batch = self.create(chunk, name=f"Chunk {i // chunk_size + 1}")
            self.process(batch["id"])
            results.append(self.wait(batch["id"]))

        return results


# Usage
sender = BatchSender("your_api_key")
batches = sender.send_all(load_invoices())

for batch in batches:
    progress = batch["progress"]
    print(f"Batch {batch['id']}: {batch['status']}")
    print(f"  {progress['success']} succeeded, {progress['failed']} failed")

    for item in sender.items(batch["id"]):
        if item["error"]:
            print(f"  ❌ {item['invoice_number']}: {item['error']['message']}")

Create a Batch​

Batch Types​

Batch Response​

Create a Batch from CSV​

Validate a Batch​

Process a Batch​

Track Batch Progress​

Batch Statuses​

Polling for Completion​

Inspect Individual Items​

Retry Failed Items​

Cancel a Batch​

List Batches​

Batch Limits​

Chunking Larger Volumes​

Webhooks for Batch Events​

Complete Example​

Next Steps​