Clari Deploy Integration

Overview

Deploy Clari export pipelines to production environments: Airflow DAGs, AWS Lambda, or Google Cloud Functions for scheduled, serverless execution.

Prerequisites

• Environment-specific cloud project/account and least-privilege runtime role
• Secret-manager reference for the Clari token, never a literal deployment value
• A reviewed export schedule, retry policy, and dead-letter/incident route
• Separate non-production validation and production approval gates

Instructions

Airflow DAG

# dags/clari_export_dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
from datetime import datetime, timedelta

def export_clari_forecast(**context):
    from clari_client import ClariClient, ClariConfig

    client = ClariClient(ClariConfig(
        api_key=Variable.get("clari_api_key"),
    ))

    period = context["params"].get("period", "2026_Q1")
    data = client.export_and_download("company_forecast", period)

    entries = data.get("entries", [])
    context["ti"].xcom_push(key="entry_count", value=len(entries))
    # Load to warehouse here

dag = DAG(
    "clari_daily_export",
    schedule_interval="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)

export_task = PythonOperator(
    task_id="export_forecast",
    python_callable=export_clari_forecast,
    dag=dag,
)

AWS Lambda

# lambda_handler.py
import json
import boto3
from clari_client import ClariClient, ClariConfig

def handler(event, context):
    ssm = boto3.client("ssm")
    api_key = ssm.get_parameter(
        Name="/clari/api-key", WithDecryption=True
    )["Parameter"]["Value"]

    client = ClariClient(ClariConfig(api_key=api_key))
    data = client.export_and_download(
        event.get("forecast_name", "company_forecast"),
        event.get("period", "2026_Q1"),
    )

    return {
        "statusCode": 200,
        "body": json.dumps({"entries": len(data.get("entries", []))}),
    }

Google Cloud Function

# main.py
import functions_framework
from google.cloud import secretmanager
from clari_client import ClariClient, ClariConfig

@functions_framework.http
def clari_export(request):
    sm = secretmanager.SecretManagerServiceClient()
    secret = sm.access_secret_version(name="projects/my-proj/secrets/clari-api-key/versions/latest")
    api_key = secret.payload.data.decode()

    client = ClariClient(ClariConfig(api_key=api_key))
    data = client.export_and_download("company_forecast", "2026_Q1")

    return {"entries": len(data.get("entries", []))}

Error Handling

Output

Produce a deployment receipt with runtime version, environment, secret reference, scheduled scope, release identifier, health check, and rollback decision. Return only aggregate job status to callers; preserve forecast data, tokens, and provider download URLs inside the authorized processing boundary.

Examples

Deploy a staging worker using a dedicated service role, run a read-only export, and verify that the emitted record count and job status match the approved manifest. Promote through an environment gate only after health and retry tests pass; on timeout or missing secret, roll back the release and investigate without widening permissions.

Resources

• Airflow Documentation
• AWS Lambda

Next Steps

For webhook setup, see clari-webhooks-events.