API Reference
  1. API Guides

Python Examples

Python Examples

All examples use the requests library and the Spartera REST API directly. Replace YOUR_API_KEY, YOUR_COMPANY_ID, and YOUR_USER_ID with your actual values before running.

Setup

import requests
import json

BASE_URL = "https://api.spartera.com"
API_KEY = "YOUR_API_KEY"
COMPANY_ID = "YOUR_COMPANY_ID"
USER_ID = "YOUR_USER_ID"

HEADERS = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json"
}

Connections

List All Connections

def list_connections():
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/connections",
        headers=HEADERS
    )
    data = response.json()

    for conn in data["data"]:
        print(f"{conn['connection_id']} — {conn['name']} (engine_id: {conn['engine_id']})")

    return data["data"]

connections = list_connections()

Create a BigQuery Connection

def create_bigquery_connection(name, description, service_account_json, provider_domain):
    """
    Create a new BigQuery connection.

    Args:
        name: Descriptive connection name
        description: What data this connection contains
        service_account_json: dict — your GCP service account key file contents
        provider_domain: Domain where the data originates (e.g. "yourcompany.com")
    """
    payload = {
        "company_id": COMPANY_ID,
        "user_id": USER_ID,
        "engine_id": 1,  # BigQuery
        "name": name,
        "description": description,
        "provider_domain": provider_domain,
        "credential_type": "SERVICE_ACCOUNT",
        "credentials": json.dumps(service_account_json),
        "verified_usage_ability": True
    }

    response = requests.post(
        f"{BASE_URL}/companies/{COMPANY_ID}/connections",
        headers=HEADERS,
        json=payload
    )
    result = response.json()

    if response.status_code == 200:
        connection_id = result["data"]["connection_id"]
        print(f"✅ Connection created: {connection_id}")
        return connection_id
    else:
        print(f"❌ Error: {result}")
        return None

# Load your service account key file
with open("service-account-key.json") as f:
    sa_key = json.load(f)

connection_id = create_bigquery_connection(
    name="Production BigQuery — Analytics",
    description="Main analytics warehouse. Customer and revenue data, refreshed daily at 2 AM UTC.",
    service_account_json=sa_key,
    provider_domain="yourcompany.com"
)

Create a Snowflake Connection

def create_snowflake_connection(name, description, account, user, password,
                                 role, warehouse, database, schema, provider_domain):
    payload = {
        "company_id": COMPANY_ID,
        "user_id": USER_ID,
        "engine_id": 15,  # Snowflake
        "name": name,
        "description": description,
        "provider_domain": provider_domain,
        "credential_type": "USERNAME_PASSWORD",
        "credentials": json.dumps({
            "account": account,
            "user": user,
            "password": password,
            "role": role,
            "warehouse": warehouse,
            "database": database,
            "schema": schema
        }),
        "verified_usage_ability": True
    }

    response = requests.post(
        f"{BASE_URL}/companies/{COMPANY_ID}/connections",
        headers=HEADERS,
        json=payload
    )
    result = response.json()

    if response.status_code == 200:
        print(f"✅ Connection created: {result['data']['connection_id']}")
        return result["data"]["connection_id"]
    else:
        print(f"❌ Error: {result}")
        return None

connection_id = create_snowflake_connection(
    name="Snowflake — Customer360",
    description="Customer 360 warehouse. Transaction history and behavioral data.",
    account="myorg-myaccount",
    user="spartera_readonly",
    password="your-password",
    role="SPARTERA_ROLE",
    warehouse="ANALYTICS_WH",
    database="ANALYTICS",
    schema="PUBLIC",
    provider_domain="yourcompany.com"
)

Test a Connection

def test_connection(connection_id):
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/connections/{connection_id}/test",
        headers=HEADERS
    )
    result = response.json()

    status = result["data"]["test_status"]
    response_time = result["data"].get("response_time_ms", "N/A")
    print(f"Test status: {status} ({response_time}ms)")

    if status == "SUCCESS":
        print("✅ Connection is working")
    elif status == "PARTIAL":
        print("⚠️  Connection partially successful")
        for rec in result["data"].get("recommendations", []):
            print(f"   → {rec}")
    else:
        print(f"❌ Connection failed: {result['data'].get('error_message')}")

    return status

test_connection(connection_id)

Delete a Connection

def delete_connection(connection_id):
    response = requests.delete(
        f"{BASE_URL}/companies/{COMPANY_ID}/connections/{connection_id}",
        headers=HEADERS
    )
    if response.status_code == 200:
        print(f"✅ Connection {connection_id} deleted")
    else:
        print(f"❌ Error: {response.json()}")

delete_connection(connection_id)

Calculation Assets

Create a Calculation Asset

def create_calculation_asset(name, description, connection_id, sql_logic,
                              industry_id=None, tags=None):
    """
    Create a new calculation asset.

    Args:
        name: Unique asset name
        description: What this asset returns and the business question it answers
        connection_id: ID of the connection to run SQL against
        sql_logic: The SQL SELECT query (must return an aggregated/computed result)
        industry_id: Optional industry classification ID
        tags: Optional list of tag strings
    """
    payload = {
        "company_id": COMPANY_ID,
        "user_id": USER_ID,
        "name": name,
        "description": description,
        "connection_id": connection_id,
        "asset_type": "CALCULATION",
        "sql_logic": sql_logic,
    }

    if industry_id:
        payload["industry_id"] = industry_id

    if tags:
        payload["tags"] = tags

    response = requests.post(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets",
        headers=HEADERS,
        json=payload
    )
    result = response.json()

    if response.status_code == 200:
        asset_id = result["data"]["asset_id"]
        print(f"✅ Asset created: {asset_id}")
        return asset_id
    else:
        print(f"❌ Error: {result}")
        return None


# Example: Monthly churn rate
asset_id = create_calculation_asset(
    name="90-Day Customer Churn Rate",
    description=(
        "Returns the customer churn rate as a percentage for the rolling 90-day window. "
        "Sourced from our production CRM database, updated daily. "
        "Use this to monitor retention health and trigger churn prevention campaigns."
    ),
    connection_id=connection_id,
    sql_logic="""
        SELECT
            ROUND(
                COUNT(CASE WHEN status = 'churned' THEN 1 END) * 100.0
                / NULLIF(COUNT(*), 0),
                2
            ) AS churn_rate_pct
        FROM customers
        WHERE created_at >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
    """,
    tags=["churn", "retention", "customers", "daily"]
)

Save Schema (Enable Dynamic Parameters)

After creating the asset, save the schema to allow buyers to filter and customize executions.

def save_asset_schema(asset_id):
    """
    Trigger schema introspection and save — enables dynamic parameter filtering.
    Must be called after the asset SQL is saved and valid.
    """
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/infoschema/save",
        headers=HEADERS
    )
    result = response.json()

    if response.status_code == 200:
        print(f"✅ Schema saved for asset {asset_id}")
        return result["data"]
    else:
        print(f"❌ Schema save failed: {result}")
        return None

save_asset_schema(asset_id)

Preview / Test an Asset

def test_asset(asset_id):
    """Run a preview execution against 10% of your data."""
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/test",
        headers=HEADERS
    )
    result = response.json()

    if response.status_code == 200:
        print(f"✅ Test result: {result['data']}")
        return result["data"]
    else:
        print(f"❌ Test failed: {result}")
        return None

test_result = test_asset(asset_id)

Set a Price

def set_asset_price(asset_id, price_usd):
    """
    Set the per-execution price for an asset.
    You earn 80% of this amount per execution.
    """
    response = requests.post(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/prices",
        headers=HEADERS,
        json={"price_usd": price_usd}
    )
    result = response.json()

    if response.status_code == 200:
        credits = result["data"].get("price_credits")
        print(f"✅ Price set: ${price_usd} ({credits} credits)")
        print(f"   Your earnings per execution: ${price_usd * 0.80:.2f}")
        return result["data"]
    else:
        print(f"❌ Error: {result}")
        return None

set_asset_price(asset_id, price_usd=5.00)

Publish to Marketplace

def publish_asset(asset_id, geographic_coverage_type="GLOBAL",
                  data_source_refresh_frequency="DAILY"):
    """
    Publish an asset to the Spartera marketplace.
    Requires Stripe to be configured for your company.

    geographic_coverage_type options:
        GLOBAL, CONTINENTAL, REGIONAL, NATIONAL, STATE, LOCAL, CUSTOM, UNKNOWN

    data_source_refresh_frequency options:
        REAL_TIME, HOURLY, DAILY, WEEKLY, MONTHLY, QUARTERLY, ANNUAL,
        ONE_TIME, CUSTOM, UNKNOWN
    """
    payload = {
        "sell_in_marketplace": True,
        "geographic_coverage_type": geographic_coverage_type,
        "data_source_refresh_frequency": data_source_refresh_frequency
    }

    response = requests.patch(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}",
        headers=HEADERS,
        json=payload
    )
    result = response.json()

    if response.status_code == 200:
        print(f"✅ Asset {asset_id} is now live in the marketplace")
        return result["data"]
    else:
        print(f"❌ Error: {result}")
        return None

publish_asset(asset_id)

Full Calculation Asset Workflow

def create_and_publish_calculation(connection_id, name, description, sql, price_usd, tags=None):
    """End-to-end: create → test → price → publish."""

    print(f"\n🚀 Creating calculation asset: {name}")

    # 1. Create
    asset_id = create_calculation_asset(name, description, connection_id, sql, tags=tags)
    if not asset_id:
        return None

    # 2. Test
    print("🔍 Running preview test...")
    test_result = test_asset(asset_id)
    if not test_result:
        print("⚠️  Test failed — review SQL before publishing")
        return asset_id

    # 3. Save schema
    print("💾 Saving schema...")
    save_asset_schema(asset_id)

    # 4. Set price
    print(f"💰 Setting price to ${price_usd}...")
    set_asset_price(asset_id, price_usd)

    # 5. Publish
    print("📢 Publishing to marketplace...")
    publish_asset(asset_id)

    print(f"\n✅ Done! Asset {asset_id} is live.")
    return asset_id


asset_id = create_and_publish_calculation(
    connection_id=connection_id,
    name="Average Revenue Per User — Last 30 Days",
    description=(
        "Returns ARPU as a single USD value for the rolling 30-day window. "
        "Calculated from transaction data in our production warehouse. "
        "Refreshed daily. Use for cohort benchmarking and pricing strategy."
    ),
    sql="""
        SELECT
            ROUND(SUM(revenue_usd) / NULLIF(COUNT(DISTINCT user_id), 0), 2) AS arpu_usd
        FROM transactions
        WHERE transaction_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
          AND status = 'completed'
    """,
    price_usd=3.00,
    tags=["arpu", "revenue", "users", "30-day", "daily"]
)

Visualization Assets

Visualization assets are configured via a single JSON column called viz_spec,
which holds a Plotly figure JSON.
The legacy viz_chart_type / viz_dep_var_col_name / viz_indep_var_col_name
columns still exist on the asset record (the API returns them for read
access) but are no longer used for chart rendering. Only viz_spec drives
the renderer.

See Visualization Spec Reference for the full schema,
the *src column-reference convention, validation rules, and the build-time
vs. render-time hydration model.

Create a Visualization Asset

This single helper accepts any valid viz_spec dict. Examples below construct
specs for common chart shapes.

def create_visualization_asset(name, description, connection_id,
                                schema_table, viz_spec,
                                data_limit=0, tags=None):
    """
    Create a visualization asset.

    Args:
        name: Unique asset name
        description: What the chart shows and the business question it answers
        connection_id: ID of the connection the renderer hydrates data from
        schema_table: "schema_name.table_name" — the source table referenced
                      by the *src keys in viz_spec
        viz_spec: A Plotly figure JSON (dict). Use *src keys (xsrc, ysrc,
                  labelssrc, valuessrc, etc.) to reference columns in
                  schema_table; the renderer hydrates them at execution time.
        data_limit: Optional row cap (1-10,000). 0 (default) means use the
                    platform max of 10,000. Lower values cap source data
                    earlier for performance — e.g., 100 for a focused chart,
                    1000 for a typical trend line.
        tags: Optional list of tag strings
    """
    schema_name, table_name = schema_table.split(".")

    payload = {
        "company_id": COMPANY_ID,
        "user_id": USER_ID,
        "name": name,
        "description": description,
        "connection_id": connection_id,
        "asset_type": "VISUALIZATION",
        "source_schema_name": schema_name,
        "source_table_name": table_name,
        "viz_spec": viz_spec,
        "viz_data_limit": data_limit,
    }

    if tags:
        payload["tags"] = tags

    response = requests.post(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets",
        headers=HEADERS,
        json=payload
    )
    result = response.json()

    if response.status_code == 200:
        asset_id = result["data"]["asset_id"]
        print(f"✅ Visualization asset created: {asset_id}")
        return asset_id
    else:
        print(f"❌ Error: {result}")
        return None

Example: Bar Chart

bar_spec = {
    "data": [
        {
            "type": "bar",
            "xsrc": "product_name",
            "ysrc": "total_revenue_usd",
            "orientation": "v"
        }
    ],
    "layout": {
        "title": "Top Products by Revenue"
    }
}

asset_id = create_visualization_asset(
    name="Top 10 Products by Revenue — Current Month",
    description=(
        "Bar chart showing the top revenue-generating products for the current "
        "month. Sourced from the sales warehouse, refreshed daily."
    ),
    connection_id=connection_id,
    schema_table="sales.product_summary",
    viz_spec=bar_spec,
    tags=["products", "revenue", "bar-chart", "monthly"]
)

Example: Line Chart

The editor's "Line" entry emits a scatter trace with mode: "lines"
(or "lines+markers") — there is no Plotly line trace type. Authoring
programmatically uses the same shape. See the trace-type tables in
Visualization Spec Reference for the full list of
synthetic UI types and their canonical Plotly mappings.

line_spec = {
    "data": [
        {
            "type": "scatter",
            "mode": "lines+markers",
            "xsrc": "month_date",
            "ysrc": "mrr_usd"
        }
    ],
    "layout": {
        "title": "Monthly Recurring Revenue — 12 Month Trend",
        "xaxis": {"title": "Month"},
        "yaxis": {"title": "MRR (USD)"}
    }
}

asset_id = create_visualization_asset(
    name="Monthly Recurring Revenue — 12 Month Trend",
    description=(
        "Line chart showing MRR trend over the last 12 months. "
        "Sourced from subscription data, refreshed monthly."
    ),
    connection_id=connection_id,
    schema_table="finance.monthly_revenue",
    viz_spec=line_spec,
    tags=["mrr", "revenue", "line-chart", "monthly", "trend"]
)

Example: Area Chart

Same pattern as line — Plotly has no area trace type. The editor's
"Area" entry emits a scatter trace with mode: "lines" and stackgroup
set, which is what triggers the filled-area rendering.

area_spec = {
    "data": [
        {
            "type": "scatter",
            "mode": "lines",
            "stackgroup": "1",
            "xsrc": "month_date",
            "ysrc": "active_users"
        }
    ],
    "layout": {
        "title": "Monthly Active Users"
    }
}

asset_id = create_visualization_asset(
    name="Monthly Active Users \u2014 Stacked Area",
    description=(
        "Area chart showing active users by month. Sourced from product "
        "analytics warehouse, refreshed daily."
    ),
    connection_id=connection_id,
    schema_table="product.monthly_active_users",
    viz_spec=area_spec,
    tags=["users", "engagement", "area-chart", "monthly"]
)

Example: Pie Chart

pie_spec = {
    "data": [
        {
            "type": "pie",
            "labelssrc": "segment_name",
            "valuessrc": "revenue_usd"
        }
    ],
    "layout": {
        "title": "Revenue Share by Market Segment"
    }
}

asset_id = create_visualization_asset(
    name="Revenue Share by Market Segment",
    description=(
        "Pie chart showing revenue distribution across market segments. "
        "Sourced from CRM and billing data. Updated monthly."
    ),
    connection_id=connection_id,
    schema_table="market.segment_revenue",
    viz_spec=pie_spec,
    tags=["segments", "revenue", "market", "pie-chart"]
)

Testing a Visualization Asset

After creating a visualization, render it against live data to verify it
looks right. The /test endpoint returns a JSON envelope; for the default
_download_type=json, the rendered chart's signed URL is at
data.asset_value.

def test_visualization_asset(asset_id, output_format="json"):
    """
    Render a visualization asset against live data.

    Args:
        asset_id: ID of the asset to test
        output_format: One of:
          - "json"    (default) — full envelope with signed URL + metadata
          - "base64"  — PNG bytes embedded as base64 inside data.asset_value
          - "image"   — minimal envelope with just data.url
          - "download" — same as "image"

    Returns:
        Parsed response dict. On success, structure depends on output_format
        (see viz-spec-reference > Testing your visualization). On failure,
        an error envelope with data.error_type / data.error_source.
    """
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/test",
        headers=HEADERS,
        params={"_download_type": output_format}
    )
    return response.json()


# Validation-only mode — cheap check that the spec is structurally valid,
# without rendering. Use this in CI / health checks.
def validate_visualization_asset(asset_id):
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/test",
        headers=HEADERS,
        params={"_bool": "true"}
    )
    payload = response.json()
    return payload.get("data", {}).get("asset_value", {}).get("is_valid", False)


# Usage — full metadata response:
result = test_visualization_asset(asset_id)
data = result.get("data", {})

if data.get("format") == "png_url":
    print(f"Rendered chart: {data['asset_value']}")
    print(f"  expires in: {data['expires_in']}s, size: {data['size_bytes']} bytes")
    # Download the PNG bytes:
    png_bytes = requests.get(data["asset_value"]).content
elif data.get("error_type"):
    print(f"Render failed: {result['message']} "
          f"(type={data['error_type']}, source={data['error_source']})")
else:
    print(f"Unexpected response: {result}")


# Alternative usage — get PNG bytes directly without a follow-up request:
import base64
result = test_visualization_asset(asset_id, output_format="base64")
data = result.get("data", {})
if data.get("format") == "png_base64":
    png_bytes = base64.b64decode(data["asset_value"])
    with open("chart.png", "wb") as f:
        f.write(png_bytes)

Editing a Visualization Asset

Use PATCH to modify a visualization. Only include the fields you want to
update — everything else stays unchanged.

def update_visualization_spec(asset_id, new_viz_spec):
    """Replace the viz_spec on an existing visualization asset."""
    response = requests.patch(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}",
        headers=HEADERS,
        json={"viz_spec": new_viz_spec}
    )
    return response.json()


# Example: switch a bar chart from vertical to horizontal orientation
result = update_visualization_spec(
    asset_id=asset_id,
    new_viz_spec={
        "data": [
            {
                "type": "bar",
                "xsrc": "product_name",
                "ysrc": "total_revenue_usd",
                "orientation": "h"  # Changed from "v"
            }
        ],
        "layout": {"title": "Top Products by Revenue (Horizontal)"}
    }
)
# On success: {"message": "success", "data": {"asset_id": "..."}}

# Re-test after editing to confirm the change rendered as expected.
test_result = test_visualization_asset(asset_id)

Discovering Source Table Columns

When constructing viz_spec, you may need to know what columns exist in
the source table. Use the infoschema endpoint:

def get_source_columns(asset_id):
    """List the columns in the source table feeding this visualization."""
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/infoschema",
        headers=HEADERS
    )
    return response.json()


# For picking sensible filter values, scan a column's distinct values:
def scan_distinct_values(asset_id, column, limit=100):
    response = requests.post(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/scan_column",
        headers=HEADERS,
        json={"column": column, "limit": limit}
    )
    return response.json()


# Usage:
columns = get_source_columns(asset_id)
regions = scan_distinct_values(asset_id, "region")

Handling Errors

All viz asset endpoints wrap responses in a {"message": ..., "data": {...}}
envelope. For the test endpoint specifically, render-time errors put their
diagnostic detail inside data (the error_type, error_source, request_id,
etc.). Route the response based on data.error_source:

import requests

def render_with_error_handling(asset_id):
    result = test_visualization_asset(asset_id)
    data = result.get("data", {})

    if data.get("format") == "png_url":
        # Success — return the signed URL
        return data["asset_value"]

    error_type = data.get("error_type")
    error_source = data.get("error_source")
    message = result.get("message", "Unknown error")

    if error_source == "BUYER":
        # Caller's filters/inputs caused the failure (NO_DATA_FOUND,
        # INVALID_QUERY, ACCESS_DENIED, QUOTA_EXCEEDED, etc.)
        print(f"Buyer-side issue ({error_type}): {message}")
    elif error_source == "SELLER":
        # The asset configuration is wrong (DATA_STRUCTURE_ERROR,
        # CONNECTION_ERROR, TIMEOUT_ERROR, etc.) — the asset owner
        # needs to fix the spec or the data source.
        print(f"Asset misconfigured ({error_type}): {message}")
    elif error_source == "SPARTERA":
        # Platform-side failure (VISUALIZATION_ERROR, UNEXPECTED_ERROR,
        # etc.) — retry; if persistent, contact support.
        print(f"Platform error ({error_type}): {message}")
    else:
        print(f"Unknown error: {message}")

    return None

See Visualization Spec Reference > Error responses
for the full table of error_type values.

Other Chart Types

The editor supports 33 trace types across six categories (Simple,
Distributions, 3D, Maps, Financial, Specialized). For the canonical type
value for each, see the "Editor-supported trace types" tables in
Visualization Spec Reference. Once you have the type
value, consult the Plotly trace reference for that trace
type's specific attributes, then construct a spec with the appropriate
*src keys pointing at columns in your source table. The structure is
always the same: {"data": [{"type": "...", "...src": "...column..."}], "layout": {...}}.

External API Assets

Create an External API Calculation Asset

def create_external_api_asset(name, description, connection_id,
                               price_usd, function_id=None,
                               schema_parameters=None, tags=None):
    """
    Create a Calculation asset backed by an External API connection.

    Args:
        connection_id: Must be an External API connection (engine_id = 20)
        function_id: Optional routing key appended to GET URL or POST body
        schema_parameters: List of parameter definitions for buyer input
        price_usd: Per-execution price
    """
    payload = {
        "company_id": COMPANY_ID,
        "user_id": USER_ID,
        "name": name,
        "description": description,
        "connection_id": connection_id,
        "asset_type": "CALCULATION",  # External API only supports CALCULATION
    }

    if function_id:
        payload["function_id"] = function_id

    # Schema parameters define typed inputs buyers provide before execution
    if schema_parameters:
        payload["asset_schema"] = {
            "tables": [{
                "name": "parameters",
                "columns": schema_parameters
            }]
        }

    if tags:
        payload["tags"] = tags

    response = requests.post(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets",
        headers=HEADERS,
        json=payload
    )
    result = response.json()

    if response.status_code == 200:
        asset_id = result["data"]["asset_id"]
        print(f"✅ External API asset created: {asset_id}")

        # Set price
        set_asset_price(asset_id, price_usd)
        return asset_id
    else:
        print(f"❌ Error: {result}")
        return None


# Schema parameter definitions
churn_parameters = [
    {
        "name": "customer_segment",
        "column_alias": "Customer Segment",
        "type": "STRING",
        "description": "The customer segment to score",
        "placeholder_text": "enterprise",
        "valid_values": ["enterprise", "smb", "startup", "consumer"]
    },
    {
        "name": "lookback_days",
        "column_alias": "Lookback Window (Days)",
        "type": "INTEGER",
        "description": "Number of days of activity to include in the prediction",
        "placeholder_text": "90"
    },
    {
        "name": "reference_date",
        "column_alias": "Reference Date",
        "type": "DATE",
        "description": "Date to score against (YYYY-MM-DD)",
        "placeholder_text": "2025-04-01"
    }
]

asset_id = create_external_api_asset(
    name="Customer Churn Probability Score",
    description=(
        "Returns a churn probability score (0.0–1.0) for a given customer segment and "
        "lookback window. Powered by our production gradient boosting model (94% accuracy). "
        "Sub-100ms latency. Updated weekly with new training data."
    ),
    connection_id="your-external-api-connection-id",
    price_usd=8.00,
    function_id="churn_predictor",
    schema_parameters=churn_parameters,
    tags=["churn", "ml", "prediction", "customers", "real-time"]
)

Common Operations

Get a Single Asset

def get_asset(asset_id):
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}",
        headers=HEADERS
    )
    result = response.json()
    asset = result["data"][0]
    print(f"Name: {asset['name']}")
    print(f"Type: {asset['asset_type']}")
    print(f"In marketplace: {asset.get('sell_in_marketplace', False)}")
    return asset

asset = get_asset(asset_id)

List All Assets (Paginated)

def list_assets(page=1, per_page=20, asset_type=None, marketplace_only=False):
    params = {
        "page": page,
        "per_page": per_page,
        "sort_field": "date_created",
        "sort_direction": "desc"
    }
    if asset_type:
        params["asset_type"] = asset_type        # CALCULATION or VISUALIZATION
    if marketplace_only:
        params["marketplace"] = "true"

    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets",
        headers=HEADERS,
        params=params
    )
    result = response.json()

    print(f"Page {result['page']} of {result['total_pages']} ({result['total']} total assets)")
    for asset in result["data"]:
        print(f"  {asset['asset_id']} — {asset['name']} ({asset['asset_type']})")

    return result

list_assets(asset_type="CALCULATION")

Update an Asset

def update_asset(asset_id, **kwargs):
    """Update any asset fields. Pass keyword args for what you want to change."""
    response = requests.patch(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}",
        headers=HEADERS,
        json={"company_id": COMPANY_ID, "user_id": USER_ID, **kwargs}
    )
    result = response.json()

    if response.status_code == 200:
        print(f"✅ Asset {asset_id} updated")
        return result["data"]
    else:
        print(f"❌ Error: {result}")
        return None

# Examples
update_asset(asset_id, description="Updated with Q2 2025 data coverage")
update_asset(asset_id, sell_in_marketplace=False)             # Unpublish
update_asset(asset_id, rate_limit_number=100,                 # Add rate limiting
                        rate_limit_period="HOUR",
                        rate_limit_granularity="USER")

Update Price

def update_price(asset_id, new_price_usd):
    """Price changes take effect immediately for new executions."""
    return set_asset_price(asset_id, new_price_usd)

update_price(asset_id, new_price_usd=7.50)

Get Price History

def get_price_history(asset_id):
    response = requests.get(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/prices?active=all",
        headers=HEADERS
    )
    result = response.json()

    print(f"Price history for {asset_id}:")
    for record in result["data"]:
        status = "ACTIVE" if record["active"] else "expired"
        print(f"  ${record['price_usd']} ({record['price_credits']} credits) — {status} — {record['date_created']}")

    return result["data"]

get_price_history(asset_id)

Unpublish an Asset

def unpublish_asset(asset_id):
    return update_asset(asset_id, sell_in_marketplace=False)

unpublish_asset(asset_id)

Delete an Asset

def delete_asset(asset_id):
    response = requests.delete(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}",
        headers=HEADERS
    )
    if response.status_code == 200:
        print(f"✅ Asset {asset_id} deleted")
    else:
        print(f"❌ Error: {response.json()}")

delete_asset(asset_id)

Execute an Asset (Full, as a Buyer)

def execute_asset(asset_id, parameters=None):
    """
    Execute an asset and get results.
    Consumes credits at the asset's current price.
    """
    payload = {}
    if parameters:
        payload["parameters"] = parameters

    response = requests.post(
        f"{BASE_URL}/companies/{COMPANY_ID}/assets/{asset_id}/process",
        headers=HEADERS,
        json=payload if payload else None
    )
    result = response.json()

    if response.status_code == 200:
        print(f"✅ Execution complete")
        print(f"   Result: {result['data']}")
        print(f"   Execution time: {result['meta']['execution_time_ms']}ms")
        print(f"   Credits used: {result['meta']['credits_used']}")
        return result
    else:
        print(f"❌ Execution failed: {result}")
        return None

# Execute without parameters
execute_asset(asset_id)

# Execute with parameters (for parameterized assets)
execute_asset(asset_id, parameters={
    "customer_segment": "enterprise",
    "lookback_days": 90,
    "reference_date": "2025-04-01"
})

Related Pages

Updated 18 days ago

Did this page help you?

Updated 18 days ago

Did this page help you?