""" KANDIMA SDK — Python Save as kandima.py. Zero deps — uses stdlib only (urllib, hmac, hashlib). Requires Python 3.9+. Generated from the OpenAPI spec at: https://kandima.ai/api/v1/openapi.json Usage: from kandima import Kandima k = Kandima(api_key=os.environ["KANDIMA_API_KEY"]) me = k.me() credits = k.credits() customers = k.customers(limit=50) # Outbound webhooks webhook = k.webhooks.create(url="https://my-app.com/webhooks/kandima", events=["purchase.created", "alert.fired"]) print("Save this secret:", webhook["secret"]) # Custom event ingest k.track(event="checkout_started", props={"plan": "pro"}) For HMAC signature verification on inbound webhooks see verify_webhook() at the bottom of this file. """ from __future__ import annotations import hmac import hashlib import json import time import urllib.parse import urllib.request import urllib.error from typing import Any, Dict, List, Optional class KandimaError(Exception): def __init__(self, status: int, code: str, message: str, details: Any = None): super().__init__(f"[{status}] {code}: {message}") self.status = status self.code = code self.message = message self.details = details class Kandima: """KANDIMA v1 REST client.""" def __init__( self, api_key: str, base_url: str = "https://kandima.ai", timeout: float = 15.0, retries: int = 0, ): if not api_key or not api_key.startswith("ka_"): raise ValueError("Kandima: invalid api key (must start with ka_)") self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self.retries = max(0, int(retries)) self.webhooks = _WebhooksAPI(self) # ──────────────── public health ──────────────── def status(self) -> Dict[str, Any]: return self._request("GET", "/api/v1/status", auth=False) # ──────────────── core resources ──────────────── def me(self) -> Dict[str, Any]: return self._request("GET", "/api/v1/me") def credits(self) -> Dict[str, Any]: return self._request("GET", "/api/v1/credits") def health(self) -> Dict[str, Any]: return self._request("GET", "/api/v1/health") def customers(self, *, limit: int = 100, cursor: Optional[str] = None) -> Dict[str, Any]: return self._request("GET", "/api/v1/customers", params={"limit": limit, "cursor": cursor}) def integrations(self) -> Dict[str, Any]: return self._request("GET", "/api/v1/integrations") def insights(self, *, status: Optional[str] = None, severity: Optional[str] = None) -> Dict[str, Any]: return self._request("GET", "/api/v1/insights", params={"status": status, "severity": severity}) def profit(self, *, days: int = 30) -> Dict[str, Any]: return self._request("GET", "/api/v1/profit", params={"days": days}) def conversations(self, *, channel: str = "whatsapp", status: Optional[str] = None, limit: int = 50) -> Dict[str, Any]: return self._request("GET", "/api/v1/conversations", params={"channel": channel, "status": status, "limit": limit}) def conversations_stats(self, *, days: int = 30) -> Dict[str, Any]: return self._request("GET", "/api/v1/conversations/stats", params={"days": days}) def anomalies(self, *, days: int = 7, severity: Optional[str] = None) -> Dict[str, Any]: return self._request("GET", "/api/v1/anomalies", params={"days": days, "severity": severity}) def inventory(self, *, status: Optional[str] = None, limit: int = 100) -> Dict[str, Any]: return self._request("GET", "/api/v1/inventory", params={"status": status, "limit": limit}) def reorder(self, *, status: str = "pending", limit: int = 50) -> Dict[str, Any]: return self._request("GET", "/api/v1/reorder", params={"status": status, "limit": limit}) def ads(self, *, platform: Optional[str] = None, level: Optional[str] = None, limit: int = 100) -> Dict[str, Any]: return self._request("GET", "/api/v1/ads", params={"platform": platform, "level": level, "limit": limit}) def creatives(self, *, recommendation: Optional[str] = None, platform: Optional[str] = None, sort: str = "winner", limit: int = 50) -> Dict[str, Any]: return self._request("GET", "/api/v1/creatives", params={ "recommendation": recommendation, "platform": platform, "sort": sort, "limit": limit, }) def track(self, *, event: str, props: Optional[Dict[str, Any]] = None, visitor_id: Optional[str] = None, **kwargs) -> Dict[str, Any]: body = {"event": event, "props": props or {}, "visitor_id": visitor_id, **kwargs} return self._request("POST", "/api/v1/track", body={k: v for k, v in body.items() if v is not None}) # ──────────────── internals ──────────────── def _request( self, method: str, path: str, params: Optional[Dict[str, Any]] = None, body: Optional[Dict[str, Any]] = None, auth: bool = True, ) -> Dict[str, Any]: url = self.base_url + path if params: clean = {k: v for k, v in params.items() if v is not None} if clean: url = url + ("&" if "?" in url else "?") + urllib.parse.urlencode(clean) data: Optional[bytes] = None headers = {"User-Agent": "kandima-python/1.0"} if auth: headers["Authorization"] = f"Bearer {self.api_key}" if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" attempts = self.retries + 1 last_err: Optional[Exception] = None for i in range(attempts): try: req = urllib.request.Request(url, data=data, method=method, headers=headers) with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.read().decode("utf-8") if resp else "" return json.loads(raw) if raw else {} except urllib.error.HTTPError as e: err_body = e.read().decode("utf-8", errors="replace") if e.fp else "" err_data: Any = None try: err_data = json.loads(err_body) if err_body else None except Exception: err_data = err_body code = (err_data or {}).get("error", f"http_{e.code}") if isinstance(err_data, dict) else f"http_{e.code}" if e.code >= 500 and i < attempts - 1: time.sleep(0.2 * (2 ** i)) continue raise KandimaError(e.code, code, f"KANDIMA {method} {path} failed", err_data) except urllib.error.URLError as e: last_err = e if i >= attempts - 1: raise KandimaError(0, "network_error", str(e)) time.sleep(0.2 * (2 ** i)) raise KandimaError(0, "network_error", str(last_err) if last_err else "unknown") class _WebhooksAPI: def __init__(self, k: Kandima): self.k = k def list(self) -> Dict[str, Any]: return self.k._request("GET", "/api/v1/webhooks") def create(self, *, url: str, events: List[str]) -> Dict[str, Any]: return self.k._request("POST", "/api/v1/webhooks", body={"url": url, "events": events}) def delete(self, webhook_id: int) -> Dict[str, Any]: return self.k._request("DELETE", f"/api/v1/webhooks?id={int(webhook_id)}") # ──────────────── HMAC signature verify ──────────────── def verify_webhook(raw_body: str, signature: str, timestamp: str, secret: str, max_skew_seconds: int = 300) -> None: """ Verify an inbound KANDIMA webhook signature. Pass the raw request body (str) plus the x-kandima-signature and x-kandima-timestamp headers, plus the secret you saved when creating the webhook. Raises KandimaError if invalid. """ clean_secret = secret.removeprefix("whsec_") if secret.startswith("whsec_") else secret try: ts = int(timestamp) except (TypeError, ValueError): raise KandimaError(400, "invalid_timestamp", "x-kandima-timestamp not an integer") if abs(int(time.time()) - ts) > max_skew_seconds: raise KandimaError(400, "timestamp_skew", f"Signature timestamp >{max_skew_seconds}s skew") expected = hmac.new( clean_secret.encode("utf-8"), f"{ts}.{raw_body}".encode("utf-8"), hashlib.sha256, ).hexdigest() if not hmac.compare_digest(signature, expected): raise KandimaError(401, "invalid_signature", "HMAC signature does not match")