Implementation:TobikoData Sqlmesh API Environments
| Knowledge Sources | |
|---|---|
| Domains | Web_Server, REST_API |
| Last Updated | 2026-02-07 20:00 GMT |
Overview
Concrete tool for managing SQLMesh virtual environments (list, delete) provided by the SQLMesh web server.
Description
This module exposes FastAPI endpoints for environment management operations. The GET endpoint retrieves all environments from the state reader, ensuring that prod and the default target environment always exist in the response. The DELETE endpoint invalidates an environment and triggers cleanup of expired environments through the state sync mechanism.
Usage
These endpoints are called by the SQLMesh web UI when displaying environment information and managing environment lifecycles. The GET endpoint populates environment selectors and displays environment metadata. The DELETE endpoint is invoked when users remove development or PR environments that are no longer needed.
Code Reference
Source Location
- Repository: TobikoData_Sqlmesh
- File: web/server/api/endpoints/environments.py
Signature
@router.get("", response_model=Environments)
async def get_environments(
context: Context = Depends(get_loaded_context),
) -> Environments
@router.delete("/{environment:str}")
async def delete_environment(
response: Response,
environment: str,
context: Context = Depends(get_loaded_context),
) -> None
Import
from web.server.api.endpoints.environments import router
I/O Contract
Inputs
| Endpoint | Method | Parameters | Description |
|---|---|---|---|
| /api/environments | GET | None | Retrieves all environments with metadata |
| /api/environments/{environment} | DELETE | environment | Invalidates and deletes the specified environment |
Outputs
| Endpoint | Response Type | Description |
|---|---|---|
| /api/environments | Environments | Dictionary of environments with pinned and default target info |
| /api/environments/{environment} | None | Returns 204 No Content on successful deletion |
Usage Examples
# Get all environments
import httpx
response = httpx.get("http://localhost:8000/api/environments")
environments_data = response.json()
# {
# "environments": {"prod": {...}, "dev": {...}},
# "pinned_environments": ["prod"],
# "default_target_environment": "prod"
# }
# Delete a development environment
response = httpx.delete("http://localhost:8000/api/environments/dev_branch")