Implementation:TobikoData Sqlmesh StateReader Get Environments

Knowledge Sources	SQLMesh SQLMesh Docs
Domains	Data_Engineering, Environment_Management
Last Updated	2026-02-07 00:00 GMT

Overview

Concrete tool for retrieving environment metadata and state provided by SQLMesh.

Description

The StateReader interface provides read-only access to environment metadata stored in SQLMesh's state synchronization backend. It offers three key methods: get_environments() returns all environments with full details, get_environment() retrieves a specific environment by name, and get_environments_summary() returns lightweight summaries of all environments. These methods query the state database (which can be the data warehouse itself or an external database) and deserialize environment metadata including snapshot references, timestamps, and configuration.

The StateReader abstraction allows different state backend implementations (in-memory, database, cloud storage) while providing a consistent interface. Environment objects include all snapshots (model versions), promoted snapshot IDs (subset with views created), plan IDs, timestamps for creation/finalization/expiration, and naming configuration.

Usage

Use StateReader.get_environments() and related methods when you need to:

• Display a list of all environments in a UI or CLI
• Check if a specific environment exists before creating or updating it
• Monitor environment lifecycle status (finalized, expired)
• Build dashboards showing environment freshness and model versions
• Implement cleanup logic that identifies expired environments
• Debug environment state issues by inspecting snapshot composition
• Audit environment usage patterns and creation history
• Validate environment readiness before running queries

Code Reference

Source Location

• Repository: sqlmesh
• File: sqlmesh/core/state_sync/base.py:L146-170

Signature

# Abstract base class methods - implemented by concrete state sync classes

@abc.abstractmethod
def get_environment(self, environment: str) -> t.Optional[Environment]:
    """Fetches the environment if it exists.

    Args:
        environment: The environment

    Returns:
        The environment object.
    """

@abc.abstractmethod
def get_environments(self) -> t.List[Environment]:
    """Fetches all environments.

    Returns:
        A list of all environments.
    """

@abc.abstractmethod
def get_environments_summary(self) -> t.List[EnvironmentSummary]:
    """Fetches all environment names along with expiry datetime.

    Returns:
        A list of all environment summaries.
    """

Import

from sqlmesh.core.context import Context

# StateReader is accessed through Context
context = Context()
state_reader = context.state_reader

I/O Contract

Inputs

For get_environment():

Name	Type	Required	Description
environment	str	Yes	Name of the environment to retrieve

For get_environments():

Name	Type	Required	Description
(no parameters)	-	-	Retrieves all environments

For get_environments_summary():

Name	Type	Required	Description
(no parameters)	-	-	Retrieves summary information for all environments

Outputs

For get_environment():

Name	Type	Description
return	Optional[Environment]	Environment object if found, None otherwise. Contains snapshots, promoted_snapshot_ids, timestamps, naming configuration

For get_environments():

Name	Type	Description
return	List[Environment]	List of all environment objects with complete metadata

For get_environments_summary():

Name	Type	Description
return	List[EnvironmentSummary]	List of lightweight environment summaries containing name, timestamps, plan_id, but not full snapshot details

Usage Examples

Basic Usage

from sqlmesh.core.context import Context

context = Context()

# Get all environments
environments = context.state_reader.get_environments()

for env in environments:
    print(f"Environment: {env.name}")
    print(f"  Finalized: {env.finalized_ts}")
    print(f"  Expires: {env.expiration_ts}")
    print(f"  Snapshots: {len(env.snapshots)}")

Check Environment Exists

from sqlmesh.core.context import Context

context = Context()

# Check if specific environment exists
env = context.state_reader.get_environment("staging")

if env:
    print(f"Staging environment exists with {len(env.snapshots)} models")
    print(f"Finalized at: {env.finalized_ts}")
else:
    print("Staging environment does not exist")

List Active Environments

from sqlmesh.core.context import Context
from sqlmesh.utils.date import now_timestamp

context = Context()

# Get lightweight summaries for listing
summaries = context.state_reader.get_environments_summary()

current_time = now_timestamp()
active_envs = [
    s for s in summaries
    if s.finalized_ts and (not s.expiration_ts or s.expiration_ts > current_time)
]

print(f"Active environments: {len(active_envs)}")
for summary in active_envs:
    print(f"  - {summary.name} (plan: {summary.plan_id})")

Inspect Environment Composition

from sqlmesh.core.context import Context

context = Context()

# Get detailed environment to inspect snapshots
env = context.state_reader.get_environment("prod")

if env:
    print(f"Production environment contains {len(env.snapshots)} snapshots:")

    for snapshot in env.snapshots:
        status = "promoted" if snapshot.snapshot_id in (env.promoted_snapshot_ids or []) else "unpromoted"
        print(f"  {snapshot.name} v{snapshot.version} [{status}]")
        print(f"    Physical table: {snapshot.table_name}")

Monitor Environment Freshness

from sqlmesh.core.context import Context
from datetime import datetime

context = Context()

environments = context.state_reader.get_environments()

for env in environments:
    if env.finalized_ts:
        finalized_dt = datetime.fromtimestamp(env.finalized_ts / 1000)
        age_hours = (datetime.now() - finalized_dt).total_seconds() / 3600

        status = "fresh" if age_hours < 24 else "stale"
        print(f"{env.name}: {status} (finalized {age_hours:.1f}h ago)")
    else:
        print(f"{env.name}: not finalized (in progress)")

Find Expired Environments

from sqlmesh.core.context import Context
from sqlmesh.utils.date import now_timestamp

context = Context()

summaries = context.state_reader.get_environments_summary()
current_time = now_timestamp()

expired = [
    s for s in summaries
    if s.expiration_ts and s.expiration_ts <= current_time
]

print(f"Found {len(expired)} expired environments:")
for summary in expired:
    print(f"  - {summary.name} (expired at {summary.expiration_ts})")

Compare Environment Snapshots

from sqlmesh.core.context import Context

context = Context()

# Compare two environments
prod_env = context.state_reader.get_environment("prod")
dev_env = context.state_reader.get_environment("dev")

if prod_env and dev_env:
    prod_snapshots = {s.name: s.version for s in prod_env.snapshots}
    dev_snapshots = {s.name: s.version for s in dev_env.snapshots}

    # Find differences
    for model_name in set(prod_snapshots.keys()) | set(dev_snapshots.keys()):
        prod_ver = prod_snapshots.get(model_name, "N/A")
        dev_ver = dev_snapshots.get(model_name, "N/A")

        if prod_ver != dev_ver:
            print(f"{model_name}: prod={prod_ver}, dev={dev_ver}")

Related Pages

Implements Principle

• Principle:TobikoData_Sqlmesh_Environment_Listing_And_Inspection