Skip to main content

Extending the API

How to add new endpoints to the Frappe/Konsol API layer.

Architecture​

The API lives in konsol/api.py within the Frappe app. It uses Frappe's @frappe.whitelist() decorator to expose Python functions as HTTP endpoints.

Excel VBA → HTTP POST → Frappe (Konsol) → ClickHouse SQL → JSON response

Frappe @whitelist Pattern​

Basic Endpoint​

import frappe

@frappe.whitelist()
def my_endpoint(param1, param2="default"):
    """Docstring shown in API docs."""
    # Frappe handles auth — only logged-in users can call this
    return {"result": param1}

Accessible at: GET /api/method/konsol.api.my_endpoint?param1=value

Guest Endpoint​

@frappe.whitelist(allow_guest=True)
def public_endpoint():
    return {"status": "ok"}

POST-Only Endpoint​

@frappe.whitelist(methods=["POST"])
def submit_data():
    data = frappe.request.get_data(as_text=True)
    import json
    payload = json.loads(data)
    # process payload
    return {"received": len(payload)}

ClickHouse Query Pattern​

The existing API uses a direct HTTP connection to ClickHouse with parameterized queries. The connection details live in the konsol.clickhouse module — api.py imports them rather than reading EPM Settings directly:

from konsol.clickhouse import connection_url as _ch_url
from konsol.clickhouse import get_connection as _get_ch_connection

Getting a ClickHouse Connection​

konsol.clickhouse.get_connection() reads the connection settings from the EPM Settings single and returns a dict with host, port, user, password, secure, and verify:

from konsol.clickhouse import get_connection

ch_settings = get_connection()
# {"host": ..., "port": ..., "user": ..., "password": ...,
#  "secure": <bool>, "verify": <bool>}

connection_url(conn) builds the base URL from that dict, choosing https when conn["secure"] is truthy.

Executing a Query​

The credentials are passed as an HTTP auth tuple — not concatenated into the URL or appended as query params — and the request honours the connection's verify (TLS) flag:

import requests
from konsol.clickhouse import connection_url as _ch_url

def _clickhouse_query(sql, params, ch_settings):
    url = _ch_url(ch_settings)
    query_params = dict(params)
    query_params["query"] = sql

    resp = requests.get(
        url,
        params=query_params,
        auth=(ch_settings["user"], ch_settings["password"]),
        timeout=30,
        verify=ch_settings.get("verify", True),
    )
    resp.raise_for_status()
    return resp.text.strip()

For statement-style execution (DDL, inserts) the module also exposes konsol.clickhouse.execute(sql, params=None), which POSTs the query with the same auth tuple.

Parameterized Queries (Security)​

Always use ClickHouse's parameterized query format to prevent SQL injection:

sql = """
    SELECT data_area_id, sum(period_net_amount) as total
    FROM epm_gold.gold_trial_balance
    WHERE data_area_id = {entity:String}
      AND fiscal_year = {year:Int32}
    GROUP BY data_area_id
"""
params = {"param_entity": "USMF", "param_year": "2024"}
result = _clickhouse_query(sql, params, ch_settings)

Parameter format: {name:Type} in SQL, param_name=value in query params.

Supported types: String, Int32, UInt16, Float64, etc.

Measure Safety​

Measures and other identifiers that come from user input are validated against the registry (see below) and re-checked with a safe-identifier regex before being interpolated into SQL. api.py uses:

import re
_SAFE_IDENTIFIER = re.compile(r'^[a-z][a-z0-9_]*$')
# Fully-qualified ClickHouse table: schema.table, lowercase identifiers only.
_SAFE_TABLE_NAME = re.compile(r'^[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*$')

if not _SAFE_IDENTIFIER.match(measure):
    # reject — never interpolate an unvalidated identifier
    ...

The table name itself comes from the Fact Table record but is also validated with _SAFE_TABLE_NAME before it reaches the FROM clause, as defence against a tampered or mistyped clickhouse_table value.

Adding a New Scenario / Fact / Measure​

Scenarios, source tables, and allowed measures are not hard-coded in api.py. There are no SCENARIO_TABLES or ALLOWED_MEASURES dicts. Validation is registry-driven: it reads from the Fact Table and Measure doctypes at request time. To make a new Gold model queryable through =EPM(), you create records — no code change is required.

The validation flow lives in api.py._resolve_and_validate() (and the matching grouping in _batch_query_clickhouse()):

  • _get_fact(fact=..., scenario=...) looks up a Fact Table record by fact_name (wins) or scenario_key.
  • _get_allowed_measures(fact) parses the fact's measures JSON; the request's measure must be in that set and in _published_measures() (the set of Measure records with status = "Published").
  • _get_fact_dimensions(fact) parses the fact's dimensions JSON; every requested dimension must be allowed by the fact.

1. Create the Measure record(s)​

Create a Measure doctype record (Module: EPM) for each measure the new fact exposes. Key fields:

  • measure_name — e.g. forecast_amount (the value clients pass as measure=)
  • expression — SQL expression, e.g. sum(forecast_amount)
  • label, cube_type (sum / count / avg)
  • status — must be Published for the measure to validate

2. Create the Fact Table record​

Create a Fact Table doctype record (Module: EPM) pointing at the Gold model:

  • fact_name — e.g. forecast (the value clients can pass as fact=; autoname is field:fact_name)
  • scenario_key — e.g. forecast (the value clients pass as scenario=)
  • clickhouse_table — e.g. epm_gold.gold_forecast_model (validated as lowercase schema.table)
  • has_scenario_id — check if the table supports scenario_id filtering
  • fact_measures (child table) — add a row per allowed Measure; each row is validated to reference a Published Measure. Saving mirrors these rows into the hidden measures JSON field that api.py reads.
  • fact_dimensions (child table) — the dimensions this fact exposes; mirrored into the hidden dimensions JSON field.
  • status — set to Published. Use the record's Publish action to apply schema (ClickHouse DDL + dbt source where applicable) and trigger a rebuild.

Optional reroute fields (reroute_measure / reroute_table / reroute_column) let a single measure resolve against a different table/column.

3. Test​

curl "http://localhost:8069/api/method/konsol.api.epm_value?\
entity=USMF&year=2025&period=5&account=401100&\
measure=forecast_amount&scenario=forecast" \
  -b cookies.txt

You can also target a fact directly with fact=forecast; fact wins over scenario when both are supplied.

Adding a Standalone Endpoint​

For endpoints that don't follow the EPM value pattern:

from konsol.clickhouse import get_connection as _get_ch_connection

@frappe.whitelist()
def get_entities():
    """Return all legal entities."""
    ch = _get_ch_connection()
    sql = "SELECT DISTINCT data_area_id, entity_name FROM epm_gold.gold_trial_balance ORDER BY data_area_id"
    result = _clickhouse_query(sql, {}, ch)
    entities = []
    for line in result.split('\n'):
        if line.strip():
            parts = line.split('\t')
            entities.append({"id": parts[0], "name": parts[1] if len(parts) > 1 else parts[0]})
    return entities

Error Handling​

Use Frappe's exception types. To validate a scenario/fact/measure, resolve it against the registry rather than a hard-coded dict — _resolve_and_validate() returns (fact_doc, error_or_None) and already builds the "Allowed: ..." list from the Fact Table / Measure records:

import frappe

@frappe.whitelist()
def validated_endpoint(scenario, measure="period_net_amount"):
    fact_doc, err = _resolve_and_validate(None, scenario, measure, dim_names=[])
    if err:
        # err already lists the valid scenarios/measures from the registry
        frappe.throw(err, frappe.ValidationError)
    # ... proceed with fact_doc.clickhouse_table

Response Format​

Frappe wraps return values in {"message": <your_return_value>}. The VBA module and other clients expect this wrapper.

# In Python:
return {"value": 123.45}

# Client receives:
# {"message": {"value": 123.45}}

Next Steps​