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 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:

[
  {
    "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.

{"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.

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

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:

{
  "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:

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

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