Python Upload for Zip Artifacts

Use this guide to programmatically upload a .zip file to any Euno source integration that accepts zip artifacts via the prepare-upload endpoint.

This flow is shared by:

• dbt Core

• Cube Core

• Hex Semantic Project

• Matillion DPC

For Custom integration bulk uploads (.json, .jsonl, .ndjson), see Handling High Volume of Observations.

Prerequisites

• Python 3.6+

• requests library (pip install requests)

• A saved source with a trigger secret and prepare-upload endpoint URL from the post-save modal

Prepare your zip file

Create a single .zip file before running the script. What to include depends on your integration:

Integration	What to zip
dbt Core	manifest.json, catalog.json, run_results.json, and optional semantic_manifest.json from your dbt target/ directory
Cube Core	Your Cube model/ directory (model/**/*.yml or model/**/*.yaml)
Hex Semantic Project	dbt_project.yml plus dbt MetricFlow YAML files with Hex configuration
Matillion DPC	Matillion export artifacts required by your source configuration

Example for Cube Core:

cd /path/to/your/cube/project
zip -r cube-model.zip model/

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"
)
trigger_secret = "your_trigger_secret_here"
zip_file_path = "artifacts.zip"

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


def get_signed_url():
    """Request a signed upload URL for the zip file."""
    body = {"filename": Path(zip_file_path).name}
    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

    payload = response.json()
    upload_url = payload.get("upload", {}).get("url")

    if not upload_url:
        print("Signed upload URL not found in response:", json.dumps(payload)[:500])
        return None

    return upload_url, payload


def upload_file(upload_url):
    """Upload the zip file to the signed URL."""
    try:
        with open(zip_file_path, "rb") as file_handle:
            response = requests.put(
                upload_url,
                data=file_handle,
                headers={"content-type": "application/zip"},
                timeout=120,
            )

        if response.status_code not in (200, 201):
            print(f"Upload failed (status {response.status_code}): {response.text[:200]}")
            return False

        return True

    except OSError as exc:
        print("An error occurred during upload:", exc)
        return False


def main():
    print("Starting zip artifact upload...")

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

    print("Requesting signed URL...")
    result = get_signed_url()
    if not result:
        return False

    upload_url, meta = result
    print("Signed URL obtained — uploading archive...")

    if upload_file(upload_url):
        print("Upload succeeded! Euno will start processing automatically.")
        print(json.dumps(meta, indent=2))
        return True

    print("Upload failed.")
    return False


if __name__ == "__main__":
    main()

Usage

1. Configure the script: Set endpoint_url, trigger_secret, and zip_file_path

2. Create the zip: Package the files required by your integration (see Prepare your zip file)

3. Install dependencies: pip install requests

4. Run the script: Execute after your pipeline produces the artifacts

5. Verify the run: Open your source on the Sources page and review the latest operation and run report

cURL alternative

Step 1: Request a signed upload URL

curl -X POST \
  -H "Authorization: Bearer YOUR_TRIGGER_SECRET" \
  -H "Content-Type: application/json" \
  -d '{"filename": "artifacts.zip"}' \
  "https://api.app.euno.ai/accounts/YOUR_ACCOUNT_ID/integrations/YOUR_INTEGRATION_ID/prepare-upload"

Step 2: Upload the zip to the signed URL

Use the url from the upload object in the response:

curl -X PUT \
  -H "Content-Type: application/zip" \
  --data-binary @artifacts.zip \
  "SIGNED_URL_FROM_STEP_1"

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