clari-performance-tuning

name	clari-performance-tuning
description	Optimize Clari API performance with caching, batch exports, and data pipeline efficiency. Use when exports take too long, optimizing data warehouse load times, or reducing API calls in multi-forecast environments. Trigger with phrases like "clari performance", "clari slow export", "optimize clari pipeline", "clari caching".
allowed-tools	Read, Write, Edit
version	1.6.0
license	MIT
author	Jeremy Longshore <[e-mail]>
tags	["saas","revenue-intelligence","forecasting","clari"]
compatibility	Designed for Claude Code

Clari Performance Tuning

Overview

Optimize Clari export pipelines: reduce export times, cache forecast data, and parallelize multi-period exports.

Instructions

Parallel Multi-Period Export

from concurrent.futures import ThreadPoolExecutor, as_completed

def parallel_export(
    client,
    forecast_name: str,
    periods: list[str],
    max_workers: int = 3,
) -> dict[str, list[dict]]:
    results = {}

    def export_period(period: str) -> tuple[str, list[dict]]:
        data = client.export_and_download(forecast_name, period)
        return period, data.get("entries", [])

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(export_period, p): p for p in periods
        }
        for future in as_completed(futures):
            period, entries = future.result()
            results[period] = entries
            print(f"  {period}: {len(entries)} entries")

    return results

Cache Export Results

import json
import hashlib
from pathlib import Path
from datetime  datetime, timedelta

 :
     ():
        .cache_dir = Path(cache_dir)
        .cache_dir.mkdir(parents=, exist_ok=)
        .ttl = timedelta(hours=ttl_hours)

     () -> :
         hashlib.md5(.encode()).hexdigest()

     () -> [] | :
        path = .cache_dir / 
          path.exists():
             
        meta = json.loads(path.read_text())
        cached_at = datetime.fromisoformat(meta[])
         datetime.utcnow() - cached_at > .ttl:
             
         meta[]

     ():
        path = .cache_dir / 
        path.write_text(json.dumps({
            : datetime.utcnow().isoformat(),
            : entries,
        }))