> For the complete documentation index, see [llms.txt](https://docs.euno.ai/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.euno.ai/sources/custom-integrations/custom-integration/handling-high-volume-of-observations.md).

# Handling High Volume of Observations

Use this flow when you need to ingest **many observations in a single integration run** and inline JSON POST requests are too large or impractical.

For small payloads (a single observation or a modest JSON array), continue using the inline [`/run` endpoint](/sources/custom-integrations/custom-integration.md#sending-custom-observations) documented on the main Custom Integration page.

## When to Use File Upload

| Approach                       | Best for                                                                 |
| ------------------------------ | ------------------------------------------------------------------------ |
| **Inline JSON POST to `/run`** | Quick tests, single observations, small batches                          |
| **Prepare-upload file flow**   | Large exports, ETL pipelines, files that exceed HTTP request size limits |

The prepare-upload flow uploads your file directly to cloud storage, then automatically triggers **one integration run** that processes every observation in the file.

## How It Works

The integration follows these steps:

1. **Request a signed upload URL**\
   Call `prepare-upload` with your integration key. Euno creates a pending operation and returns a signed URL for your observations file.
2. **Upload the observations file**\
   PUT your `.json`, `.jsonl`, or `.ndjson` file to the signed URL.
3. **Automatic processing**\
   When the upload completes, Euno starts the integration run, validates each observation, and ingests them into your data model. Track progress on the **Sources** page under the integration's run history.

## Supported File Formats

Upload **exactly one** observations file per run.

### JSON (`.json`)

A single observation object, or a JSON **array** of observation objects:

```json
[
  {
    "uri": "custom.marketing.campaigns",
    "properties": {
      "type": "table",
      "name": "campaigns",
      "tags": ["marketing", "campaigns"]
    }
  },
  {
    "uri": "custom.marketing.campaign_analytics",
    "properties": {
      "type": "view",
      "name": "campaign_analytics",
      "tags": ["marketing", "analytics"]
    }
  }
]
```

### JSONL / NDJSON (`.jsonl`, `.ndjson`) — recommended for large volumes

One observation per line (newline-delimited JSON). Empty lines are ignored.

```jsonl
{"uri": "custom.marketing.campaigns", "properties": {"type": "table", "name": "campaigns"}}
{"uri": "custom.marketing.campaign_analytics", "properties": {"type": "view", "name": "campaign_analytics"}}
```

Each line must be a complete JSON object with the same structure as inline observations (`uri` plus supported `properties`). See [Custom Integration Properties Reference](/sources/custom-integrations/custom-integration/custom-integration-properties-reference.md).

## Prerequisites

* A configured **Custom** source in Euno
* Your **integration key** (Bearer token) from the Custom source's trigger key settings
* Your **account ID** and **integration ID** (visible in the upload endpoint URL shown in the Euno UI)

## Upload Sequence (cURL)

Replace `YOUR_ACCOUNT_ID`, `YOUR_INTEGRATION_ID`, and `YOUR_INTEGRATION_KEY_HERE` with your values.

### Step 1: Request a signed upload URL

```bash
curl -X POST \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_INTEGRATION_KEY_HERE" \
  -d '{"filename": "observations.jsonl", "content_type": "application/x-ndjson"}' \
  https://api.app.euno.ai/accounts/YOUR_ACCOUNT_ID/integrations/YOUR_INTEGRATION_ID/prepare-upload
```

Example response:

```json
{
  "operation_id": 12345,
  "upload": {
    "url": "https://storage.googleapis.com/...",
    "method": "PUT",
    "headers": {
      "Content-Type": "application/x-ndjson"
    }
  }
}
```

Supported `content_type` values:

| File extension      | `content_type`         |
| ------------------- | ---------------------- |
| `.json`             | `application/json`     |
| `.jsonl`, `.ndjson` | `application/x-ndjson` |

If you omit `content_type`, Euno selects the appropriate default based on the filename.

### Step 2: Upload the file to the signed URL

Use the `url`, `method`, and `headers` from the prepare-upload response:

```bash
curl -X PUT \
  -H "Content-Type: application/x-ndjson" \
  --data-binary @observations.jsonl \
  "SIGNED_URL_FROM_STEP_1"
```

When the upload succeeds, Euno automatically starts processing the file. No further API call is required.

### Step 3: Verify the run

1. Open the **Sources** page in Euno
2. Select your Custom source
3. Review the latest run in the operations history
4. Open the run report for counts of valid observations, skipped rows, and validation errors

## Python Upload Script

This script follows the same prepare-upload sequence as the cURL example above.

### Prerequisites

* Python 3.6+
* `requests` library (`pip install requests`)

### Script

```python
import json
import os
from pathlib import Path

import requests

# Configuration — replace with your values
endpoint_url = (
    "https://api.app.euno.ai/accounts/YOUR_ACCOUNT_ID/"
    "integrations/YOUR_INTEGRATION_ID/prepare-upload"
)
integration_key = "your_key_here"
observations_file_path = "observations.jsonl"

headers = {
    "authorization": f"Bearer {integration_key}",
    "content-type": "application/json",
}


def content_type_for_filename(filename: str) -> str:
    lower_name = filename.lower()
    if lower_name.endswith(".json"):
        return "application/json"
    if lower_name.endswith((".jsonl", ".ndjson")):
        return "application/x-ndjson"
    raise ValueError("File must be .json, .jsonl, or .ndjson")


def get_signed_upload():
    filename = Path(observations_file_path).name
    body = {
        "filename": filename,
        "content_type": content_type_for_filename(filename),
    }
    response = requests.post(endpoint_url, headers=headers, json=body, timeout=30)
    if response.status_code != 200:
        print(f"Failed to obtain signed URL (status {response.status_code}): {response.text}")
        return None
    return response.json()


def upload_observations_file(prepare_response: dict) -> bool:
    upload = prepare_response["upload"]
    upload_url = upload["url"]
    upload_headers = upload.get("headers", {})
    try:
        with open(observations_file_path, "rb") as observations_file:
            response = requests.put(
                upload_url,
                data=observations_file,
                headers=upload_headers,
                timeout=300,
            )
        if response.status_code not in (200, 201):
            print(f"Upload failed (status {response.status_code}): {response.text[:200]}")
            return False
        return True
    except Exception as exc:
        print("An error occurred during upload:", exc)
        return False


def main():
    if not os.path.exists(observations_file_path):
        print(f"Error: file '{observations_file_path}' not found.")
        return False

    print("Requesting signed upload URL...")
    prepare_response = get_signed_upload()
    if not prepare_response:
        return False

    print(f"Operation ID: {prepare_response.get('operation_id')}")
    print("Uploading observations file...")

    if upload_observations_file(prepare_response):
        print("Upload succeeded. Euno will process the file automatically.")
        print(json.dumps(prepare_response, indent=2))
        return True

    print("Upload failed.")
    return False


if __name__ == "__main__":
    main()
```

### Usage

1. **Configure the script**: Set `endpoint_url`, `integration_key`, and `observations_file_path`
2. **Install dependencies**: `pip install requests`
3. **Prepare your file**: Export observations as `.json`, `.jsonl`, or `.ndjson`
4. **Run the script**: `python upload_observations.py`
5. **Verify in Euno**: Confirm the run completed successfully in the source's operation history

## Important Notes

* **One input source per run** — provide either inline JSON (via `/run`) **or** one observations file, not both.
* **One file per run** — each prepare-upload request accepts a single observations file.
* **Invalid rows are skipped** — malformed lines or observations that fail validation are counted in the run report; the run continues processing the rest of the file.
* **Reuse your integration key** — the same Bearer token used for inline `/run` requests works for prepare-upload.
* **Generating a new trigger key** — go to the **Sources** page, open the three-dot menu on your Custom source, and select **Reset Trigger Key**.

## Related Documentation

* [Custom Integration setup and inline uploads](/sources/custom-integrations/custom-integration.md)


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.euno.ai/sources/custom-integrations/custom-integration/handling-high-volume-of-observations.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.