Clari Sdk Patterns

Clari SDK Patterns

Overview

Clari has no official SDK -- build typed wrappers around the v4 REST API. These patterns cover the Export API for forecasts, job polling, and data transformation pipelines.

Prerequisites

• Completed clari-install-auth setup
• Python 3.10+ (primary) or TypeScript 5+

Instructions

Step 1: Python Client

# clari_client.py
import os
import time
import requests
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ClariConfig:
    api_key: str
    base_url: str = "https://api.clari.com/v4"
    poll_interval: int = 5
    max_poll_attempts: int = 60

class ClariClient:
    def __init__(self, config: Optional[ClariConfig] = None):
        self.config = config or ClariConfig(
            api_key=os.environ["CLARI_API_KEY"]
        )
        self.session = requests.Session()
        self.session.headers.update({
            "apikey": self.config.api_key,
            "Content-Type": "text/plain",
        })

    def list_forecasts(self) -> list[dict]:
        resp = self.session.get(f"{self.config.base_url}/export/forecast/list")
        resp.raise_for_status()
        return resp.json()["forecasts"]

    def export_forecast(
        self,
        forecast_name: str,
        time_period: str,
        types: list[str] = None,
        currency: str = "USD",
        export_format: str = "JSON",
    ) -> dict:
        payload = {
            "timePeriod": time_period,
            "typesToExport": types or [
                "forecast", "quota", "forecast_updated",
                "adjustment", "crm_total", "crm_closed"
            ],
            "currency": currency,
            "schedule": "NONE",
            "includeHistorical": False,
            "exportFormat": export_format,
        }

        resp = self.session.post(
            f"{self.config.base_url}/export/forecast/{forecast_name}",
            json=payload,
        )
        resp.raise_for_status()
        return resp.json()

    def wait_for_job(self, job_id: str) -> dict:
        for attempt in range(self.config.max_poll_attempts):
            resp = self.session.get(
                f"{self.config.base_url}/export/jobs/{job_id}",
            )
            resp.raise_for_status()
            status = resp.json()

            if status["status"] == "COMPLETED":
                return status
            if status["status"] == "FAILED":
                raise ClariExportError(f"Job {job_id} failed: {status}")

            time.sleep(self.config.poll_interval)

        raise ClariExportError(f"Job {job_id} timed out after {self.config.max_poll_attempts} attempts")

    def download_export(self, download_url: str) -> dict:
        resp = requests.get(download_url)
        resp.raise_for_status()
        return resp.json()

    def export_and_download(
        self, forecast_name: str, time_period: str
    ) -> dict:
        job = self.export_forecast(forecast_name, time_period)
        completed = self.wait_for_job(job["jobId"])
        return self.download_export(completed["downloadUrl"])

class ClariExportError(Exception):
    pass

Step 2: TypeScript Client

// clari-client.ts
interface ClariConfig {
  apiKey: string;
  baseUrl?: string;
  pollIntervalMs?: number;
  maxPollAttempts?: number;
}

interface ForecastExport {
  entries: ForecastEntry[];
}

interface ForecastEntry {
  ownerName: string;
  ownerEmail: string;
  forecastAmount: number;
  quotaAmount: number;
  crmTotal: number;
  crmClosed: number;
  adjustmentAmount: number;
  timePeriod: string;
}

class ClariClient {
  private apiKey: string;
  private baseUrl: string;
  private pollIntervalMs: number;
  private maxPollAttempts: number;

  constructor(config: ClariConfig) {
    this.apiKey = config.apiKey;
    this.baseUrl = config.baseUrl ?? "https://api.clari.com/v4";
    this.pollIntervalMs = config.pollIntervalMs ?? 5000;
    this.maxPollAttempts = config.maxPollAttempts ?? 60;
  }

  private async request<T>(path: string, options?: RequestInit): Promise<T> {
    const response = await fetch(`${this.baseUrl}${path}`, {
      ...options,
      headers: {
        apikey: this.apiKey,
        "Content-Type": "text/plain",
        ...options?.headers,
      },
    });

    if (!response.ok) {
      throw new Error(`Clari API ${response.status}: ${await response.text()}`);
    }

    return response.json();
  }

  async listForecasts(): Promise<{ forecasts: any[] }> {
    return this.request("/export/forecast/list");
  }

  async exportForecast(forecastName: string, timePeriod: string): Promise<any> {
    return this.request(`/export/forecast/${forecastName}`, {
      method: "POST",
      body: JSON.stringify({
        timePeriod,
        typesToExport: ["forecast", "quota", "crm_total", "crm_closed"],
        currency: "USD",
        schedule: "NONE",
        includeHistorical: false,
        exportFormat: "JSON",
      }),
    });
  }

  async exportAndDownload(
    forecastName: string,
    timePeriod: string
  ): Promise<ForecastExport> {
    const job = await this.exportForecast(forecastName, timePeriod);
    const completed = await this.waitForJob(job.jobId);
    const resp = await fetch(completed.downloadUrl);
    return resp.json();
  }

  private async waitForJob(jobId: string): Promise<any> {
    for (let i = 0; i < this.maxPollAttempts; i++) {
      const status = await this.request(`/export/jobs/${jobId}`);
      if (status.status === "COMPLETED") return status;
      if (status.status === "FAILED") throw new Error(`Job failed: ${jobId}`);
      await new Promise((r) => setTimeout(r, this.pollIntervalMs));
    }
    throw new Error(`Job ${jobId} timed out`);
  }
}

Error Handling

Resources

• Clari API Reference
• Clari Community API Guide

Next Steps

Apply patterns in clari-core-workflow-a for forecast export pipelines.