Developer/Integrator Guide¶

import requests

def upload_workbook(file_path: str, name: str, token: str) -> dict:
    """Upload an Excel workbook to CalcBridge."""
    url = "https://api.calcbridge.io/api/v1/workbooks/upload"

    headers = {
        "Authorization": f"Bearer {token}"
    }

    with open(file_path, "rb") as f:
        files = {
            "file": (file_path, f, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
        }
        data = {
            "name": name
        }

        response = requests.post(url, headers=headers, files=files, data=data)
        response.raise_for_status()
        return response.json()

# Usage
result = upload_workbook("portfolio.xlsx", "Q1 Holdings", access_token)
print(f"Workbook ID: {result['id']}")

import time

def wait_for_processing(workbook_id: str, token: str, timeout: int = 300) -> dict:
    """Wait for workbook processing to complete."""
    url = f"https://api.calcbridge.io/api/v1/workbooks/{workbook_id}"
    headers = {"Authorization": f"Bearer {token}"}

    start_time = time.time()
    while time.time() - start_time < timeout:
        response = requests.get(url, headers=headers)
        workbook = response.json()

        if workbook["status"] == "ready":
            return workbook
        elif workbook["status"] == "failed":
            raise Exception(f"Processing failed: {workbook.get('error')}")

        time.sleep(2)  # Poll every 2 seconds

    raise TimeoutError("Workbook processing timed out")

class CalcBridgeClient:
    """Python client for CalcBridge calculations."""

    def __init__(self, api_key: str, base_url: str = "https://api.calcbridge.io"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers["X-API-Key"] = api_key

    def evaluate(self, formula: str, context: dict = None) -> any:
        """Evaluate a single formula."""
        response = self.session.post(
            f"{self.base_url}/api/v1/calculations/evaluate",
            json={"formula": formula, "context": context or {}}
        )
        response.raise_for_status()
        return response.json()["result"]

    def evaluate_batch(self, formulas: list) -> list:
        """Evaluate multiple formulas."""
        response = self.session.post(
            f"{self.base_url}/api/v1/calculations/batch",
            json={"formulas": formulas}
        )
        response.raise_for_status()
        return response.json()["results"]

    def validate(self, formula: str) -> dict:
        """Validate formula syntax."""
        response = self.session.post(
            f"{self.base_url}/api/v1/calculations/validate",
            json={"formula": formula}
        )
        response.raise_for_status()
        return response.json()

# Usage
client = CalcBridgeClient(api_key="cb_sk_live_...")
result = client.evaluate("=SUMIFS(A:A, B:B, \">100\")", {"A1": 50, "B1": 150, "A2": 100, "B2": 50})
print(f"Result: {result}")

# backend/services/compliance_service.py
from functools import lru_cache
import redis
from calcbridge import CalcBridgeClient

class ComplianceService:
    def __init__(self):
        self.client = CalcBridgeClient(api_key=os.environ["CALCBRIDGE_API_KEY"])
        self.redis = redis.Redis(host="localhost", port=6379)

    def get_compliance_summary(self, workbook_id: str) -> dict:
        """Get compliance summary with caching."""
        cache_key = f"compliance:{workbook_id}"

        # Check cache
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)

        # Fetch from API
        run = self.client.run_compliance_tests(workbook_id, "standard")
        results = self.client.get_compliance_results(run["run_id"])

        summary = {
            "total": len(results["tests"]),
            "passing": len([t for t in results["tests"] if t["status"] == "pass"]),
            "warnings": len([t for t in results["tests"] if t["status"] == "warning"]),
            "failing": len([t for t in results["tests"] if t["status"] == "fail"]),
            "tests": results["tests"],
            "last_updated": datetime.utcnow().isoformat()
        }

        # Cache for 5 minutes
        self.redis.setex(cache_key, 300, json.dumps(summary))

        return summary

def make_api_request(url: str, method: str = "GET", **kwargs) -> dict:
    """Make API request with proper error handling."""
    try:
        response = requests.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 401:
            # Token expired, refresh and retry
            refresh_token()
            return make_api_request(url, method, **kwargs)
        elif e.response.status_code == 429:
            # Rate limited, wait and retry
            retry_after = int(e.response.headers.get("Retry-After", 60))
            time.sleep(retry_after)
            return make_api_request(url, method, **kwargs)
        else:
            raise