Bulk calculations with Python

Use case

Perform multiple calculations in a single batch request to improve speed and reduce API overhead.

When you have many sets of input data, it is much more efficient to process them together in one API call rather than sending a separate request for each set. This approach reduces network traffic, minimizes waiting time, and improves overall performance when working with large datasets.

In many cases, you can perform bulk calculations directly using the provided CalcTree plugins.

This example, however, shows how to perform the same bulk calculation process by interacting with the CalcTree API directly using Python.

Example

This example demonstrates how to automate sending multiple rows of data through a CalcTree template page using the API.

The script is broken into four main parts:

1. Extract IDs: Extract the workspaceId and calculationId from a CalcTree URL.

2. Fetch Page Structure: Retrieve the calculation structure to match input names to statementIds.

3. Submit Batch Mutations: Package all input rows into a single GraphQL batch request.

4. Process Results: Parse the JSON response and format output values (including units) into a clean results array.

We’ll step through each part individually, and then bring everything together into a full working example at the end.

1. Setup and Constants

Define the API endpoint and revision ID.

GRAPHQL_ENDPOINT = "https://graph.calctree.com/graphql"
REVISION_ID = "fffffffffff"

2. Helper Functions

2.1 Extract IDs from URL

Here, we extract the workspaceId and calculationId from the page URL.

from typing import List, Dict, Tuple

def extract_ids_from_url(url: str) -> Tuple[str, str]:
    """Extract workspaceId and calculationId from a CalcTree URL."""
    parts = url.strip().split("/")
    if len(parts) >= 6 and parts[3] == "edit":
        workspace_id = parts[4]
        calculation_id = parts[5]
        return workspace_id, calculation_id
    else:
        raise ValueError("Invalid URL format. Expected format: https://.../edit/{workspaceId}/{calculationId}")

2.2 Fetch Calculation Structure

Here, we use the getCalculation query to collect the structure of the target calculation and statementIds.

import requests
import json

def fetch_calculation_structure(api_key: str, workspace_id: str, calculation_id: str) -> List[Dict]:
    """Fetches calculation structure to retrieve parameter names and statement IDs."""
    query = """
    query GetCalculation($workspaceId: ID!, $calculationId: ID!, $revisionId: ID!) {
      calculation(workspaceId: $workspaceId, calculationId: $calculationId, revisionId: $revisionId) {
        calculationId
        statements {
          statementId
          title
          engine
          formula
          namedValues {
            name
            value
          }
        }
      }
    }
    """
    payload = {
        "query": query,
        "variables": {
            "workspaceId": workspace_id,
            "calculationId": calculation_id,
            "revisionId": REVISION_ID,
        }
    }
    headers = {
        "Content-Type": "application/json",
        "x-api-key": api_key,
    }
    response = requests.post(GRAPHQL_ENDPOINT, headers=headers, json=payload)
    if not response.ok:
        raise Exception(f"Failed to fetch calculation structure: {response.status_code} {response.text}")
    data = response.json()
    if "errors" in data:
        raise Exception(f"GraphQL errors: {data['errors']}")
    return data.get("data", {}).get("calculation", {}).get("statements", [])

2.3 Format Output Values

CalcTree's API returns results for parameters as json objects which break parameters up into their values, formula, units etc. This function extracts values and the default units of the response parameters and returns them in a neet results array.

def format_output_value(raw_value) -> str:
    """Formats output values to handle units and simple numbers."""
    if isinstance(raw_value, str):
        try:
            parsed = json.loads(raw_value)
            if isinstance(parsed, dict):
                if parsed.get("mathjs") == "Unit":
                    value = parsed.get("value", "")
                    unit = parsed.get("unit", "")
                    try:
                        value = round(float(value), 6)
                    except (TypeError, ValueError):
                        pass
                    return f"{value} {unit}".strip()
                if parsed.get("type") == "image":
                    return "[Image]"
        except json.JSONDecodeError:
            pass
    return str(raw_value)

3. API Call: Batch Calculation Mutation

Here, we use the Calculate GraphQL query, but structure it to send a batch request. Instead of making a separate API call for each row of input data, we package all rows into a single API call.

This approach is much more efficient because it reduces network overhead and speeds up the bulk processing of larger datasets.

def calculate_mutations_batch(api_key: str, batch_requests: List[Dict]) -> List[Dict]:
    """Send a batch of GraphQL mutations to perform multiple calculations."""
    variables = {}
    mutations = []

    for idx, req in enumerate(batch_requests):
        base_name = f"EXL_BTCH_calc{idx}"
        variables[f"{base_name}_workspaceId"] = req["workspaceId"]
        variables[f"{base_name}_calculationId"] = req["calculationId"]
        variables[f"{base_name}_revisionId"] = REVISION_ID
        variables[f"{base_name}_statements"] = json.loads(req["statements"])

        mutation = f"""
        {base_name}: calculate(
          workspaceId: ${base_name}_workspaceId,
          calculationId: ${base_name}_calculationId,
          revisionId: ${base_name}_revisionId,
          statements: ${base_name}_statements
        ) {{
          statements {{
            statementId
            errors
            namedValues {{
              name
              value
            }}
          }}
        }}
        """
        mutations.append(mutation)

    query = f"""
    mutation ({', '.join([f"${key}: {'[StatementInput!]!' if key.endswith('_statements') else 'ID!'}" for key in variables])}) {{
      {''.join(mutations)}
    }}
    """
    headers = {
        "Content-Type": "application/json",
        "x-api-key": api_key,
    }
    payload = {
        "query": query,
        "variables": variables,
    }
    response = requests.post(GRAPHQL_ENDPOINT, headers=headers, json=payload)
    if not response.ok:
        raise Exception(f"Batch calculation API call failed: {response.status_code} {response.text}")
    
    data = response.json()

    results = []
    for idx in range(len(batch_requests)):
        base_name = f"EXL_BTCH_calc{idx}"
        result = data.get("data", {}).get(base_name)
        if not result:
            results.append({"error": f"Missing result for {base_name}"})
            continue
        output = {}
        for stmt in result.get("statements", []):
            for nv in stmt.get("namedValues", []):
                output[nv["name"]] = nv["value"]
        results.append(output)

    return results

4. Main Function: Bulk Calculate

The bulk_calculate function is the driver of the entire bulk calculation process.

It calls the prior functions to performs the following steps:

• Extracts the workspace ID and calculation ID from the page URL.

• Fetches the structure of the calculation to match input parameter names to their correct statementId.

• Organizes each input row into a set of calculation statements.

• Builds a list of batch requests to submit together.

• Calls calculate_mutations_batch to execute all calculations in one API call.

• Formats the output by cleaning up unit values and organizing results into a structured table.

This function allows you to run many calculations at once on a CalcTree page using only one API request, making the process both scalable and fast.

def bulk_calculate(api_key: str, url: str, input_data: List[List], output_names: List = None) -> List[List]:
    """Perform bulk calculations on a CalcTree template page."""
    workspace_id, calculation_id = extract_ids_from_url(url)

    if not input_data or len(input_data) < 2:
        raise ValueError("Input data must include at least a header and one data row.")

    input_names = input_data[0]
    input_values = input_data[1:]

    # Fetch page structure
    statements = fetch_calculation_structure(api_key, workspace_id, calculation_id)

    batch_requests = []
    for row in input_values:
        if len(row) != len(input_names):
            raise ValueError("Input row length must match header length.")

        statements_list = []
        for name, value in zip(input_names, row):
            matching_statement = next(
                (stmt for stmt in statements if any(nv.get("name") == name for nv in stmt.get("namedValues", []))),
                None
            )
            if not matching_statement:
                raise ValueError(f"No matching statement found for input: {name}")

            statement_id = matching_statement["statementId"]
            statements_list.append({
                "statementId": statement_id,
                "formula": f"{name}={value}"
            })

        batch_requests.append({
            "workspaceId": workspace_id,
            "calculationId": calculation_id,
            "statements": json.dumps(statements_list),
            "apiKey": api_key
        })

    batch_results = calculate_mutations_batch(api_key, batch_requests)

    # Format results
    formatted_results = []
    all_keys = []
    for res in batch_results:
        if isinstance(res, dict) and "error" not in res:
            all_keys = list(res.keys())
            break

    if not all_keys:
        return [["Error: No valid results returned."]]

    formatted_results.append(all_keys)

    for res in batch_results:
        row = []
        for key in all_keys:
            raw_value = res.get(key, "")
            row.append(format_output_value(raw_value))
        formatted_results.append(row)

    return formatted_results

5. Example Usage

Here we run the the bulk_calculate function by feeding input data into the function.

Note that the header row of the input data array must use the name of the parameter, as opposed to the display name shown on CalcTree pages. You can find this name in the settings for your parameters in CalcTree, or from the getCalculation API call (used here in step 2.2).

if __name__ == "__main__":
    API_KEY = "YOUR_API_KEY"
    URL = "https://app.calctree.com/edit/your-workspace-id/your-calculation-id"

    input_data = [
        ["w", "l"],
        [1, 4],
        [2, 5],
        [3, 4.5],
        [4, 6],
        [5, 3],
        [5, 4]
    ]

    result_table = bulk_calculate(API_KEY, URL, input_data)

    for row in result_table:
        print(row)

6. Results

For a calculation page that multiples w*l, to calculate Area, this returns the results:

['w', 'l', 'area']
['1', '4', '10.0 m^2']
['2', '5', '10.0 m^2']
['3', '4.5', '10.0 m^2']
['4', '6', '10.0 m^2']
['5', '3', '10.0 m^2']
['5', '4', '10.0 m^2']