From clari-pack
Manages Clari API rate limits with Python exponential backoff for export job polling and sequential scheduling to handle 429 errors.
npx claudepluginhub jeremylongshore/claude-code-plugins-plus-skills --plugin clari-packThis skill is limited to using the following tools:
The Clari API enforces rate limits per API key. Export jobs are asynchronous and queued server-side, so the primary concern is polling frequency and concurrent export requests.
Builds typed Python and TypeScript clients for Clari v4 REST API, handling forecast exports, job polling, and data pipelines.
Handles Klaviyo API rate limits with Retry-After backoff, exponential retries, and queuing for 429 errors to optimize request throughput.
Handles ClickUp API rate limits using exponential backoff with jitter, header monitoring, and retry logic for 429 errors. Optimizes throughput against per-plan limits.
Share bugs, ideas, or general feedback.
The Clari API enforces rate limits per API key. Export jobs are asynchronous and queued server-side, so the primary concern is polling frequency and concurrent export requests.
| Aspect | Value |
|---|---|
| Scope | Per API key |
| Response on limit | HTTP 429 |
| Export job queue | Server-managed, async |
| Recommended polling | 5-10 second intervals |
import time
import requests
def poll_with_backoff(
job_id: str,
api_key: str,
max_attempts: int = 60,
base_delay: float = 5.0,
max_delay: float = 60.0,
) -> dict:
for attempt in range(max_attempts):
resp = requests.get(
f"https://api.clari.com/v4/export/jobs/{job_id}",
headers={"apikey": api_key},
)
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", base_delay))
time.sleep(retry_after)
continue
resp.raise_for_status()
status = resp.json()
if status["status"] in ("COMPLETED", "FAILED"):
return status
delay = min(base_delay * (1.5 ** attempt), max_delay)
time.sleep(delay)
raise TimeoutError(f"Job {job_id} did not complete in {max_attempts} attempts")
def export_all_periods(
client,
forecast_name: str,
periods: list[str],
delay_between: float = 10.0,
) -> list[dict]:
results = []
for period in periods:
print(f"Exporting {period}...")
job = client.export_forecast(forecast_name, period)
result = poll_with_backoff(job["jobId"], client.config.api_key)
results.append(result)
time.sleep(delay_between) # Avoid hitting rate limits
return results
| Scenario | Detection | Response |
|---|---|---|
| 429 with Retry-After | Check header | Wait exact duration |
| 429 without header | Status code only | Backoff from 5s |
| Job queue full | Multiple pending jobs | Wait for completion before new exports |
For security configuration, see clari-security-basics.