feat: compile a query to warehouse SQL without executing it (#30)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: jpetey75

docs/SDK_GUIDE.md

@@ -373,6 +373,40 @@ lazy_df = result.to_df_lazy()
 
 ---
 
+## Compiling to SQL
+
+Use `.compile()` to get the warehouse SQL for a query **without executing it**.
+Nothing is run and no rows are fetched — it returns the exact SQL Lightdash
+would issue:
+
+```python
+sql = (
+    model.query()
+    .metrics(model.metrics.revenue)
+    .dimensions(model.dimensions.country)
+    .filter(model.dimensions.status == "active")
+    .compile()
+)
+print(sql)
+# SELECT ... FROM ... WHERE ... GROUP BY ... LIMIT 500
+```
+
+This is handy for inspecting or debugging a query, or for running it **directly
+against your warehouse** when you need more rows than the query API returns —
+for example feeding it to BigQuery/bigframes, dbt, or a data pipeline:
+
+```python
+import bigframes.pandas as bpd
+
+sql = model.query().metrics(model.metrics.revenue).limit(10_000_000).compile()
+df = bpd.read_gbq(sql)   # run it yourself, no Lightdash row cap
+```
+
+The query's `limit` is included in the SQL as `LIMIT n`; set a large limit (or
+strip the trailing clause) if you're running it yourself.
+
+---
+
 ## SQL Runner
 
 Execute raw SQL queries directly against your data warehouse:

lightdash/query.py

@@ -1,5 +1,6 @@
 """Query functionality for Lightdash models."""
 import time
+import uuid
 import warnings
 from typing import Any, Dict, List, Optional, Union, Sequence, Iterator
 
@@ -12,6 +13,28 @@
 from .results import BaseResult
 
 
+def _inject_filter_ids(filters: Dict[str, Any]) -> Dict[str, Any]:
+    """Add the ``id`` fields the v1 ``compileQuery`` endpoint requires on every
+    filter group and rule.
+
+    The v2 query endpoint injects these server-side, but v1 validates them
+    strictly. Returns a new dict; the input is not mutated.
+    """
+    def visit_group(group: Dict[str, Any]) -> Dict[str, Any]:
+        out: Dict[str, Any] = {"id": str(uuid.uuid4())}
+        for key in ("and", "or"):
+            if key in group:
+                out[key] = [visit_item(item) for item in group[key]]
+        return out
+
+    def visit_item(item: Dict[str, Any]) -> Dict[str, Any]:
+        if "and" in item or "or" in item:  # nested group
+            return visit_group(item)
+        return {**item, "id": str(uuid.uuid4())}  # leaf rule
+
+    return {ftype: visit_group(group) for ftype, group in filters.items()}
+
+
 class _QueryExecutor:
     """Internal class that handles V2 async query submission and polling."""
 
@@ -693,6 +716,54 @@ def execute(
         )
         return self._result
 
+    def compile(self) -> str:
+        """
+        Compile the query to warehouse SQL without executing it.
+
+        Returns the SQL Lightdash would run for this query. Nothing is executed
+        and no rows are fetched — useful for inspecting or debugging a query, or
+        for running it directly against your warehouse (e.g. BigQuery/bigframes,
+        dbt, or a data pipeline) when you need more rows than the query API
+        returns.
+
+        The query's ``limit`` is included in the SQL (as ``LIMIT n``); set a
+        large limit, or strip the trailing clause, if you intend to run it
+        yourself without Lightdash's row cap.
+
+        Returns:
+            The compiled SQL as a string.
+
+        Example:
+            sql = (
+                model.query()
+                .metrics(model.metrics.revenue)
+                .dimensions(model.dimensions.country)
+                .filter(model.dimensions.status == "active")
+                .compile()
+            )
+        """
+        if self._model._client is None:
+            raise RuntimeError("Model not properly initialized with client reference")
+
+        client = self._model._client
+        payload = self._build_payload()
+
+        # The v1 compileQuery endpoint is stricter than the v2 query endpoint:
+        # it requires additionalMetrics and an `id` on every filter group/rule.
+        payload.setdefault("additionalMetrics", [])
+        filters = payload.get("filters", {})
+        has_rules = any(
+            group.get("and") or group.get("or") for group in filters.values()
+        )
+        payload["filters"] = _inject_filter_ids(filters) if has_rules else {}
+
+        result = client._make_request(
+            "POST",
+            f"/api/v1/projects/{client.project_uuid}/explores/{self._model.name}/compileQuery",
+            json=payload,
+        )
+        return result["query"]
+
     def to_records(self) -> List[Dict[str, Any]]:
         """Get all query results as a list of dictionaries."""
         return self.execute().to_records()

tests/test_acceptance.py

@@ -643,3 +643,30 @@ def test_query_with_filters_class(first_model):
         row[dim1_label] == filter1_value and row[dim2_label] == filter2_value
         for row in filtered_results
     ), "Filters did not correctly filter results"
+
+
+def test_compile_query(first_model):
+    """Compiling a query returns warehouse SQL without executing it."""
+    dimensions = first_model.list_dimensions()
+    metrics = first_model.list_metrics()
+    if not dimensions or not metrics:
+        pytest.skip("No dimensions or metrics available for testing")
+
+    # Plain compile -> SELECT ... FROM ...
+    sql = first_model.query(
+        dimensions=[dimensions[0].field_id],
+        metrics=[metrics[0].field_id],
+        limit=10,
+    ).compile()
+    assert isinstance(sql, str)
+    assert "select" in sql.lower()
+    assert "from" in sql.lower()
+
+    # With a filter -> exercises filter-id injection on the v1 endpoint
+    sql_filtered = first_model.query(
+        dimensions=[dimensions[0].field_id],
+        metrics=[metrics[0].field_id],
+        filters=dimensions[0].is_not_null(),
+        limit=10,
+    ).compile()
+    assert "where" in sql_filtered.lower()

tests/test_compile.py

@@ -0,0 +1,78 @@
+"""
+Unit tests for Query.compile() — compile a query to warehouse SQL (issue #5).
+
+Uses a fake client (no network); the live path is covered by the env-gated
+acceptance test ``test_compile_query``.
+"""
+
+import pytest
+from lightdash.models import Model
+from lightdash.dimensions import Dimension
+
+
+class FakeClient:
+    project_uuid = "proj-uuid"
+
+    def __init__(self):
+        self.requests = []
+
+    def get_query_limits(self):
+        return {}
+
+    def _make_request(self, method, path, params=None, json=None):
+        self.requests.append({"method": method, "path": path, "json": json})
+        return {"query": "SELECT 1 AS x FROM `t`", "parameterReferences": []}
+
+
+@pytest.fixture
+def model():
+    m = Model(name="orders", type="default", database_name="db", schema_name="s")
+    m._client = FakeClient()
+    return m
+
+
+def dim(name):
+    return Dimension(name=name, model_name="orders")
+
+
+class TestCompile:
+    def test_returns_sql_string(self, model):
+        sql = model.query(metrics=["orders_revenue"], dimensions=["orders_country"]).compile()
+        assert sql == "SELECT 1 AS x FROM `t`"
+
+    def test_posts_to_compile_endpoint(self, model):
+        model.query(metrics=["orders_revenue"]).compile()
+        req = model._client.requests[-1]
+        assert req["method"] == "POST"
+        assert req["path"] == "/api/v1/projects/proj-uuid/explores/orders/compileQuery"
+
+    def test_payload_includes_additional_metrics(self, model):
+        model.query(metrics=["orders_revenue"]).compile()
+        assert model._client.requests[-1]["json"]["additionalMetrics"] == []
+
+    def test_no_filters_sends_empty_object(self, model):
+        """With no filters, send {} — the shape the v1 endpoint accepts."""
+        model.query(metrics=["orders_revenue"]).compile()
+        assert model._client.requests[-1]["json"]["filters"] == {}
+
+    def test_filters_get_ids_on_group_and_rules(self, model):
+        """The v1 endpoint requires an id on every filter group and rule."""
+        model.query(metrics=["orders_revenue"]).filter(dim("country") == "USA").compile()
+        filters = model._client.requests[-1]["json"]["filters"]
+        group = filters["dimensions"]
+        assert "id" in group
+        rule = group["and"][0]
+        assert "id" in rule
+        assert rule["target"]["fieldId"] == "orders_country"
+        assert rule["operator"] == "equals"
+        assert rule["values"] == ["USA"]
+
+    def test_limit_included(self, model):
+        """The limit flows into the payload (and thus the SQL) unbounded."""
+        model.query(metrics=["orders_revenue"]).limit(10_000_000).compile()
+        assert model._client.requests[-1]["json"]["limit"] == 10_000_000
+
+    def test_requires_client(self):
+        m = Model(name="orders", type="default", database_name="db", schema_name="s")
+        with pytest.raises(RuntimeError, match="not properly initialized"):
+            m.query(metrics=["orders_revenue"]).compile()