> ## Documentation Index
> Fetch the complete documentation index at: https://docs.runpulse.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Cloud intake

> Parse a bucket of documents as one batch and write each result back to your storage.

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](https://docs.runpulse.com/v2/cookbooks/overview.md#before-you-start).
* `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

**1. Set up the clients**

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

```python theme={null}
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"}
```

**2. Sign a URL for each document**

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

```python theme={null}
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.

**3. Send the batch**

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

```python theme={null}
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.

**4. 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.

```python theme={null}
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)
```

**5. Run until every batch ends**

This loop drains each batch in turn. A brief `429` or `5xx` [error](https://docs.runpulse.com/v2/errors.md) stops `drain`; the loop waits and starts it again.

```python theme={null}
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:

```mermaid theme={null}
%%{init: {"themeVariables": {"fontSize": "14px"}, "flowchart": {"nodeSpacing": 24, "rankSpacing": 32, "padding": 8}}}%%
flowchart TD
    A[Documents in your bucket] -->|signed URLs| B["POST /parse/batch"]
    B -->|batch_id| C["Manifest: parse_id, status"]
    C -->|each new parse_id| D[Results in your bucket]
```

On each pass, `drain` reads the batch's status and the manifest:

```mermaid theme={null}
%%{init: {"themeVariables": {"fontSize": "14px"}, "sequence": {"mirrorActors": false, "messageMargin": 28, "height": 44, "boxMargin": 6}}}%%
sequenceDiagram
    participant App
    participant Pulse
    loop every 30 s until the batch ends
        App->>Pulse: GET the next manifest page
        Pulse-->>App: ended jobs + Link
        App->>Pulse: GET each new result
    end
```

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

> [accordion] 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.

> [accordion] 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](https://docs.runpulse.com/v2/jobs.md#retention). The free plan can't turn saving off.

> [accordion] 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.

> [accordion] A public folder
>
> If anyone can list and read the folder, send it as [`folder_url`](https://docs.runpulse.com/v2/batch) instead of signing a URL for each document.

> [accordion] More steps per document
>
> To extract from each document as well, start the extract from a `parse.completed` [webhook](https://docs.runpulse.com/v2/webhooks.md). Then read each parse in `handle` only after its extract has ended, in case your organization deletes a result the first time it's read.

## Next steps

> [card] [Batch](https://docs.runpulse.com/v2/batch.md)
>
> Sources, statuses, limits and the manifest.

> [card] [Webhooks](https://docs.runpulse.com/v2/webhooks.md)
>
> Get a signed event as a job's status changes.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.