Cloud intake

Parse every document in a storage bucket with one request, and write each result back to the bucket as its job ends. Use it for backfills and folder drops.

What you need

The Python SDK and an API key, set up as in Cookbooks.
boto3, with AWS credentials that can list and read the documents and write the results.
An Amazon S3 bucket of PDFs. Signed URLs from Google Cloud Storage or Azure Blob Storage work the same way.

Build it

Set up the clients

This code creates the Pulse and S3 clients, and names the bucket and its folders.

import os
import time
from urllib.parse import parse_qs, urlparse

import boto3
from pulse import Pulse
from pulse.core.api_error import ApiError

client = Pulse(api_key=os.environ["PULSE_API_KEY"])
s3 = boto3.client("s3")
BUCKET = "acme-intake"  # your bucket
PREFIX = "incoming/"  # the folder to parse
RESULTS = "pulse-results/"  # where the results go
CONFIG = {"model": "ultra-2"}

Sign a URL for each document

This code signs a download URL, valid for 48 hours, for each PDF in the folder.

listing = s3.get_paginator("list_objects_v2").paginate(Bucket=BUCKET, Prefix=PREFIX)
keys = [item["Key"] for page in listing for item in page.get("Contents", []) if item["Key"].lower().endswith(".pdf")]
urls = [
    s3.generate_presigned_url("get_object", Params={"Bucket": BUCKET, "Key": key}, ExpiresIn=48 * 3600)
    for key in keys
]

Pulse fetches each URL when its job runs, up to 48 hours after you send the batch, so sign with credentials that last that long. A URL signed with temporary credentials, such as an assumed role’s, stops working when they expire.

Send the batch

This code sends the URLs in batches of up to 1,000 and keeps each batch_id.

batch_ids = []
for start in range(0, len(urls), 1000):
    batch = client.parse.batch.run(file_urls=urls[start : start + 1000], config=CONFIG)
    batch_ids.append(batch.batch_id)  # store these too: you need them to resume

A batch always runs async, and each URL becomes its own parse job. A URL that doesn’t return the document when its job runs, such as one whose signature has expired, fails only its own job.

Write each result as its job ends

drain reads a batch’s manifest, the list of its jobs that have ended, and passes each new job to handle, which writes its result to the bucket.

ENDED = ("completed", "failed", "canceled")


def next_cursor(headers):
    """The cursor in a manifest page's Link header; None on the last page."""
    link = next((value for name, value in headers.items() if name.lower() == "link"), None)
    if link is None:
        return None
    return parse_qs(urlparse(link.split(";")[0].strip(" <>")).query)["cursor"][0]


def handle(job):
    """Write one ended job's result to the bucket, or report why there is none."""
    if job.status != "completed":
        print(f"{job.parse_id} {job.status}: {job.error}")
        return
    try:
        result = client.parse.get_result(job.parse_id)
    except ApiError as error:
        if error.status_code != 410:
            raise
        print(f"{job.parse_id}: the result was already deleted")
        return
    s3.put_object(Bucket=BUCKET, Key=f"{RESULTS}{job.parse_id}.json", Body=result.model_dump_json().encode())


def drain(batch_id, handled):
    """Handle each of the batch's jobs once, as it ends, until the batch ends."""
    cursor = None
    while True:
        ended = client.parse.batch.get(batch_id).status in ENDED
        page = client.parse.batch.with_raw_response.get_result(batch_id, limit=500, cursor=cursor)
        for job in page.data:
            if job.parse_id not in handled:
                handle(job)
                handled.add(job.parse_id)
        if found := next_cursor(page.headers):
            cursor = found
        elif ended:
            return
        else:
            time.sleep(30)

Run until every batch ends

This loop drains each batch in turn. A brief 429 or 5xx error stops drain; the loop waits and starts it again.

RETRY = {429, 500, 502, 503, 504}
handled = set()  # the parse_ids already written or reported: keep these in your database

for batch_id in batch_ids:
    while True:
        try:
            drain(batch_id, handled)
            break
        except ApiError as error:
            if error.status_code not in RETRY:
                raise
            time.sleep(30)

handled makes sure no result is read twice and none is skipped when drain starts again, even after a restart if you store it with the batch ids.

How it works

The id each step hands to the next: On each pass, drain reads the batch’s status and the manifest: The manifest’s entries never move, so drain follows each page’s Link header to the next page, then asks for the last page again until the batch ends. Each result is read once, so the recipe also works where your organization deletes a result the first time it’s read.

Edge cases

Mapping results to objects

A manifest entry names its parse_id, not its URL, and the result’s history.file_name holds only the file’s name, which can repeat across folders. To file results by object key, or to retry one object, send one async /parse per object and store each parse_id with its key.

Keeping results only in your bucket

Add "storage": {"save": False} to CONFIG. Each result is then deleted 24 hours after its batch ends, or sooner if your organization sets a 1-hour retention. The free plan can’t turn saving off.

A batch that ends failed or canceled

A batch ends failed when it isn’t done within 48 hours of being sent, or when it’s interrupted. After either, or after you cancel a batch, some of its jobs can still end: run drain once more a few minutes later, with the same handled, to collect them. An interrupted batch can also leave documents that never ran; to find those, track each object with its own parse_id, as above.

A public folder

If anyone can list and read the folder, send it as folder_url instead of signing a URL for each document.

Cookbooks

What you need

Build it

How it works

Edge cases

Next steps

Batch

Webhooks

​What you need

​Build it

​How it works

​Edge cases

​Next steps

Batch

Webhooks

What you need

Build it

How it works

Edge cases

Next steps