Implementation:TobikoData Sqlmesh API Models
| Knowledge Sources | |
|---|---|
| Domains | Web_Server, REST_API |
| Last Updated | 2026-02-07 20:00 GMT |
Overview
Concrete tool for retrieving SQLMesh model metadata and definitions provided by the SQLMesh web server.
Description
This module exposes FastAPI endpoints for model operations. The list endpoint returns all models with serialized metadata including columns, types, and model details. The individual model endpoint provides full model information with rendered SQL query. Model serialization includes column descriptions derived from lineage analysis, model configuration details (kind, cron, time_column, partitioning), and computed metadata like next/previous cron execution times.
Usage
These endpoints are called by the SQLMesh web UI to display model information. The list endpoint populates model browsers and search interfaces. The individual model endpoint loads detailed model views showing SQL definitions, column schemas, and configuration. The UI uses this data to render model cards, dependency graphs, and editor views.
Code Reference
Source Location
- Repository: TobikoData_Sqlmesh
- File: web/server/api/endpoints/models.py
Signature
@router.get(
"",
response_model=t.Union[t.List[models.Model], models.ApiExceptionPayload],
response_model_exclude_unset=True,
response_model_exclude_none=True,
)
def get_models(context: Context = Depends(get_loaded_context)) -> t.List[models.Model]
@router.get(
"/{name:str}",
response_model=models.Model,
response_model_exclude_unset=True,
response_model_exclude_none=True,
)
def get_model(
name: str,
context: Context = Depends(get_loaded_context),
) -> models.Model
Import
from web.server.api.endpoints.models import router
I/O Contract
Inputs
| Endpoint | Method | Parameters | Description |
|---|---|---|---|
| /api/models | GET | None | Retrieves all models with metadata |
| /api/models/{name} | GET | name | Retrieves individual model with rendered SQL |
Outputs
| Endpoint | Response Type | Description |
|---|---|---|
| /api/models | List[Model] | Array of models sorted by name |
| /api/models/{name} | Model | Full model with columns, details, and SQL query |
Usage Examples
# Get all models
import httpx
response = httpx.get("http://localhost:8000/api/models")
models = response.json()
for model in models:
print(f"{model['name']}: {model['type']} ({model['kind']})")
# Get specific model with details
response = httpx.get("http://localhost:8000/api/models/my_schema.my_model")
model = response.json()
# {
# "name": "my_schema.my_model",
# "fqn": "catalog.my_schema.my_model",
# "path": "models/my_model.sql",
# "dialect": "snowflake",
# "columns": [
# {"name": "id", "type": "INT", "description": "Primary key"},
# {"name": "name", "type": "VARCHAR", "description": "Customer name"}
# ],
# "details": {
# "kind": "INCREMENTAL_BY_TIME_RANGE",
# "cron": "@daily",
# "time_column": "created_at | %Y-%m-%d"
# },
# "sql": "SELECT * FROM source_table WHERE ..."
# }