Implementation:TobikoData Sqlmesh API Lineage
| Knowledge Sources | |
|---|---|
| Domains | Web_Server, REST_API |
| Last Updated | 2026-02-07 20:00 GMT |
Overview
Concrete tool for analyzing column and model lineage graphs provided by the SQLMesh web server.
Description
This module exposes FastAPI endpoints for lineage analysis. The column lineage endpoint traces data flow for a specific column through models, CTEs, and derived tables, returning an adjacency list representation of the lineage graph. It supports two modes: full lineage with SQL expressions, or models-only lineage for simpler dependency graphs. The model lineage endpoint returns the full DAG dependencies for a model.
Usage
These endpoints are called by the SQLMesh web UI lineage visualization components. The column lineage endpoint populates detailed column-level dependency graphs showing transformations and sources. The models-only mode provides faster responses for high-level dependency visualization. The model lineage endpoint shows all upstream and downstream model dependencies.
Code Reference
Source Location
- Repository: TobikoData_Sqlmesh
- File: web/server/api/endpoints/lineage.py
Signature
@router.get("/{model_name:str}/{column_name:str}")
def column_lineage(
model_name: str,
column_name: str,
models_only: bool = False,
context: Context = Depends(get_loaded_context),
) -> t.Dict[str, t.Dict[str, LineageColumn]]
@router.get("/{model_name:str}")
def model_lineage(
model_name: str,
context: Context = Depends(get_loaded_context),
) -> t.Dict[str, t.Set[str]]
Import
from web.server.api.endpoints.lineage import router
I/O Contract
Inputs
| Endpoint | Method | Parameters | Description |
|---|---|---|---|
| /api/lineage/{model}/{column} | GET | model_name, column_name, models_only | Returns column lineage with optional SQL expressions |
| /api/lineage/{model} | GET | model_name | Returns model DAG lineage |
Outputs
| Endpoint | Response Type | Description |
|---|---|---|
| /api/lineage/{model}/{column} | Dict[str, Dict[str, LineageColumn]] | Adjacency list of column dependencies with expressions |
| /api/lineage/{model} | Dict[str, Set[str]] | DAG graph of model dependencies |
Usage Examples
# Get column lineage with full details
import httpx
response = httpx.get(
"http://localhost:8000/api/lineage/my_schema.my_model/revenue"
)
lineage = response.json()
# {
# "my_schema.my_model": {
# "revenue": {
# "expression": "SUM(amount)",
# "source": "SELECT SUM(amount) AS revenue FROM ...",
# "models": {"source_table": ["amount"]}
# }
# }
# }
# Get models-only lineage (faster)
response = httpx.get(
"http://localhost:8000/api/lineage/my_schema.my_model/revenue",
params={"models_only": True}
)
# Get model lineage
response = httpx.get(
"http://localhost:8000/api/lineage/my_schema.my_model"
)
model_deps = response.json()
# {"upstream_model1": {"downstream_model1"}, ...}