1"""
   2# Context
   3
   4A SQLMesh context encapsulates a SQLMesh environment. When you create a new context, it will discover and
   5load your project's models, macros, and audits. Afterwards, you can use the context to create and apply
   6plans, visualize your model's lineage, run your audits and model tests, and perform various other tasks.
   7For more information regarding what a context can do, see `sqlmesh.core.context.Context`.
   8
   9# Examples:
  10
  11Creating and applying a plan against the staging environment.
  12```python
  13from sqlmesh.core.context import Context
  14context = Context(paths="example", config="local_config")
  15plan = context.plan("staging")
  16context.apply(plan)
  17```
  18
  19Running audits on your data.
  20```python
  21from sqlmesh.core.context import Context
  22context = Context(paths="example", config="local_config")
  23context.audit("yesterday", "now")
  24```
  25
  26Running tests on your models.
  27```python
  28from sqlmesh.core.context import Context
  29context = Context(paths="example")
  30context.test()
  31```
  32"""
  33
  34from __future__ import annotations
  35
  36import abc
  37import collections
  38import logging
  39import sys
  40import time
  41import traceback
  42import typing as t
  43from functools import cached_property
  44from io import StringIO
  45from itertools import chain
  46from pathlib import Path
  47from shutil import rmtree
  48from types import MappingProxyType
  49from datetime import datetime
  50
  51from sqlglot import Dialect, exp
  52from sqlglot.helper import first
  53from sqlglot.lineage import GraphHTML
  54
  55from sqlmesh.core import analytics
  56from sqlmesh.core import constants as c
  57from sqlmesh.core.analytics import python_api_analytics
  58from sqlmesh.core.audit import Audit, ModelAudit, StandaloneAudit
  59from sqlmesh.core.config import (
  60    CategorizerConfig,
  61    Config,
  62    load_configs,
  63)
  64from sqlmesh.core.config.connection import ConnectionConfig
  65from sqlmesh.core.config.loader import C
  66from sqlmesh.core.config.root import RegexKeyDict
  67from sqlmesh.core.console import get_console
  68from sqlmesh.core.context_diff import ContextDiff
  69from sqlmesh.core.dialect import (
  70    format_model_expressions,
  71    is_meta_expression,
  72    normalize_model_name,
  73    pandas_to_sql,
  74    parse,
  75    parse_one,
  76)
  77from sqlmesh.core.engine_adapter import EngineAdapter
  78from sqlmesh.core.environment import Environment, EnvironmentNamingInfo, EnvironmentStatements
  79from sqlmesh.core.loader import Loader
  80from sqlmesh.core.linter.definition import AnnotatedRuleViolation, Linter
  81from sqlmesh.core.linter.rules import BUILTIN_RULES
  82from sqlmesh.core.macros import ExecutableOrMacro, macro
  83from sqlmesh.core.metric import Metric, rewrite
  84from sqlmesh.core.model import Model, update_model_schemas
  85from sqlmesh.core.config.model import ModelDefaultsConfig
  86from sqlmesh.core.notification_target import (
  87    NotificationEvent,
  88    NotificationTarget,
  89    NotificationTargetManager,
  90)
  91from sqlmesh.core.plan import Plan, PlanBuilder, SnapshotIntervals, PlanExplainer
  92from sqlmesh.core.plan.definition import UserProvidedFlags
  93from sqlmesh.core.reference import ReferenceGraph
  94from sqlmesh.core.scheduler import Scheduler, CompletionStatus
  95from sqlmesh.core.schema_loader import create_external_models_file
  96from sqlmesh.core.selector import Selector, NativeSelector
  97from sqlmesh.core.snapshot import (
  98    DeployabilityIndex,
  99    Snapshot,
 100    SnapshotEvaluator,
 101    SnapshotFingerprint,
 102    missing_intervals,
 103    to_table_mapping,
 104)
 105from sqlmesh.core.snapshot.definition import get_next_model_interval_start
 106from sqlmesh.core.state_sync import (
 107    CachingStateSync,
 108    StateReader,
 109    StateSync,
 110)
 111from sqlmesh.core.janitor import cleanup_expired_views, delete_expired_snapshots
 112from sqlmesh.core.table_diff import TableDiff
 113from sqlmesh.core.test import (
 114    ModelTextTestResult,
 115    ModelTestMetadata,
 116    generate_test,
 117    run_tests,
 118    filter_tests_by_patterns,
 119)
 120from sqlmesh.core.user import User
 121from sqlmesh.utils import UniqueKeyDict, Verbosity
 122from sqlmesh.utils.concurrency import concurrent_apply_to_values
 123from sqlmesh.utils.dag import DAG
 124from sqlmesh.utils.date import (
 125    TimeLike,
 126    to_timestamp,
 127    format_tz_datetime,
 128    now_timestamp,
 129    now,
 130    to_datetime,
 131    make_exclusive,
 132)
 133from sqlmesh.utils.errors import (
 134    CircuitBreakerError,
 135    ConfigError,
 136    PlanError,
 137    SQLMeshError,
 138    UncategorizedPlanError,
 139    LinterError,
 140)
 141from sqlmesh.utils.config import print_config
 142from sqlmesh.utils.jinja import JinjaMacroRegistry
 143from sqlmesh.utils.windows import IS_WINDOWS, fix_windows_path
 144
 145if t.TYPE_CHECKING:
 146    import pandas as pd
 147    from typing_extensions import Literal
 148
 149    from sqlmesh.core.engine_adapter._typing import (
 150        BigframeSession,
 151        DF,
 152        PySparkDataFrame,
 153        PySparkSession,
 154        SnowparkSession,
 155    )
 156    from sqlmesh.core.snapshot import Node
 157
 158    from sqlmesh.core.snapshot.definition import Intervals
 159
 160    ModelOrSnapshot = t.Union[str, Model, Snapshot]
 161    NodeOrSnapshot = t.Union[str, Model, StandaloneAudit, Snapshot]
 162
 163logger = logging.getLogger(__name__)
 164
 165
 166class BaseContext(abc.ABC):
 167    """The base context which defines methods to execute a model."""
 168
 169    @property
 170    @abc.abstractmethod
 171    def default_dialect(self) -> t.Optional[str]:
 172        """Returns the default dialect."""
 173
 174    @property
 175    @abc.abstractmethod
 176    def _model_tables(self) -> t.Dict[str, str]:
 177        """Returns a mapping of model names to tables."""
 178
 179    @property
 180    @abc.abstractmethod
 181    def engine_adapter(self) -> EngineAdapter:
 182        """Returns an engine adapter."""
 183
 184    @property
 185    def spark(self) -> t.Optional[PySparkSession]:
 186        """Returns the spark session if it exists."""
 187        return self.engine_adapter.spark
 188
 189    @property
 190    def snowpark(self) -> t.Optional[SnowparkSession]:
 191        """Returns the snowpark session if it exists."""
 192        return self.engine_adapter.snowpark
 193
 194    @property
 195    def bigframe(self) -> t.Optional[BigframeSession]:
 196        """Returns the bigframe session if it exists."""
 197        return self.engine_adapter.bigframe
 198
 199    @property
 200    def default_catalog(self) -> t.Optional[str]:
 201        raise NotImplementedError
 202
 203    def table(self, model_name: str) -> str:
 204        get_console().log_warning(
 205            "The SQLMesh context's `table` method is deprecated and will be removed "
 206            "in a future release. Please use the `resolve_table` method instead."
 207        )
 208        return self.resolve_table(model_name)
 209
 210    def resolve_table(self, model_name: str) -> str:
 211        """Gets the physical table name for a given model.
 212
 213        Args:
 214            model_name: The model name.
 215
 216        Returns:
 217            The physical table name.
 218        """
 219        model_name = normalize_model_name(model_name, self.default_catalog, self.default_dialect)
 220
 221        if model_name not in self._model_tables:
 222            model_name_list = "\n".join(list(self._model_tables))
 223            logger.debug(
 224                f"'{model_name}' not found in model to table mapping. Available model names: \n{model_name_list}"
 225            )
 226            raise SQLMeshError(
 227                f"Unable to find a table mapping for model '{model_name}'. Has it been spelled correctly?"
 228            )
 229
 230        # We generate SQL for the default dialect because the table name may be used in a
 231        # fetchdf call and so the quotes need to be correct (eg. backticks for bigquery)
 232        return parse_one(self._model_tables[model_name]).sql(
 233            dialect=self.default_dialect, identify=True
 234        )
 235
 236    def fetchdf(
 237        self, query: t.Union[exp.Expr, str], quote_identifiers: bool = False
 238    ) -> pd.DataFrame:
 239        """Fetches a dataframe given a sql string or sqlglot expression.
 240
 241        Args:
 242            query: SQL string or sqlglot expression.
 243            quote_identifiers: Whether to quote all identifiers in the query.
 244
 245        Returns:
 246            The default dataframe is Pandas, but for Spark a PySpark dataframe is returned.
 247        """
 248        return self.engine_adapter.fetchdf(query, quote_identifiers=quote_identifiers)
 249
 250    def fetch_pyspark_df(
 251        self, query: t.Union[exp.Expr, str], quote_identifiers: bool = False
 252    ) -> PySparkDataFrame:
 253        """Fetches a PySpark dataframe given a sql string or sqlglot expression.
 254
 255        Args:
 256            query: SQL string or sqlglot expression.
 257            quote_identifiers: Whether to quote all identifiers in the query.
 258
 259        Returns:
 260            A PySpark dataframe.
 261        """
 262        return self.engine_adapter.fetch_pyspark_df(query, quote_identifiers=quote_identifiers)
 263
 264
 265class ExecutionContext(BaseContext):
 266    """The minimal context needed to execute a model.
 267
 268    Args:
 269        engine_adapter: The engine adapter to execute queries against.
 270        snapshots: All upstream snapshots (by model name) to use for expansion and mapping of physical locations.
 271        deployability_index: Determines snapshots that are deployable in the context of this evaluation.
 272    """
 273
 274    def __init__(
 275        self,
 276        engine_adapter: EngineAdapter,
 277        snapshots: t.Dict[str, Snapshot],
 278        deployability_index: t.Optional[DeployabilityIndex] = None,
 279        default_dialect: t.Optional[str] = None,
 280        default_catalog: t.Optional[str] = None,
 281        is_restatement: t.Optional[bool] = None,
 282        parent_intervals: t.Optional[Intervals] = None,
 283        variables: t.Optional[t.Dict[str, t.Any]] = None,
 284        blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
 285    ):
 286        self.snapshots = snapshots
 287        self.deployability_index = deployability_index
 288        self._engine_adapter = engine_adapter
 289        self._default_catalog = default_catalog
 290        self._default_dialect = default_dialect
 291        self._variables = variables or {}
 292        self._blueprint_variables = blueprint_variables or {}
 293        self._is_restatement = is_restatement
 294        self._parent_intervals = parent_intervals
 295
 296    @property
 297    def default_dialect(self) -> t.Optional[str]:
 298        return self._default_dialect
 299
 300    @property
 301    def engine_adapter(self) -> EngineAdapter:
 302        """Returns an engine adapter."""
 303        return self._engine_adapter
 304
 305    @cached_property
 306    def _model_tables(self) -> t.Dict[str, str]:
 307        """Returns a mapping of model names to tables."""
 308        return to_table_mapping(self.snapshots.values(), self.deployability_index)
 309
 310    @property
 311    def default_catalog(self) -> t.Optional[str]:
 312        return self._default_catalog
 313
 314    @property
 315    def gateway(self) -> t.Optional[str]:
 316        """Returns the gateway name."""
 317        return self.var(c.GATEWAY)
 318
 319    @property
 320    def is_restatement(self) -> t.Optional[bool]:
 321        return self._is_restatement
 322
 323    @property
 324    def parent_intervals(self) -> t.Optional[Intervals]:
 325        return self._parent_intervals
 326
 327    def var(self, var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
 328        """Returns a variable value."""
 329        return self._variables.get(var_name.lower(), default)
 330
 331    def blueprint_var(self, var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
 332        """Returns a blueprint variable value."""
 333        return self._blueprint_variables.get(var_name.lower(), default)
 334
 335    def with_variables(
 336        self,
 337        variables: t.Dict[str, t.Any],
 338        blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
 339    ) -> ExecutionContext:
 340        """Returns a new ExecutionContext with additional variables."""
 341        return ExecutionContext(
 342            self._engine_adapter,
 343            self.snapshots,
 344            self.deployability_index,
 345            self._default_dialect,
 346            self._default_catalog,
 347            self._is_restatement,
 348            variables=variables,
 349            blueprint_variables=blueprint_variables,
 350        )
 351
 352
 353class GenericContext(BaseContext, t.Generic[C]):
 354    """Encapsulates a SQLMesh environment supplying convenient functions to perform various tasks.
 355
 356    Args:
 357        notification_targets: The notification target to use. Defaults to what is defined in config.
 358        paths: The directories containing SQLMesh files.
 359        config: A Config object or the name of a Config object in config.py.
 360        connection: The name of the connection. If not specified the first connection as it appears
 361            in configuration will be used.
 362        test_connection: The name of the connection to use for tests. If not specified the first
 363            connection as it appears in configuration will be used.
 364        concurrent_tasks: The maximum number of tasks that can use the connection concurrently.
 365        load: Whether or not to automatically load all models and macros (default True).
 366        load_state: Whether to merge remote state into the local project during load (default True).
 367            Only intended for local-only operations like format; plan/apply in multi-repo projects
 368            require it to see models owned by other projects.
 369        console: The rich instance used for printing out CLI command results.
 370        users: A list of users to make known to SQLMesh.
 371    """
 372
 373    CONFIG_TYPE: t.Type[C]
 374    """The type of config object to use (default: Config)."""
 375
 376    PLAN_BUILDER_TYPE = PlanBuilder
 377    """The type of plan builder object to use (default: PlanBuilder)."""
 378
 379    def __init__(
 380        self,
 381        notification_targets: t.Optional[t.List[NotificationTarget]] = None,
 382        state_sync: t.Optional[StateSync] = None,
 383        paths: t.Union[str | Path, t.Iterable[str | Path]] = "",
 384        config: t.Optional[t.Union[C, str, t.Dict[Path, C]]] = None,
 385        gateway: t.Optional[str] = None,
 386        concurrent_tasks: t.Optional[int] = None,
 387        loader: t.Optional[t.Type[Loader]] = None,
 388        load: bool = True,
 389        users: t.Optional[t.List[User]] = None,
 390        config_loader_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
 391        selector: t.Optional[t.Type[Selector]] = None,
 392        load_state: bool = True,
 393    ):
 394        self.configs = (
 395            config
 396            if isinstance(config, dict)
 397            else load_configs(config, self.CONFIG_TYPE, paths, **(config_loader_kwargs or {}))
 398        )
 399        self._projects = {config.project for config in self.configs.values()}
 400        self.dag: DAG[str] = DAG()
 401        self._models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
 402        self._audits: UniqueKeyDict[str, ModelAudit] = UniqueKeyDict("audits")
 403        self._standalone_audits: UniqueKeyDict[str, StandaloneAudit] = UniqueKeyDict(
 404            "standaloneaudits"
 405        )
 406        self._model_test_metadata: t.List[ModelTestMetadata] = []
 407        self._model_test_metadata_path_index: t.Dict[Path, t.List[ModelTestMetadata]] = {}
 408        self._model_test_metadata_fully_qualified_name_index: t.Dict[str, ModelTestMetadata] = {}
 409        self._models_with_tests: t.Set[str] = set()
 410
 411        self._macros: UniqueKeyDict[str, ExecutableOrMacro] = UniqueKeyDict("macros")
 412        self._metrics: UniqueKeyDict[str, Metric] = UniqueKeyDict("metrics")
 413        self._jinja_macros = JinjaMacroRegistry()
 414        self._requirements: t.Dict[str, str] = {}
 415        self._environment_statements: t.List[EnvironmentStatements] = []
 416        self._excluded_requirements: t.Set[str] = set()
 417        self._engine_adapter: t.Optional[EngineAdapter] = None
 418        self._linters: t.Dict[str, Linter] = {}
 419        self._loaded: bool = False
 420        self._load_state: bool = load_state
 421        self._selector_cls = selector or NativeSelector
 422
 423        self.path, self.config = t.cast(t.Tuple[Path, C], next(iter(self.configs.items())))
 424
 425        self._all_dialects: t.Set[str] = {self.config.dialect or ""}
 426
 427        if self.config.disable_anonymized_analytics:
 428            analytics.disable_analytics()
 429
 430        self.gateway = gateway
 431        self._scheduler = self.config.get_scheduler(self.gateway)
 432        self.environment_ttl = self.config.environment_ttl
 433        self.pinned_environments = Environment.sanitize_names(self.config.pinned_environments)
 434        self.auto_categorize_changes = self.config.plan.auto_categorize_changes
 435        self.selected_gateway = (gateway or self.config.default_gateway_name).lower()
 436
 437        gw_model_defaults = self.config.get_gateway(self.selected_gateway).model_defaults
 438        if gw_model_defaults:
 439            # Merge global model defaults with the selected gateway's, if it's overriden
 440            global_defaults = self.config.model_defaults.model_dump(exclude_unset=True)
 441            gateway_defaults = gw_model_defaults.model_dump(exclude_unset=True)
 442
 443            self.config.model_defaults = ModelDefaultsConfig(
 444                **{**global_defaults, **gateway_defaults}
 445            )
 446
 447        # This allows overriding the default dialect's normalization strategy, so for example
 448        # one can do `dialect="duckdb,normalization_strategy=lowercase"` and this will be
 449        # applied to the DuckDB dialect globally
 450        if "normalization_strategy" in str(self.config.dialect):
 451            dialect = Dialect.get_or_raise(self.config.dialect)
 452            type(dialect).NORMALIZATION_STRATEGY = dialect.normalization_strategy
 453
 454        self._loaders = [
 455            (loader or config.loader)(self, path, **config.loader_kwargs)
 456            for path, config in self.configs.items()
 457        ]
 458
 459        self._concurrent_tasks = concurrent_tasks
 460        self._state_connection_config = (
 461            self.config.get_state_connection(self.gateway) or self.connection_config
 462        )
 463
 464        self._snapshot_evaluator: t.Optional[SnapshotEvaluator] = None
 465
 466        self.console = get_console()
 467        setattr(self.console, "dialect", self.config.dialect)
 468
 469        self._provided_state_sync: t.Optional[StateSync] = state_sync
 470        self._state_sync: t.Optional[StateSync] = None
 471
 472        # Should we dedupe notification_targets? If so how?
 473        self.notification_targets = (notification_targets or []) + self.config.notification_targets
 474        self.users = (users or []) + self.config.users
 475        self.users = list({user.username: user for user in self.users}.values())
 476        self._register_notification_targets()
 477
 478        if load:
 479            self.load()
 480
 481    @property
 482    def default_dialect(self) -> t.Optional[str]:
 483        return self.config.dialect
 484
 485    @property
 486    def engine_adapter(self) -> EngineAdapter:
 487        """Returns the default engine adapter."""
 488        if self._engine_adapter is None:
 489            self._engine_adapter = self.connection_config.create_engine_adapter()
 490        return self._engine_adapter
 491
 492    @property
 493    def snapshot_evaluator(self) -> SnapshotEvaluator:
 494        if not self._snapshot_evaluator:
 495            self._ensure_virtual_catalog_injection()
 496            self._snapshot_evaluator = SnapshotEvaluator(
 497                {
 498                    gateway: adapter.with_settings(execute_log_level=logging.INFO)
 499                    for gateway, adapter in self.engine_adapters.items()
 500                },
 501                ddl_concurrent_tasks=self.concurrent_tasks,
 502                selected_gateway=self.selected_gateway,
 503            )
 504        return self._snapshot_evaluator
 505
 506    def _ensure_virtual_catalog_injection(self) -> None:
 507        """Ensure virtual catalog injection has run before adapters are cloned for SnapshotEvaluator.
 508
 509        Injection is a side effect of get_default_catalog_per_gateway. In normal usage it fires
 510        earlier (default_catalog is accessed during model loading), but this guard covers the edge
 511        case where snapshot_evaluator is accessed directly on a fresh context before any model ops.
 512        """
 513        _ = self.default_catalog_per_gateway
 514
 515    def execution_context(
 516        self,
 517        deployability_index: t.Optional[DeployabilityIndex] = None,
 518        engine_adapter: t.Optional[EngineAdapter] = None,
 519        snapshots: t.Optional[t.Dict[str, Snapshot]] = None,
 520    ) -> ExecutionContext:
 521        """Returns an execution context."""
 522        return ExecutionContext(
 523            engine_adapter=engine_adapter or self.engine_adapter,
 524            snapshots=snapshots or self.snapshots,
 525            deployability_index=deployability_index,
 526            default_dialect=self.default_dialect,
 527            default_catalog=self.default_catalog,
 528        )
 529
 530    @python_api_analytics
 531    def upsert_model(self, model: t.Union[str, Model], **kwargs: t.Any) -> Model:
 532        """Update or insert a model.
 533
 534        The context's models dictionary will be updated to include these changes.
 535
 536        Args:
 537            model: Model name or instance to update.
 538            kwargs: The kwargs to update the model with.
 539
 540        Returns:
 541            A new instance of the updated or inserted model.
 542        """
 543        model = self.get_model(model, raise_if_missing=True)
 544        if not model.enabled:
 545            raise SQLMeshError(f"The disabled model '{model.name}' cannot be upserted")
 546        path = model._path
 547
 548        model = model.copy(update=kwargs)
 549        model._path = path
 550
 551        self.dag.add(model.fqn, model.depends_on)
 552
 553        self._models.update(
 554            {
 555                model.fqn: model,
 556                # bust the fingerprint cache for all downstream models
 557                **{fqn: self._models[fqn].copy() for fqn in self.dag.downstream(model.fqn)},
 558            }
 559        )
 560
 561        update_model_schemas(
 562            self.dag,
 563            models=self._models,
 564            cache_dir=self.cache_dir,
 565        )
 566
 567        if model.dialect:
 568            self._all_dialects.add(model.dialect)
 569
 570        model.validate_definition()
 571
 572        return model
 573
 574    def scheduler(
 575        self,
 576        environment: t.Optional[str] = None,
 577        snapshot_evaluator: t.Optional[SnapshotEvaluator] = None,
 578    ) -> Scheduler:
 579        """Returns the built-in scheduler.
 580
 581        Args:
 582            environment: The target environment to source model snapshots from, or None
 583                if snapshots should be sourced from the currently loaded local state.
 584
 585        Returns:
 586            The built-in scheduler instance.
 587        """
 588        snapshots: t.Iterable[Snapshot]
 589        if environment is not None:
 590            stored_environment = self.state_sync.get_environment(environment)
 591            if stored_environment is None:
 592                raise ConfigError(f"Environment '{environment}' was not found.")
 593            snapshots = self.state_sync.get_snapshots(stored_environment.snapshots).values()
 594        else:
 595            snapshots = self.snapshots.values()
 596
 597        if not snapshots:
 598            raise ConfigError("No models were found")
 599
 600        return self.create_scheduler(snapshots, snapshot_evaluator or self.snapshot_evaluator)
 601
 602    def create_scheduler(
 603        self, snapshots: t.Iterable[Snapshot], snapshot_evaluator: SnapshotEvaluator
 604    ) -> Scheduler:
 605        """Creates the built-in scheduler.
 606
 607        Args:
 608            snapshots: The snapshots to schedule.
 609
 610        Returns:
 611            The built-in scheduler instance.
 612        """
 613        return Scheduler(
 614            snapshots,
 615            snapshot_evaluator,
 616            self.state_sync,
 617            default_catalog=self.default_catalog,
 618            max_workers=self.concurrent_tasks,
 619            console=self.console,
 620            notification_target_manager=self.notification_target_manager,
 621        )
 622
 623    @property
 624    def state_sync(self) -> StateSync:
 625        if not self._state_sync:
 626            self._state_sync = self._new_state_sync()
 627
 628            if self._state_sync.get_versions(validate=False).schema_version == 0:
 629                self.console.log_status_update("Initializing new project state...")
 630                self._state_sync.migrate()
 631            self._state_sync.get_versions()
 632            self._state_sync = CachingStateSync(self._state_sync)  # type: ignore
 633        return self._state_sync
 634
 635    @property
 636    def state_reader(self) -> StateReader:
 637        return self.state_sync
 638
 639    def refresh(self) -> None:
 640        """Refresh all models that have been updated."""
 641        if any(loader.reload_needed() for loader in self._loaders):
 642            self.load()
 643
 644    def load(self, update_schemas: bool = True) -> GenericContext[C]:
 645        """Load all files in the context's path."""
 646        load_start_ts = time.perf_counter()
 647
 648        loaded_projects = [loader.load() for loader in self._loaders]
 649
 650        self.dag = DAG()
 651        self._standalone_audits.clear()
 652        self._audits.clear()
 653        self._macros.clear()
 654        self._models.clear()
 655        self._metrics.clear()
 656        self._requirements.clear()
 657        self._excluded_requirements.clear()
 658        self._linters.clear()
 659        self._environment_statements = []
 660        self._model_test_metadata.clear()
 661        self._model_test_metadata_path_index.clear()
 662        self._model_test_metadata_fully_qualified_name_index.clear()
 663        self._models_with_tests.clear()
 664
 665        for loader, project in zip(self._loaders, loaded_projects):
 666            self._jinja_macros = self._jinja_macros.merge(project.jinja_macros)
 667            self._macros.update(project.macros)
 668            self._models.update(project.models)
 669            self._metrics.update(project.metrics)
 670            self._audits.update(project.audits)
 671            self._standalone_audits.update(project.standalone_audits)
 672            self._requirements.update(project.requirements)
 673            self._excluded_requirements.update(project.excluded_requirements)
 674            self._environment_statements.extend(project.environment_statements)
 675
 676            self._model_test_metadata.extend(project.model_test_metadata)
 677            for metadata in project.model_test_metadata:
 678                if metadata.path not in self._model_test_metadata_path_index:
 679                    self._model_test_metadata_path_index[metadata.path] = []
 680                self._model_test_metadata_path_index[metadata.path].append(metadata)
 681                self._model_test_metadata_fully_qualified_name_index[
 682                    metadata.fully_qualified_test_name
 683                ] = metadata
 684                self._models_with_tests.add(metadata.model_name)
 685
 686            config = loader.config
 687            self._linters[config.project] = Linter.from_rules(
 688                BUILTIN_RULES.union(project.user_rules), config.linter
 689            )
 690
 691        # Load environment statements from state for projects not in current load
 692        if self._load_state and any(self._projects):
 693            prod = self.state_reader.get_environment(c.PROD)
 694            if prod:
 695                existing_statements = self.state_reader.get_environment_statements(c.PROD)
 696                for stmt in existing_statements:
 697                    if stmt.project and stmt.project not in self._projects:
 698                        self._environment_statements.append(stmt)
 699
 700        uncached = set()
 701
 702        if self._load_state and any(self._projects):
 703            prod = self.state_reader.get_environment(c.PROD)
 704
 705            if prod:
 706                for snapshot in self.state_reader.get_snapshots(prod.snapshots).values():
 707                    if snapshot.node.project in self._projects:
 708                        uncached.add(snapshot.name)
 709                    else:
 710                        local_store = self._standalone_audits if snapshot.is_audit else self._models
 711                        if snapshot.name in local_store:
 712                            uncached.add(snapshot.name)
 713                        else:
 714                            local_store[snapshot.name] = snapshot.node  # type: ignore
 715
 716        for model in self._models.values():
 717            self.dag.add(model.fqn, model.depends_on)
 718
 719        if update_schemas:
 720            for fqn in self.dag:
 721                model = self._models.get(fqn)  # type: ignore
 722
 723                if not model or fqn in uncached:
 724                    continue
 725
 726                # make a copy of remote models that depend on local models or in the downstream chain
 727                # without this, a SELECT * FROM local will not propogate properly because the downstream
 728                # model will get mutated (schema changes) but the object is the same as the remote cache
 729                if any(dep in uncached for dep in model.depends_on):
 730                    uncached.add(fqn)
 731                    self._models.update({fqn: model.copy(update={"mapping_schema": {}})})
 732                    continue
 733
 734            update_model_schemas(
 735                self.dag,
 736                models=self._models,
 737                cache_dir=self.cache_dir,
 738            )
 739
 740            models = self.models.values()
 741            for model in models:
 742                # The model definition can be validated correctly only after the schema is set.
 743                model.validate_definition()
 744
 745        duplicates = set(self._models) & set(self._standalone_audits)
 746        if duplicates:
 747            raise ConfigError(
 748                f"Models and Standalone audits cannot have the same name: {duplicates}"
 749            )
 750
 751        self._all_dialects = {m.dialect for m in self._models.values() if m.dialect} | {
 752            self.default_dialect or ""
 753        }
 754
 755        analytics.collector.on_project_loaded(
 756            project_type=self._project_type,
 757            models_count=len(self._models),
 758            audits_count=len(self._audits),
 759            standalone_audits_count=len(self._standalone_audits),
 760            macros_count=len(self._macros),
 761            jinja_macros_count=len(self._jinja_macros.root_macros),
 762            load_time_sec=time.perf_counter() - load_start_ts,
 763            state_sync_fingerprint=self._scheduler.state_sync_fingerprint(self),
 764            project_name=self.config.project,
 765        )
 766
 767        self._loaded = True
 768        return self
 769
 770    @python_api_analytics
 771    def run(
 772        self,
 773        environment: t.Optional[str] = None,
 774        *,
 775        start: t.Optional[TimeLike] = None,
 776        end: t.Optional[TimeLike] = None,
 777        execution_time: t.Optional[TimeLike] = None,
 778        skip_janitor: bool = False,
 779        ignore_cron: bool = False,
 780        select_models: t.Optional[t.Collection[str]] = None,
 781        exit_on_env_update: t.Optional[int] = None,
 782        no_auto_upstream: bool = False,
 783    ) -> CompletionStatus:
 784        """Run the entire dag through the scheduler.
 785
 786        Args:
 787            environment: The target environment to source model snapshots from and virtually update. Default: prod.
 788            start: The start of the interval to render.
 789            end: The end of the interval to render.
 790            execution_time: The date/time time reference to use for execution time. Defaults to now.
 791            skip_janitor: Whether to skip the janitor task.
 792            ignore_cron: Whether to ignore the model's cron schedule and run all available missing intervals.
 793            select_models: A list of model selection expressions to filter models that should run. Note that
 794                upstream dependencies of selected models will also be evaluated.
 795            exit_on_env_update: If set, exits with the provided code if the run is interrupted by an update
 796                to the target environment.
 797            no_auto_upstream: Whether to not force upstream models to run. Only applicable when using `select_models`.
 798
 799        Returns:
 800            True if the run was successful, False otherwise.
 801        """
 802        environment = environment or self.config.default_target_environment
 803        environment = Environment.sanitize_name(environment)
 804        if not skip_janitor and environment.lower() == c.PROD:
 805            self._run_janitor()
 806
 807        self.notification_target_manager.notify(
 808            NotificationEvent.RUN_START, environment=environment
 809        )
 810        analytics_run_id = analytics.collector.on_run_start(
 811            engine_type=self.snapshot_evaluator.adapter.dialect,
 812            state_sync_type=self.state_sync.state_type(),
 813        )
 814        self._load_materializations()
 815
 816        env_check_attempts_num = max(
 817            1,
 818            self.config.run.environment_check_max_wait
 819            // self.config.run.environment_check_interval,
 820        )
 821
 822        def _block_until_finalized() -> str:
 823            for _ in range(env_check_attempts_num):
 824                assert environment is not None  # mypy
 825                environment_state = self.state_sync.get_environment(environment)
 826                if not environment_state:
 827                    raise SQLMeshError(f"Environment '{environment}' was not found.")
 828                if environment_state.finalized_ts:
 829                    return environment_state.plan_id
 830                self.console.log_warning(
 831                    f"Environment '{environment}' is being updated by plan '{environment_state.plan_id}'. "
 832                    f"Retrying in {self.config.run.environment_check_interval} seconds..."
 833                )
 834                time.sleep(self.config.run.environment_check_interval)
 835            raise SQLMeshError(
 836                f"Exceeded the maximum wait time for environment '{environment}' to be ready. "
 837                "This means that the environment either failed to update or the update is taking longer than expected. "
 838                "See https://sqlmesh.readthedocs.io/en/stable/reference/configuration/#run to adjust the timeout settings."
 839            )
 840
 841        success = False
 842        interrupted = False
 843        done = False
 844        while not done:
 845            plan_id_at_start = _block_until_finalized()
 846
 847            def _has_environment_changed() -> bool:
 848                assert environment is not None  # mypy
 849                current_environment_state = self.state_sync.get_environment(environment)
 850                return (
 851                    not current_environment_state
 852                    or current_environment_state.plan_id != plan_id_at_start
 853                    or not current_environment_state.finalized_ts
 854                )
 855
 856            try:
 857                completion_status = self._run(
 858                    environment,
 859                    start=start,
 860                    end=end,
 861                    execution_time=execution_time,
 862                    ignore_cron=ignore_cron,
 863                    select_models=select_models,
 864                    circuit_breaker=_has_environment_changed,
 865                    no_auto_upstream=no_auto_upstream,
 866                )
 867                done = True
 868            except CircuitBreakerError:
 869                self.console.log_warning(
 870                    f"Environment '{environment}' modified while running. Restarting the run..."
 871                )
 872                if exit_on_env_update:
 873                    interrupted = True
 874                    done = True
 875            except Exception as e:
 876                self.notification_target_manager.notify(
 877                    NotificationEvent.RUN_FAILURE, traceback.format_exc()
 878                )
 879                logger.info("Run failed.", exc_info=e)
 880                analytics.collector.on_run_end(
 881                    run_id=analytics_run_id, succeeded=False, interrupted=False, error=e
 882                )
 883                raise e
 884
 885        if completion_status.is_success or interrupted:
 886            self.notification_target_manager.notify(
 887                NotificationEvent.RUN_END, environment=environment
 888            )
 889            self.console.log_success(f"Run finished for environment '{environment}'")
 890        elif completion_status.is_failure:
 891            self.notification_target_manager.notify(
 892                NotificationEvent.RUN_FAILURE, "See console logs for details."
 893            )
 894
 895        analytics.collector.on_run_end(
 896            run_id=analytics_run_id, succeeded=success, interrupted=interrupted
 897        )
 898
 899        if interrupted and exit_on_env_update is not None:
 900            sys.exit(exit_on_env_update)
 901
 902        return completion_status
 903
 904    @python_api_analytics
 905    def run_janitor(
 906        self,
 907        ignore_ttl: bool,
 908        force_delete: bool = False,
 909        environment: t.Optional[str] = None,
 910    ) -> bool:
 911        if environment is not None:
 912            environment = Environment.sanitize_name(environment)
 913
 914        success = False
 915
 916        if self.console.start_cleanup(ignore_ttl):
 917            try:
 918                self._run_janitor(ignore_ttl, force_delete=force_delete, environment=environment)
 919                success = True
 920            finally:
 921                self.console.stop_cleanup(success=success)
 922
 923        return success
 924
 925    @python_api_analytics
 926    def destroy(self) -> bool:
 927        success = False
 928
 929        # Collect resources to be deleted
 930        environments = self.state_reader.get_environments()
 931        schemas_to_delete = set()
 932        tables_to_delete = set()
 933        views_to_delete = set()
 934        all_snapshot_infos = set()
 935
 936        # For each environment find schemas and tables
 937        for environment in environments:
 938            all_snapshot_infos.update(environment.snapshots)
 939            snapshots = self.state_reader.get_snapshots(environment.snapshots).values()
 940            for snapshot in snapshots:
 941                if snapshot.is_model and not snapshot.is_symbolic:
 942                    # Get the appropriate adapter
 943                    if environment.gateway_managed and snapshot.model_gateway:
 944                        adapter = self.engine_adapters.get(
 945                            snapshot.model_gateway, self.engine_adapter
 946                        )
 947                    else:
 948                        adapter = self.engine_adapter
 949
 950                    if environment.suffix_target.is_schema or environment.suffix_target.is_catalog:
 951                        schema = snapshot.qualified_view_name.schema_for_environment(
 952                            environment.naming_info, dialect=adapter.dialect
 953                        )
 954                        catalog = snapshot.qualified_view_name.catalog_for_environment(
 955                            environment.naming_info, dialect=adapter.dialect
 956                        )
 957                        if catalog:
 958                            schemas_to_delete.add(f"{catalog}.{schema}")
 959                        else:
 960                            schemas_to_delete.add(schema)
 961
 962                    if environment.suffix_target.is_table:
 963                        view_name = snapshot.qualified_view_name.for_environment(
 964                            environment.naming_info, dialect=adapter.dialect
 965                        )
 966                        views_to_delete.add(view_name)
 967
 968                    # Add snapshot tables
 969                    table_name = snapshot.table_name()
 970                    tables_to_delete.add(table_name)
 971
 972        if self.console.start_destroy(schemas_to_delete, views_to_delete, tables_to_delete):
 973            try:
 974                success = self._destroy()
 975            finally:
 976                self.console.stop_destroy(success=success)
 977
 978        return success
 979
 980    @t.overload
 981    def get_model(
 982        self, model_or_snapshot: ModelOrSnapshot, raise_if_missing: Literal[True] = True
 983    ) -> Model: ...
 984
 985    @t.overload
 986    def get_model(
 987        self,
 988        model_or_snapshot: ModelOrSnapshot,
 989        raise_if_missing: Literal[False] = False,
 990    ) -> t.Optional[Model]: ...
 991
 992    def get_model(
 993        self, model_or_snapshot: ModelOrSnapshot, raise_if_missing: bool = False
 994    ) -> t.Optional[Model]:
 995        """Returns a model with the given name or None if a model with such name doesn't exist.
 996
 997        Args:
 998            model_or_snapshot: A model name, model, or snapshot.
 999            raise_if_missing: Raises an error if a model is not found.
1000
1001        Returns:
1002            The expected model.
1003        """
1004        if isinstance(model_or_snapshot, Snapshot):
1005            return model_or_snapshot.model
1006        if not isinstance(model_or_snapshot, str):
1007            return model_or_snapshot
1008
1009        try:
1010            # We should try all dialects referenced in the project for cases when models use mixed dialects.
1011            for dialect in self._all_dialects:
1012                normalized_name = normalize_model_name(
1013                    model_or_snapshot,
1014                    dialect=dialect,
1015                    default_catalog=self.default_catalog,
1016                )
1017                if normalized_name in self._models:
1018                    return self._models[normalized_name]
1019        except:
1020            pass
1021
1022        if raise_if_missing:
1023            if model_or_snapshot.endswith((".sql", ".py")):
1024                msg = "Resolving models by path is not supported, please pass in the model name instead."
1025            else:
1026                msg = f"Cannot find model with name '{model_or_snapshot}'"
1027
1028            raise SQLMeshError(msg)
1029
1030        return None
1031
1032    @t.overload
1033    def get_snapshot(self, node_or_snapshot: NodeOrSnapshot) -> t.Optional[Snapshot]: ...
1034
1035    @t.overload
1036    def get_snapshot(
1037        self, node_or_snapshot: NodeOrSnapshot, raise_if_missing: Literal[True]
1038    ) -> Snapshot: ...
1039
1040    @t.overload
1041    def get_snapshot(
1042        self, node_or_snapshot: NodeOrSnapshot, raise_if_missing: Literal[False]
1043    ) -> t.Optional[Snapshot]: ...
1044
1045    def get_snapshot(
1046        self, node_or_snapshot: NodeOrSnapshot, raise_if_missing: bool = False
1047    ) -> t.Optional[Snapshot]:
1048        """Returns a snapshot with the given name or None if a snapshot with such name doesn't exist.
1049
1050        Args:
1051            node_or_snapshot: A node name, node, or snapshot.
1052            raise_if_missing: Raises an error if a snapshot is not found.
1053
1054        Returns:
1055            The expected snapshot.
1056        """
1057        if isinstance(node_or_snapshot, Snapshot):
1058            return node_or_snapshot
1059        fqn = self._node_or_snapshot_to_fqn(node_or_snapshot)
1060        snapshot = self.snapshots.get(fqn)
1061
1062        if raise_if_missing and not snapshot:
1063            raise SQLMeshError(f"Cannot find snapshot for '{fqn}'")
1064
1065        return snapshot
1066
1067    def config_for_path(self, path: Path) -> t.Tuple[Config, Path]:
1068        """Returns the config and path of the said project for a given file path."""
1069        for config_path, config in self.configs.items():
1070            try:
1071                path.relative_to(config_path)
1072                return config, config_path
1073            except ValueError:
1074                pass
1075        return self.config, self.path
1076
1077    def config_for_node(self, node: Model | Audit) -> Config:
1078        path = node._path
1079        if path is None:
1080            return self.config
1081        return self.config_for_path(path)[0]  # type: ignore
1082
1083    @property
1084    def models(self) -> MappingProxyType[str, Model]:
1085        """Returns all registered models in this context."""
1086        return MappingProxyType(self._models)
1087
1088    @property
1089    def metrics(self) -> MappingProxyType[str, Metric]:
1090        """Returns all registered metrics in this context."""
1091        return MappingProxyType(self._metrics)
1092
1093    @property
1094    def standalone_audits(self) -> MappingProxyType[str, StandaloneAudit]:
1095        """Returns all registered standalone audits in this context."""
1096        return MappingProxyType(self._standalone_audits)
1097
1098    @property
1099    def models_with_tests(self) -> t.Set[str]:
1100        """Returns all models with tests in this context."""
1101        return self._models_with_tests
1102
1103    @property
1104    def snapshots(self) -> t.Dict[str, Snapshot]:
1105        """Generates and returns snapshots based on models registered in this context.
1106
1107        If one of the snapshots has been previously stored in the persisted state, the stored
1108        instance will be returned.
1109        """
1110        return self._snapshots()
1111
1112    @property
1113    def requirements(self) -> t.Dict[str, str]:
1114        """Returns the Python dependencies of the project loaded in this context."""
1115        return self._requirements.copy()
1116
1117    @cached_property
1118    def default_catalog(self) -> t.Optional[str]:
1119        return self.default_catalog_per_gateway.get(self.selected_gateway)
1120
1121    @python_api_analytics
1122    def render(
1123        self,
1124        model_or_snapshot: ModelOrSnapshot,
1125        *,
1126        start: t.Optional[TimeLike] = None,
1127        end: t.Optional[TimeLike] = None,
1128        execution_time: t.Optional[TimeLike] = None,
1129        expand: t.Union[bool, t.Iterable[str]] = False,
1130        **kwargs: t.Any,
1131    ) -> exp.Expr:
1132        """Renders a model's query, expanding macros with provided kwargs, and optionally expanding referenced models.
1133
1134        Args:
1135            model_or_snapshot: The model, model name, or snapshot to render.
1136            start: The start of the interval to render.
1137            end: The end of the interval to render.
1138            execution_time: The date/time time reference to use for execution time. Defaults to now.
1139            expand: Whether or not to use expand materialized models, defaults to False.
1140                If True, all referenced models are expanded as raw queries.
1141                If a list, only referenced models are expanded as raw queries.
1142
1143        Returns:
1144            The rendered expression.
1145        """
1146        execution_time = execution_time or now()
1147
1148        model = self.get_model(model_or_snapshot, raise_if_missing=True)
1149
1150        if expand and not isinstance(expand, bool):
1151            expand = {
1152                normalize_model_name(
1153                    x, default_catalog=self.default_catalog, dialect=self.default_dialect
1154                )
1155                for x in expand
1156            }
1157
1158        expand = self.dag.upstream(model.fqn) if expand is True else expand or []
1159
1160        if model.is_seed:
1161            import pandas as pd
1162
1163            df = next(
1164                model.render(
1165                    context=self.execution_context(
1166                        engine_adapter=self._get_engine_adapter(model.gateway)
1167                    ),
1168                    start=start,
1169                    end=end,
1170                    execution_time=execution_time,
1171                    **kwargs,
1172                )
1173            )
1174            return next(pandas_to_sql(t.cast(pd.DataFrame, df), model.columns_to_types))
1175
1176        snapshots = self.snapshots
1177        deployability_index = DeployabilityIndex.create(snapshots.values(), start=start)
1178
1179        return model.render_query_or_raise(
1180            start=start,
1181            end=end,
1182            execution_time=execution_time,
1183            snapshots=snapshots,
1184            expand=expand,
1185            deployability_index=deployability_index,
1186            engine_adapter=self._get_engine_adapter(model.gateway),
1187            **kwargs,
1188        )
1189
1190    @python_api_analytics
1191    def evaluate(
1192        self,
1193        model_or_snapshot: ModelOrSnapshot,
1194        start: TimeLike,
1195        end: TimeLike,
1196        execution_time: TimeLike,
1197        limit: t.Optional[int] = None,
1198        **kwargs: t.Any,
1199    ) -> DF:
1200        """Evaluate a model or snapshot (running its query against a DB/Engine).
1201
1202        This method is used to test or iterate on models without side effects.
1203
1204        Args:
1205            model_or_snapshot: The model, model name, or snapshot to render.
1206            start: The start of the interval to evaluate.
1207            end: The end of the interval to evaluate.
1208            execution_time: The date/time time reference to use for execution time.
1209            limit: A limit applied to the model.
1210        """
1211        snapshots = self.snapshots
1212        fqn = self._node_or_snapshot_to_fqn(model_or_snapshot)
1213        if fqn not in snapshots:
1214            raise SQLMeshError(f"Cannot find snapshot for '{fqn}'")
1215        snapshot = snapshots[fqn]
1216
1217        # Expand all uncategorized parents since physical tables don't exist for them yet
1218        expand = [
1219            parent
1220            for parent in self.dag.upstream(snapshot.model.fqn)
1221            if (parent_snapshot := snapshots.get(parent))
1222            and parent_snapshot.is_model
1223            and parent_snapshot.model.is_sql
1224            and not parent_snapshot.categorized
1225        ]
1226
1227        df = self.snapshot_evaluator.evaluate_and_fetch(
1228            snapshot,
1229            start=start,
1230            end=end,
1231            execution_time=execution_time,
1232            snapshots=self.snapshots,
1233            limit=limit or c.DEFAULT_MAX_LIMIT,
1234            expand=expand,
1235        )
1236
1237        if df is None:
1238            raise RuntimeError(f"Error evaluating {snapshot.name}")
1239
1240        return df
1241
1242    @python_api_analytics
1243    def format(
1244        self,
1245        transpile: t.Optional[str] = None,
1246        rewrite_casts: t.Optional[bool] = None,
1247        append_newline: t.Optional[bool] = None,
1248        *,
1249        check: t.Optional[bool] = None,
1250        paths: t.Optional[t.Tuple[t.Union[str, Path], ...]] = None,
1251        **kwargs: t.Any,
1252    ) -> bool:
1253        """Format all SQL models and audits."""
1254        filtered_targets = [
1255            target
1256            for target in chain(self._models.values(), self._audits.values())
1257            if target._path is not None
1258            and target._path.suffix == ".sql"
1259            and (not paths or any(target._path.samefile(p) for p in paths))
1260        ]
1261        unformatted_file_paths = []
1262
1263        for target in filtered_targets:
1264            if (
1265                target._path is None or target.formatting is False
1266            ):  # introduced to satisfy type checker as still want to pull filter out as many targets as possible before loop
1267                continue
1268
1269            with open(target._path, "r+", encoding="utf-8") as file:
1270                before = file.read()
1271
1272                after = self._format(
1273                    target,
1274                    before,
1275                    transpile=transpile,
1276                    rewrite_casts=rewrite_casts,
1277                    append_newline=append_newline,
1278                    **kwargs,
1279                )
1280
1281                if not check:
1282                    file.seek(0)
1283                    file.write(after)
1284                    file.truncate()
1285                elif before != after:
1286                    unformatted_file_paths.append(target._path)
1287
1288        if unformatted_file_paths:
1289            for path in unformatted_file_paths:
1290                self.console.log_status_update(f"{path} needs reformatting.")
1291            self.console.log_status_update(
1292                f"\n{len(unformatted_file_paths)} file(s) need reformatting."
1293            )
1294            return False
1295
1296        return True
1297
1298    def _format(
1299        self,
1300        target: Model | Audit,
1301        before: str,
1302        *,
1303        transpile: t.Optional[str] = None,
1304        rewrite_casts: t.Optional[bool] = None,
1305        append_newline: t.Optional[bool] = None,
1306        **kwargs: t.Any,
1307    ) -> str:
1308        expressions = parse(before, default_dialect=self.config_for_node(target).dialect)
1309        if transpile and is_meta_expression(expressions[0]):
1310            for prop in expressions[0].expressions:
1311                if prop.name.lower() == "dialect":
1312                    prop.replace(
1313                        exp.Property(
1314                            this="dialect",
1315                            value=exp.Literal.string(transpile or target.dialect),
1316                        )
1317                    )
1318
1319        format_config = self.config_for_node(target).format
1320        after = format_model_expressions(
1321            expressions,
1322            transpile or target.dialect,
1323            rewrite_casts=(
1324                rewrite_casts if rewrite_casts is not None else not format_config.no_rewrite_casts
1325            ),
1326            **{**format_config.generator_options, **kwargs},
1327        )
1328
1329        if append_newline is None:
1330            append_newline = format_config.append_newline
1331        if append_newline:
1332            after += "\n"
1333
1334        return after
1335
1336    @python_api_analytics
1337    def plan(
1338        self,
1339        environment: t.Optional[str] = None,
1340        *,
1341        start: t.Optional[TimeLike] = None,
1342        end: t.Optional[TimeLike] = None,
1343        execution_time: t.Optional[TimeLike] = None,
1344        create_from: t.Optional[str] = None,
1345        skip_tests: t.Optional[bool] = None,
1346        restate_models: t.Optional[t.Iterable[str]] = None,
1347        no_gaps: t.Optional[bool] = None,
1348        skip_backfill: t.Optional[bool] = None,
1349        empty_backfill: t.Optional[bool] = None,
1350        forward_only: t.Optional[bool] = None,
1351        allow_destructive_models: t.Optional[t.Collection[str]] = None,
1352        allow_additive_models: t.Optional[t.Collection[str]] = None,
1353        no_prompts: t.Optional[bool] = None,
1354        auto_apply: t.Optional[bool] = None,
1355        no_auto_categorization: t.Optional[bool] = None,
1356        effective_from: t.Optional[TimeLike] = None,
1357        include_unmodified: t.Optional[bool] = None,
1358        select_models: t.Optional[t.Collection[str]] = None,
1359        backfill_models: t.Optional[t.Collection[str]] = None,
1360        categorizer_config: t.Optional[CategorizerConfig] = None,
1361        enable_preview: t.Optional[bool] = None,
1362        no_diff: t.Optional[bool] = None,
1363        run: t.Optional[bool] = None,
1364        diff_rendered: t.Optional[bool] = None,
1365        skip_linter: t.Optional[bool] = None,
1366        explain: t.Optional[bool] = None,
1367        ignore_cron: t.Optional[bool] = None,
1368        min_intervals: t.Optional[int] = None,
1369    ) -> Plan:
1370        """Interactively creates a plan.
1371
1372        This method compares the current context with the target environment. It then presents
1373        the differences and asks whether to backfill each modified model.
1374
1375        Args:
1376            environment: The environment to diff and plan against.
1377            start: The start date of the backfill if there is one.
1378            end: The end date of the backfill if there is one.
1379            execution_time: The date/time reference to use for execution time. Defaults to now.
1380            create_from: The environment to create the target environment from if it
1381                doesn't exist. If not specified, the "prod" environment will be used.
1382            skip_tests: Unit tests are run by default so this will skip them if enabled
1383            restate_models: A list of either internal or external models, or tags, that need to be restated
1384                for the given plan interval. If the target environment is a production environment,
1385                ALL snapshots that depended on these upstream tables will have their intervals deleted
1386                (even ones not in this current environment). Only the snapshots in this environment will
1387                be backfilled whereas others need to be recovered on a future plan application. For development
1388                environments only snapshots that are part of this plan will be affected.
1389            no_gaps:  Whether to ensure that new snapshots for models that are already a
1390                part of the target environment have no data gaps when compared against previous
1391                snapshots for same models.
1392            skip_backfill: Whether to skip the backfill step. Default: False.
1393            empty_backfill: Like skip_backfill, but also records processed intervals.
1394            forward_only: Whether the purpose of the plan is to make forward only changes.
1395            allow_destructive_models: Models whose forward-only changes are allowed to be destructive.
1396            allow_additive_models: Models whose forward-only changes are allowed to be additive.
1397            no_prompts: Whether to disable interactive prompts for the backfill time range. Please note that
1398                if this flag is set to true and there are uncategorized changes the plan creation will
1399                fail. Default: False.
1400            auto_apply: Whether to automatically apply the new plan after creation. Default: False.
1401            no_auto_categorization: Indicates whether to disable automatic categorization of model
1402                changes (breaking / non-breaking). If not provided, then the corresponding configuration
1403                option determines the behavior.
1404            categorizer_config: The configuration for the categorizer. Uses the categorizer configuration defined in the
1405                project config by default.
1406            effective_from: The effective date from which to apply forward-only changes on production.
1407            include_unmodified: Indicates whether to include unmodified models in the target development environment.
1408            select_models: A list of model selection strings to filter the models that should be included into this plan.
1409            backfill_models: A list of model selection strings to filter the models for which the data should be backfilled.
1410            enable_preview: Indicates whether to enable preview for forward-only models in development environments.
1411            no_diff: Hide text differences for changed models.
1412            run: Whether to run latest intervals as part of the plan application.
1413            diff_rendered: Whether the diff should compare raw vs rendered models
1414            skip_linter: Linter runs by default so this will skip it if enabled
1415            explain: Whether to explain the plan instead of applying it.
1416            min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
1417                on every model when checking for missing intervals
1418
1419        Returns:
1420            The populated Plan object.
1421        """
1422        plan_builder = self.plan_builder(
1423            environment,
1424            start=start,
1425            end=end,
1426            execution_time=execution_time,
1427            create_from=create_from,
1428            skip_tests=skip_tests,
1429            restate_models=restate_models,
1430            no_gaps=no_gaps,
1431            skip_backfill=skip_backfill,
1432            empty_backfill=empty_backfill,
1433            forward_only=forward_only,
1434            allow_destructive_models=allow_destructive_models,
1435            allow_additive_models=allow_additive_models,
1436            no_auto_categorization=no_auto_categorization,
1437            effective_from=effective_from,
1438            include_unmodified=include_unmodified,
1439            select_models=select_models,
1440            backfill_models=backfill_models,
1441            categorizer_config=categorizer_config,
1442            enable_preview=enable_preview,
1443            run=run,
1444            diff_rendered=diff_rendered,
1445            skip_linter=skip_linter,
1446            explain=explain,
1447            ignore_cron=ignore_cron,
1448            min_intervals=min_intervals,
1449        )
1450
1451        plan = plan_builder.build()
1452
1453        self._warn_if_virtual_catalog_rematerialization(plan)
1454
1455        if no_auto_categorization or plan.uncategorized:
1456            # Prompts are required if the auto categorization is disabled
1457            # or if there are any uncategorized snapshots in the plan
1458            no_prompts = False
1459
1460        if explain:
1461            auto_apply = True
1462
1463        self.console.plan(
1464            plan_builder,
1465            auto_apply if auto_apply is not None else self.config.plan.auto_apply,
1466            self.default_catalog,
1467            no_diff=no_diff if no_diff is not None else self.config.plan.no_diff,
1468            no_prompts=no_prompts if no_prompts is not None else self.config.plan.no_prompts,
1469        )
1470
1471        return plan
1472
1473    @python_api_analytics
1474    def plan_builder(
1475        self,
1476        environment: t.Optional[str] = None,
1477        *,
1478        start: t.Optional[TimeLike] = None,
1479        end: t.Optional[TimeLike] = None,
1480        execution_time: t.Optional[TimeLike] = None,
1481        create_from: t.Optional[str] = None,
1482        skip_tests: t.Optional[bool] = None,
1483        restate_models: t.Optional[t.Iterable[str]] = None,
1484        no_gaps: t.Optional[bool] = None,
1485        skip_backfill: t.Optional[bool] = None,
1486        empty_backfill: t.Optional[bool] = None,
1487        forward_only: t.Optional[bool] = None,
1488        allow_destructive_models: t.Optional[t.Collection[str]] = None,
1489        allow_additive_models: t.Optional[t.Collection[str]] = None,
1490        no_auto_categorization: t.Optional[bool] = None,
1491        effective_from: t.Optional[TimeLike] = None,
1492        include_unmodified: t.Optional[bool] = None,
1493        select_models: t.Optional[t.Collection[str]] = None,
1494        backfill_models: t.Optional[t.Collection[str]] = None,
1495        categorizer_config: t.Optional[CategorizerConfig] = None,
1496        enable_preview: t.Optional[bool] = None,
1497        preview_start: t.Optional[TimeLike] = None,
1498        preview_min_intervals: t.Optional[int] = None,
1499        run: t.Optional[bool] = None,
1500        diff_rendered: t.Optional[bool] = None,
1501        skip_linter: t.Optional[bool] = None,
1502        explain: t.Optional[bool] = None,
1503        ignore_cron: t.Optional[bool] = None,
1504        min_intervals: t.Optional[int] = None,
1505        always_include_local_changes: t.Optional[bool] = None,
1506    ) -> PlanBuilder:
1507        """Creates a plan builder.
1508
1509        Args:
1510            environment: The environment to diff and plan against.
1511            start: The start date of the backfill if there is one.
1512            end: The end date of the backfill if there is one.
1513            execution_time: The date/time reference to use for execution time. Defaults to now.
1514            create_from: The environment to create the target environment from if it
1515                doesn't exist. If not specified, the "prod" environment will be used.
1516            skip_tests: Unit tests are run by default so this will skip them if enabled
1517            restate_models: A list of either internal or external models, or tags, that need to be restated
1518                for the given plan interval. If the target environment is a production environment,
1519                ALL snapshots that depended on these upstream tables will have their intervals deleted
1520                (even ones not in this current environment). Only the snapshots in this environment will
1521                be backfilled whereas others need to be recovered on a future plan application. For development
1522                environments only snapshots that are part of this plan will be affected.
1523            no_gaps:  Whether to ensure that new snapshots for models that are already a
1524                part of the target environment have no data gaps when compared against previous
1525                snapshots for same models.
1526            skip_backfill: Whether to skip the backfill step. Default: False.
1527            empty_backfill: Like skip_backfill, but also records processed intervals.
1528            forward_only: Whether the purpose of the plan is to make forward only changes.
1529            allow_destructive_models: Models whose forward-only changes are allowed to be destructive.
1530            no_auto_categorization: Indicates whether to disable automatic categorization of model
1531                changes (breaking / non-breaking). If not provided, then the corresponding configuration
1532                option determines the behavior.
1533            categorizer_config: The configuration for the categorizer. Uses the categorizer configuration defined in the
1534                project config by default.
1535            effective_from: The effective date from which to apply forward-only changes on production.
1536            include_unmodified: Indicates whether to include unmodified models in the target development environment.
1537            select_models: A list of model selection strings to filter the models that should be included into this plan.
1538            backfill_models: A list of model selection strings to filter the models for which the data should be backfilled.
1539            enable_preview: Indicates whether to enable preview for forward-only models in development environments.
1540            preview_start: The start date for forward-only previews.
1541            preview_min_intervals: The minimum number of intervals to preview for each forward-only preview snapshot.
1542            run: Whether to run latest intervals as part of the plan application.
1543            diff_rendered: Whether the diff should compare raw vs rendered models
1544            min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
1545                on every model when checking for missing intervals
1546            always_include_local_changes: Usually when restatements are present, local changes in the filesystem are ignored.
1547                However, it can be desirable to deploy changes + restatements in the same plan, so this flag overrides the default behaviour.
1548
1549        Returns:
1550            The plan builder.
1551        """
1552        kwargs: t.Dict[str, t.Optional[UserProvidedFlags]] = {
1553            "start": start,
1554            "end": end,
1555            "execution_time": execution_time,
1556            "create_from": create_from,
1557            "skip_tests": skip_tests,
1558            "restate_models": list(restate_models) if restate_models is not None else None,
1559            "no_gaps": no_gaps,
1560            "skip_backfill": skip_backfill,
1561            "empty_backfill": empty_backfill,
1562            "forward_only": forward_only,
1563            "allow_destructive_models": list(allow_destructive_models)
1564            if allow_destructive_models is not None
1565            else None,
1566            "allow_additive_models": list(allow_additive_models)
1567            if allow_additive_models is not None
1568            else None,
1569            "no_auto_categorization": no_auto_categorization,
1570            "effective_from": effective_from,
1571            "include_unmodified": include_unmodified,
1572            "select_models": list(select_models) if select_models is not None else None,
1573            "backfill_models": list(backfill_models) if backfill_models is not None else None,
1574            "enable_preview": enable_preview,
1575            "preview_start": preview_start,
1576            "preview_min_intervals": preview_min_intervals,
1577            "run": run,
1578            "diff_rendered": diff_rendered,
1579            "skip_linter": skip_linter,
1580            "min_intervals": min_intervals,
1581        }
1582        user_provided_flags: t.Dict[str, UserProvidedFlags] = {
1583            k: v for k, v in kwargs.items() if v is not None
1584        }
1585
1586        skip_tests = explain or skip_tests or False
1587        no_gaps = no_gaps or False
1588        skip_backfill = skip_backfill or False
1589        empty_backfill = empty_backfill or False
1590        run = run or False
1591        diff_rendered = diff_rendered or False
1592        skip_linter = skip_linter or False
1593        min_intervals = min_intervals or 0
1594
1595        environment = environment or self.config.default_target_environment
1596        environment = Environment.sanitize_name(environment)
1597        is_dev = environment != c.PROD
1598
1599        if include_unmodified is None:
1600            include_unmodified = self.config.plan.include_unmodified
1601
1602        if skip_backfill and not no_gaps and not is_dev:
1603            # note: we deliberately don't mention the --no-gaps flag in case the plan came from the sqlmesh_dbt command
1604            # todo: perhaps we could have better error messages if we check sys.argv[0] for which cli is running?
1605            self.console.log_warning(
1606                "Skipping the backfill stage for production can lead to unexpected results, such as tables being empty or incremental data with non-contiguous time ranges being made available.\n"
1607                "If you are doing this deliberately to create an empty version of a table to test a change, please consider using Virtual Data Environments instead."
1608            )
1609
1610        if not skip_linter:
1611            self.lint_models()
1612
1613        self._run_plan_tests(skip_tests=skip_tests)
1614
1615        environment_ttl = (
1616            self.environment_ttl if environment not in self.pinned_environments else None
1617        )
1618
1619        model_selector = self._new_selector()
1620
1621        if allow_destructive_models:
1622            expanded_destructive_models = model_selector.expand_model_selections(
1623                allow_destructive_models
1624            )
1625        else:
1626            expanded_destructive_models = None
1627
1628        if allow_additive_models:
1629            expanded_additive_models = model_selector.expand_model_selections(allow_additive_models)
1630        else:
1631            expanded_additive_models = None
1632
1633        if backfill_models:
1634            backfill_models = model_selector.expand_model_selections(backfill_models)
1635        else:
1636            backfill_models = None
1637
1638        models_override: t.Optional[UniqueKeyDict[str, Model]] = None
1639        selected_fqns: t.Set[str] = set()
1640        selected_deletion_fqns: t.Set[str] = set()
1641        if select_models:
1642            try:
1643                models_override, selected_fqns = model_selector.select_models(
1644                    select_models,
1645                    environment,
1646                    fallback_env_name=create_from or c.PROD,
1647                    ensure_finalized_snapshots=self.config.plan.use_finalized_state,
1648                )
1649            except SQLMeshError as e:
1650                logger.exception(e)  # ensure the full stack trace is logged
1651                raise PlanError(
1652                    f"{e}\nCheck the SQLMesh log file for the full stack trace.\nIf the model has been fixed locally, please ensure that the --select-model expression includes it."
1653                )
1654            if not backfill_models:
1655                # Only backfill selected models unless explicitly specified.
1656                backfill_models = model_selector.expand_model_selections(select_models)
1657
1658            if not backfill_models:
1659                # The selection matched nothing locally. Check whether it matched models
1660                # in the deployed environment that were deleted locally.
1661                selected_deletion_fqns = selected_fqns - set(self._models)
1662
1663        expanded_restate_models = None
1664        if restate_models is not None:
1665            expanded_restate_models = model_selector.expand_model_selections(restate_models)
1666
1667        if (restate_models is not None and not expanded_restate_models) or (
1668            backfill_models is not None and not backfill_models and not selected_deletion_fqns
1669        ):
1670            raise PlanError(
1671                "Selector did not return any models. Please check your model selection and try again."
1672            )
1673
1674        if always_include_local_changes is None:
1675            # default behaviour - if restatements are detected; we operate entirely out of state and ignore local changes
1676            force_no_diff = restate_models is not None or (
1677                backfill_models is not None and not backfill_models and not selected_deletion_fqns
1678            )
1679        else:
1680            force_no_diff = not always_include_local_changes
1681
1682        snapshots = self._snapshots(models_override)
1683        context_diff = self._context_diff(
1684            environment or c.PROD,
1685            snapshots=snapshots,
1686            create_from=create_from,
1687            force_no_diff=force_no_diff,
1688            ensure_finalized_snapshots=self.config.plan.use_finalized_state,
1689            diff_rendered=diff_rendered,
1690            always_recreate_environment=self.config.plan.always_recreate_environment,
1691        )
1692        modified_model_names = {
1693            *context_diff.modified_snapshots,
1694            *[s.name for s in context_diff.added],
1695        }
1696
1697        if (
1698            is_dev
1699            and not include_unmodified
1700            and backfill_models is None
1701            and expanded_restate_models is None
1702        ):
1703            # Only backfill modified and added models.
1704            # This ensures that no models outside the impacted sub-DAG(s) will be backfilled unexpectedly.
1705            backfill_models = modified_model_names or None
1706
1707        max_interval_end_per_model = None
1708        default_start, default_end = None, None
1709        if not run:
1710            ignore_cron = False
1711            max_interval_end_per_model = self._get_max_interval_end_per_model(
1712                snapshots, backfill_models
1713            )
1714            # If no end date is specified, use the max interval end from prod
1715            # to prevent unintended evaluation of the entire DAG.
1716            default_start, default_end = self._get_plan_default_start_end(
1717                snapshots,
1718                max_interval_end_per_model,
1719                backfill_models,
1720                modified_model_names,
1721                execution_time or now(),
1722            )
1723
1724            execution_time_ts = to_timestamp(execution_time) if execution_time is not None else None
1725            if (
1726                execution_time_ts is not None
1727                and end is None
1728                and default_end is not None
1729                and execution_time_ts > default_end
1730            ):
1731                # An explicit execution time is the plan's effective "now", so the default end may
1732                # extend past the recorded prod frontier (as an explicit `end` already does via
1733                # PlanBuilder.override_end). Raising every per-model cap to it keeps a plain
1734                # `plan --execution-time X` in step with `plan --run --execution-time X`, which
1735                # already runs with no caps.
1736                default_end = execution_time_ts
1737                execution_time_dt = to_datetime(execution_time_ts)
1738                max_interval_end_per_model = {
1739                    model_fqn: max(interval_end, execution_time_dt)
1740                    for model_fqn, interval_end in max_interval_end_per_model.items()
1741                }
1742
1743            # Refresh snapshot intervals to ensure that they are up to date with values reflected in the max_interval_end_per_model.
1744            self.state_sync.refresh_snapshot_intervals(context_diff.snapshots.values())
1745
1746        start_override_per_model = self._calculate_start_override_per_model(
1747            min_intervals,
1748            start or default_start,
1749            end or default_end,
1750            execution_time or now(),
1751            backfill_models,
1752            snapshots,
1753            max_interval_end_per_model,
1754        )
1755
1756        if not self.config.virtual_environment_mode.is_full:
1757            forward_only = True
1758        elif forward_only is None:
1759            forward_only = self.config.plan.forward_only
1760
1761        # When handling prod restatements, only clear intervals from other model versions if we are using full virtual environments
1762        # If we are not, then there is no point, because none of the data in dev environments can be promoted by definition
1763        restate_all_snapshots = (
1764            expanded_restate_models is not None
1765            and not is_dev
1766            and self.config.virtual_environment_mode.is_full
1767        )
1768
1769        return self.PLAN_BUILDER_TYPE(
1770            context_diff=context_diff,
1771            start=start,
1772            end=end,
1773            execution_time=execution_time,
1774            apply=self.apply,
1775            restate_models=expanded_restate_models,
1776            restate_all_snapshots=restate_all_snapshots,
1777            backfill_models=backfill_models,
1778            no_gaps=no_gaps,
1779            skip_backfill=skip_backfill,
1780            empty_backfill=empty_backfill,
1781            is_dev=is_dev,
1782            forward_only=forward_only,
1783            allow_destructive_models=expanded_destructive_models,
1784            allow_additive_models=expanded_additive_models,
1785            environment_ttl=environment_ttl,
1786            environment_suffix_target=self.config.environment_suffix_target,
1787            environment_catalog_mapping=self.environment_catalog_mapping,
1788            categorizer_config=categorizer_config or self.auto_categorize_changes,
1789            auto_categorization_enabled=not no_auto_categorization,
1790            effective_from=effective_from,
1791            include_unmodified=include_unmodified,
1792            default_start=default_start,
1793            default_end=default_end,
1794            enable_preview=(
1795                enable_preview if enable_preview is not None else self._plan_preview_enabled
1796            ),
1797            preview_start=preview_start,
1798            preview_min_intervals=preview_min_intervals or 0,
1799            end_bounded=not run,
1800            ensure_finalized_snapshots=self.config.plan.use_finalized_state,
1801            start_override_per_model=start_override_per_model,
1802            end_override_per_model=max_interval_end_per_model,
1803            console=self.console,
1804            user_provided_flags=user_provided_flags,
1805            selected_models={
1806                dbt_unique_id
1807                for model in model_selector.expand_model_selections(select_models or "*")
1808                if (dbt_unique_id := snapshots[model].node.dbt_unique_id)
1809            },
1810            explain=explain or False,
1811            ignore_cron=ignore_cron or False,
1812        )
1813
1814    def apply(
1815        self,
1816        plan: Plan,
1817        circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
1818    ) -> None:
1819        """Applies a plan by pushing snapshots and backfilling data.
1820
1821        Given a plan, it pushes snapshots into the state sync and then uses the scheduler
1822        to backfill all models.
1823
1824        Args:
1825            plan: The plan to apply.
1826            circuit_breaker: An optional handler which checks if the apply should be aborted.
1827        """
1828        if (
1829            not plan.context_diff.has_changes
1830            and not plan.requires_backfill
1831            and not plan.has_unmodified_unpromoted
1832        ):
1833            return
1834        if plan.uncategorized:
1835            raise UncategorizedPlanError("Can't apply a plan with uncategorized changes.")
1836
1837        if plan.explain:
1838            explainer = PlanExplainer(
1839                state_reader=self.state_reader,
1840                default_catalog=self.default_catalog,
1841                console=self.console,
1842            )
1843            explainer.evaluate(plan.to_evaluatable())
1844            return
1845
1846        self.notification_target_manager.notify(
1847            NotificationEvent.APPLY_START,
1848            environment=plan.environment_naming_info.name,
1849            plan_id=plan.plan_id,
1850        )
1851        try:
1852            self._apply(plan, circuit_breaker)
1853        except Exception as e:
1854            self.notification_target_manager.notify(
1855                NotificationEvent.APPLY_FAILURE,
1856                environment=plan.environment_naming_info.name,
1857                plan_id=plan.plan_id,
1858                exc=traceback.format_exc(),
1859            )
1860            logger.info("Plan application failed.", exc_info=e)
1861            raise e
1862        self.notification_target_manager.notify(
1863            NotificationEvent.APPLY_END,
1864            environment=plan.environment_naming_info.name,
1865            plan_id=plan.plan_id,
1866        )
1867
1868    @python_api_analytics
1869    def invalidate_environment(self, name: str, sync: bool = False) -> None:
1870        """Invalidates the target environment by setting its expiration timestamp to now.
1871
1872        Args:
1873            name: The name of the environment to invalidate.
1874            sync: If True, the call blocks until the environment is deleted. Otherwise, the environment will
1875                be deleted asynchronously by the janitor process.
1876        """
1877        name = Environment.sanitize_name(name)
1878        self.state_sync.invalidate_environment(name)
1879        if sync:
1880            self._cleanup_environments(name=name)
1881            self.console.log_success(f"Environment '{name}' deleted.")
1882        else:
1883            self.console.log_success(f"Environment '{name}' invalidated.")
1884
1885    @python_api_analytics
1886    def diff(self, environment: t.Optional[str] = None, detailed: bool = False) -> bool:
1887        """Show a diff of the current context with a given environment.
1888
1889        Args:
1890            environment: The environment to diff against.
1891            detailed: Show the actual SQL differences if True.
1892
1893        Returns:
1894            True if there are changes, False otherwise.
1895        """
1896        environment = environment or self.config.default_target_environment
1897        environment = Environment.sanitize_name(environment)
1898        context_diff = self._context_diff(environment)
1899        self.console.show_environment_difference_summary(
1900            context_diff,
1901            no_diff=not detailed,
1902        )
1903        if context_diff.has_changes:
1904            self.console.show_model_difference_summary(
1905                context_diff,
1906                EnvironmentNamingInfo.from_environment_catalog_mapping(
1907                    self.environment_catalog_mapping,
1908                    name=environment,
1909                    suffix_target=self.config.environment_suffix_target,
1910                    normalize_name=context_diff.normalize_environment_name,
1911                ),
1912                self.default_catalog,
1913                no_diff=not detailed,
1914            )
1915        return context_diff.has_changes
1916
1917    @python_api_analytics
1918    def table_diff(
1919        self,
1920        source: str,
1921        target: str,
1922        on: t.Optional[t.List[str] | exp.Expr] = None,
1923        skip_columns: t.Optional[t.List[str]] = None,
1924        select_models: t.Optional[t.Collection[str]] = None,
1925        where: t.Optional[str | exp.Expr] = None,
1926        limit: int = 20,
1927        show: bool = True,
1928        show_sample: bool = True,
1929        decimals: int = 3,
1930        skip_grain_check: bool = False,
1931        warn_grain_check: bool = False,
1932        temp_schema: t.Optional[str] = None,
1933        schema_diff_ignore_case: bool = False,
1934        **kwargs: t.Any,  # catch-all to prevent an 'unexpected keyword argument' error if an table_diff extension passes in some extra arguments
1935    ) -> t.List[TableDiff]:
1936        """Show a diff between two tables.
1937
1938        Args:
1939            source: The source environment or table.
1940            target: The target environment or table.
1941            on: The join condition, table aliases must be "s" and "t" for source and target.
1942                If omitted, the table's grain will be used.
1943            skip_columns: The columns to skip when computing the table diff.
1944            select_models: The models or snapshots to use when environments are passed in.
1945            where: An optional where statement to filter results.
1946            limit: The limit of the sample dataframe.
1947            show: Show the table diff output in the console.
1948            show_sample: Show the sample dataframe in the console. Requires show=True.
1949            decimals: The number of decimal places to keep when comparing floating point columns.
1950            skip_grain_check: Skip check for rows that contain null or duplicate grains.
1951            temp_schema: The schema to use for temporary tables.
1952
1953        Returns:
1954            The list of TableDiff objects containing schema and summary differences.
1955        """
1956
1957        if "|" in source or "|" in target:
1958            raise ConfigError(
1959                "Cross-database table diffing is available in Tobiko Cloud. Read more here: "
1960                "https://sqlmesh.readthedocs.io/en/stable/guides/tablediff/#diffing-tables-or-views-across-gateways"
1961            )
1962
1963        table_diffs: t.List[TableDiff] = []
1964
1965        # Diffs multiple or a single model across two environments
1966        if select_models:
1967            source_env = self.state_reader.get_environment(source)
1968            target_env = self.state_reader.get_environment(target)
1969            if not source_env:
1970                raise SQLMeshError(f"Could not find environment '{source}'")
1971            if not target_env:
1972                raise SQLMeshError(f"Could not find environment '{target}'")
1973            criteria = ", ".join(f"'{c}'" for c in select_models)
1974            try:
1975                selected_models = self._new_selector().expand_model_selections(select_models)
1976                if not selected_models:
1977                    self.console.log_status_update(
1978                        f"No models matched the selection criteria: {criteria}"
1979                    )
1980            except Exception as e:
1981                raise SQLMeshError(e)
1982
1983            models_to_diff: t.List[
1984                t.Tuple[Model, EngineAdapter, str, str, t.Optional[t.List[str] | exp.Expr]]
1985            ] = []
1986            models_without_grain: t.List[Model] = []
1987            source_snapshots_to_name = {
1988                snapshot.name: snapshot for snapshot in source_env.snapshots
1989            }
1990            target_snapshots_to_name = {
1991                snapshot.name: snapshot for snapshot in target_env.snapshots
1992            }
1993
1994            for model_fqn in selected_models:
1995                model = self._models[model_fqn]
1996                adapter = self._get_engine_adapter(model.gateway)
1997                source_snapshot = source_snapshots_to_name.get(model.fqn)
1998                target_snapshot = target_snapshots_to_name.get(model.fqn)
1999
2000                if target_snapshot and source_snapshot:
2001                    if (source_snapshot.fingerprint != target_snapshot.fingerprint) and (
2002                        (source_snapshot.version != target_snapshot.version)
2003                        or source_snapshot.is_forward_only
2004                    ):
2005                        # Compare the virtual layer instead of the physical layer because the virtual layer is guaranteed to point
2006                        # to the correct/active snapshot for the model in the specified environment, taking into account things like dev previews
2007                        source = source_snapshot.qualified_view_name.for_environment(
2008                            source_env.naming_info, adapter.dialect
2009                        )
2010                        target = target_snapshot.qualified_view_name.for_environment(
2011                            target_env.naming_info, adapter.dialect
2012                        )
2013                        model_on = on or model.on
2014                        if not model_on:
2015                            models_without_grain.append(model)
2016                        else:
2017                            models_to_diff.append((model, adapter, source, target, model_on))
2018
2019            if models_without_grain:
2020                model_names = "\n".join(
2021                    f"─ {model.name} \n  at '{model._path}'" for model in models_without_grain
2022                )
2023                message = (
2024                    "SQLMesh doesn't know how to join the tables for the following models:\n"
2025                    f"{model_names}\n\n"
2026                    "Please specify a `grain` in each model definition. It must be unique and not null."
2027                )
2028                if warn_grain_check:
2029                    self.console.log_warning(message)
2030                else:
2031                    raise SQLMeshError(message)
2032
2033            if models_to_diff:
2034                self.console.show_table_diff_details(
2035                    [model[0].name for model in models_to_diff],
2036                )
2037
2038                self.console.start_table_diff_progress(len(models_to_diff))
2039                try:
2040                    tasks_num = min(len(models_to_diff), self.concurrent_tasks)
2041                    table_diffs = concurrent_apply_to_values(
2042                        list(models_to_diff),
2043                        lambda model_info: self._model_diff(
2044                            model=model_info[0],
2045                            adapter=model_info[1],
2046                            source=model_info[2],
2047                            target=model_info[3],
2048                            on=model_info[4],
2049                            source_alias=source_env.name,
2050                            target_alias=target_env.name,
2051                            limit=limit,
2052                            decimals=decimals,
2053                            skip_columns=skip_columns,
2054                            where=where,
2055                            show=show,
2056                            temp_schema=temp_schema,
2057                            skip_grain_check=skip_grain_check,
2058                            schema_diff_ignore_case=schema_diff_ignore_case,
2059                        ),
2060                        tasks_num=tasks_num,
2061                    )
2062                    self.console.stop_table_diff_progress(success=True)
2063                except:
2064                    self.console.stop_table_diff_progress(success=False)
2065                    raise
2066            elif selected_models:
2067                self.console.log_status_update(
2068                    f"No models contain differences with the selection criteria: {criteria}"
2069                )
2070
2071        else:
2072            table_diffs = [
2073                self._table_diff(
2074                    source=source,
2075                    target=target,
2076                    source_alias=source,
2077                    target_alias=target,
2078                    limit=limit,
2079                    decimals=decimals,
2080                    adapter=self.engine_adapter,
2081                    on=on,
2082                    skip_columns=skip_columns,
2083                    where=where,
2084                    schema_diff_ignore_case=schema_diff_ignore_case,
2085                )
2086            ]
2087
2088        if show:
2089            self.console.show_table_diff(table_diffs, show_sample, skip_grain_check, temp_schema)
2090
2091        return table_diffs
2092
2093    def _model_diff(
2094        self,
2095        model: Model,
2096        adapter: EngineAdapter,
2097        source: str,
2098        target: str,
2099        source_alias: str,
2100        target_alias: str,
2101        limit: int,
2102        decimals: int,
2103        on: t.Optional[t.List[str] | exp.Expr] = None,
2104        skip_columns: t.Optional[t.List[str]] = None,
2105        where: t.Optional[str | exp.Expr] = None,
2106        show: bool = True,
2107        temp_schema: t.Optional[str] = None,
2108        skip_grain_check: bool = False,
2109        schema_diff_ignore_case: bool = False,
2110    ) -> TableDiff:
2111        self.console.start_table_diff_model_progress(model.name)
2112
2113        table_diff = self._table_diff(
2114            on=on,
2115            skip_columns=skip_columns,
2116            where=where,
2117            limit=limit,
2118            decimals=decimals,
2119            model=model,
2120            adapter=adapter,
2121            source=source,
2122            target=target,
2123            source_alias=source_alias,
2124            target_alias=target_alias,
2125            schema_diff_ignore_case=schema_diff_ignore_case,
2126        )
2127
2128        if show:
2129            # Trigger row_diff in parallel execution so it's available for ordered display later
2130            table_diff.row_diff(temp_schema=temp_schema, skip_grain_check=skip_grain_check)
2131
2132        self.console.update_table_diff_progress(model.name)
2133
2134        return table_diff
2135
2136    def _table_diff(
2137        self,
2138        source: str,
2139        target: str,
2140        source_alias: str,
2141        target_alias: str,
2142        limit: int,
2143        decimals: int,
2144        adapter: EngineAdapter,
2145        on: t.Optional[t.List[str] | exp.Expr] = None,
2146        model: t.Optional[Model] = None,
2147        skip_columns: t.Optional[t.List[str]] = None,
2148        where: t.Optional[str | exp.Expr] = None,
2149        schema_diff_ignore_case: bool = False,
2150    ) -> TableDiff:
2151        if not on:
2152            raise SQLMeshError(
2153                "SQLMesh doesn't know how to join the two tables. Specify the `grains` in each model definition or pass join column names in separate `-o` flags."
2154            )
2155
2156        return TableDiff(
2157            adapter=adapter.with_settings(execute_log_level=logger.getEffectiveLevel()),
2158            source=source,
2159            target=target,
2160            on=on,
2161            skip_columns=skip_columns,
2162            where=where,
2163            source_alias=source_alias,
2164            target_alias=target_alias,
2165            limit=limit,
2166            decimals=decimals,
2167            model_name=model.name if model else None,
2168            model_dialect=model.dialect if model else None,
2169            schema_diff_ignore_case=schema_diff_ignore_case,
2170        )
2171
2172    @python_api_analytics
2173    def get_dag(
2174        self, select_models: t.Optional[t.Collection[str]] = None, **options: t.Any
2175    ) -> GraphHTML:
2176        """Gets an HTML object representation of the DAG.
2177
2178        Args:
2179            select_models: A list of model selection strings that should be included in the dag.
2180        Returns:
2181            An html object that renders the dag.
2182        """
2183        dag = (
2184            self.dag.prune(*self._new_selector().expand_model_selections(select_models))
2185            if select_models
2186            else self.dag
2187        )
2188
2189        nodes = {}
2190        edges: t.List[t.Dict] = []
2191
2192        for node, deps in dag.graph.items():
2193            nodes[node] = {
2194                "id": node,
2195                "label": node.split(".")[-1],
2196                "title": f"<span>{node}</span>",
2197            }
2198            edges.extend({"from": d, "to": node} for d in deps)
2199
2200        return GraphHTML(
2201            nodes,
2202            edges,
2203            options={
2204                "height": "100%",
2205                "width": "100%",
2206                "interaction": {},
2207                "layout": {
2208                    "hierarchical": {
2209                        "enabled": True,
2210                        "nodeSpacing": 200,
2211                        "sortMethod": "directed",
2212                    },
2213                },
2214                "nodes": {
2215                    "shape": "box",
2216                },
2217                **options,
2218            },
2219        )
2220
2221    @python_api_analytics
2222    def render_dag(self, path: str, select_models: t.Optional[t.Collection[str]] = None) -> None:
2223        """Render the dag as HTML and save it to a file.
2224
2225        Args:
2226            path: filename to save the dag html to
2227            select_models: A list of model selection strings that should be included in the dag.
2228        """
2229        file_path = Path(path)
2230        suffix = file_path.suffix
2231        if suffix != ".html":
2232            if suffix:
2233                get_console().log_warning(
2234                    f"The extension {suffix} does not designate an html file. A file with a `.html` extension will be created instead."
2235                )
2236            path = str(file_path.with_suffix(".html"))
2237
2238        with open(path, "w", encoding="utf-8") as file:
2239            file.write(str(self.get_dag(select_models)))
2240
2241    @python_api_analytics
2242    def create_test(
2243        self,
2244        model: str,
2245        input_queries: t.Dict[str, str],
2246        overwrite: bool = False,
2247        variables: t.Optional[t.Dict[str, str]] = None,
2248        path: t.Optional[str] = None,
2249        name: t.Optional[str] = None,
2250        include_ctes: bool = False,
2251    ) -> None:
2252        """Generate a unit test fixture for a given model.
2253
2254        Args:
2255            model: The model to test.
2256            input_queries: Mapping of model names to queries. Each model included in this mapping
2257                will be populated in the test based on the results of the corresponding query.
2258            overwrite: Whether to overwrite the existing test in case of a file path collision.
2259                When set to False, an error will be raised if there is such a collision.
2260            variables: Key-value pairs that will define variables needed by the model.
2261            path: The file path corresponding to the fixture, relative to the test directory.
2262                By default, the fixture will be created under the test directory and the file name
2263                will be inferred from the test's name.
2264            name: The name of the test. This is inferred from the model name by default.
2265            include_ctes: When true, CTE fixtures will also be generated.
2266        """
2267        input_queries = {
2268            # The get_model here has two purposes: return normalized names & check for missing deps
2269            self.get_model(dep, raise_if_missing=True).fqn: query
2270            for dep, query in input_queries.items()
2271        }
2272
2273        try:
2274            model_to_test = self.get_model(model, raise_if_missing=True)
2275            test_adapter = self.test_connection_config.create_engine_adapter(
2276                register_comments_override=False
2277            )
2278
2279            generate_test(
2280                model=model_to_test,
2281                input_queries=input_queries,
2282                models=self._models,
2283                engine_adapter=self._get_engine_adapter(model_to_test.gateway),
2284                test_engine_adapter=test_adapter,
2285                project_path=self.path,
2286                overwrite=overwrite,
2287                variables=variables,
2288                path=path,
2289                name=name,
2290                include_ctes=include_ctes,
2291            )
2292        finally:
2293            if test_adapter:
2294                test_adapter.close()
2295
2296    @python_api_analytics
2297    def test(
2298        self,
2299        match_patterns: t.Optional[t.List[str]] = None,
2300        tests: t.Optional[t.List[str]] = None,
2301        verbosity: Verbosity = Verbosity.DEFAULT,
2302        preserve_fixtures: bool = False,
2303        stream: t.Optional[t.TextIO] = None,
2304    ) -> ModelTextTestResult:
2305        """Discover and run model tests"""
2306        if verbosity >= Verbosity.VERBOSE:
2307            import pandas as pd
2308
2309            pd.set_option("display.max_columns", None)
2310
2311        test_meta = self.select_tests(tests=tests, patterns=match_patterns)
2312
2313        result = run_tests(
2314            model_test_metadata=test_meta,
2315            models=self._models,
2316            config=self.config,
2317            selected_gateway=self.selected_gateway,
2318            dialect=self.default_dialect,
2319            verbosity=verbosity,
2320            preserve_fixtures=preserve_fixtures,
2321            stream=stream,
2322            default_catalog=self.default_catalog,
2323            default_catalog_dialect=self.config.dialect or "",
2324        )
2325
2326        self.console.log_test_results(
2327            result,
2328            self.test_connection_config._engine_adapter.DIALECT,
2329        )
2330
2331        return result
2332
2333    @python_api_analytics
2334    def audit(
2335        self,
2336        start: TimeLike,
2337        end: TimeLike,
2338        *,
2339        models: t.Optional[t.Iterator[str]] = None,
2340        execution_time: t.Optional[TimeLike] = None,
2341    ) -> bool:
2342        """Audit models.
2343
2344        Args:
2345            start: The start of the interval to audit.
2346            end: The end of the interval to audit.
2347            models: The models to audit. All models will be audited if not specified.
2348            execution_time: The date/time time reference to use for execution time. Defaults to now.
2349
2350        Returns:
2351            False if any of the audits failed, True otherwise.
2352        """
2353
2354        snapshots = (
2355            [self.get_snapshot(model, raise_if_missing=True) for model in models]
2356            if models
2357            else self.snapshots.values()
2358        )
2359
2360        num_audits = sum(len(snapshot.node.audits_with_args) for snapshot in snapshots)
2361        self.console.log_status_update(f"Found {num_audits} audit(s).")
2362
2363        errors = []
2364        skipped_count = 0
2365        for snapshot in snapshots:
2366            for audit_result in self.snapshot_evaluator.audit(
2367                snapshot=snapshot,
2368                start=start,
2369                end=end,
2370                execution_time=execution_time,
2371                snapshots=self.snapshots,
2372            ):
2373                audit_id = f"{audit_result.audit.name}"
2374                if audit_result.model:
2375                    audit_id += f" on model {audit_result.model.name}"
2376
2377                if audit_result.skipped:
2378                    self.console.log_status_update(f"{audit_id} ⏸️ SKIPPED.")
2379                    skipped_count += 1
2380                elif audit_result.count:
2381                    errors.append(audit_result)
2382                    self.console.log_status_update(
2383                        f"{audit_id} ❌ [red]FAIL [{audit_result.count}][/red]."
2384                    )
2385                else:
2386                    self.console.log_status_update(f"{audit_id} ✅ [green]PASS[/green].")
2387
2388        self.console.log_status_update(
2389            f"\nFinished with {len(errors)} audit error{'' if len(errors) == 1 else 's'} "
2390            f"and {skipped_count} audit{'' if skipped_count == 1 else 's'} skipped."
2391        )
2392        for error in errors:
2393            self.console.log_status_update(
2394                f"\nFailure in audit {error.audit.name} ({error.audit._path})."
2395            )
2396            self.console.log_status_update(f"Got {error.count} results, expected 0.")
2397            if error.query:
2398                self.console.show_sql(
2399                    f"{error.query.sql(dialect=self.snapshot_evaluator.adapter.dialect)}"
2400                )
2401
2402        self.console.log_status_update("Done.")
2403        return not errors
2404
2405    @python_api_analytics
2406    def rewrite(self, sql: str, dialect: str = "") -> exp.Expr:
2407        """Rewrite a sql expression with semantic references into an executable query.
2408
2409        https://sqlmesh.readthedocs.io/en/latest/concepts/metrics/overview/
2410
2411        Args:
2412            sql: The sql string to rewrite.
2413            dialect: The dialect of the sql string, defaults to the project dialect.
2414
2415        Returns:
2416            A SQLGlot expression with semantic references expanded.
2417        """
2418        return rewrite(
2419            sql,
2420            graph=ReferenceGraph(self.models.values()),
2421            metrics=self._metrics,
2422            dialect=dialect or self.default_dialect,
2423        )
2424
2425    @python_api_analytics
2426    def check_intervals(
2427        self,
2428        environment: t.Optional[str],
2429        no_signals: bool,
2430        select_models: t.Collection[str],
2431        start: t.Optional[TimeLike] = None,
2432        end: t.Optional[TimeLike] = None,
2433    ) -> t.Dict[Snapshot, SnapshotIntervals]:
2434        """Check intervals for a given environment.
2435
2436        Args:
2437            environment: The environment or prod if None.
2438            select_models: A list of model selection strings to show intervals for.
2439            start: The start of the intervals to check.
2440            end: The end of the intervals to check.
2441        """
2442
2443        environment = environment or c.PROD
2444        env = self.state_reader.get_environment(environment)
2445        if not env:
2446            raise SQLMeshError(f"Environment '{environment}' was not found.")
2447
2448        snapshots = {k.name: v for k, v in self.state_sync.get_snapshots(env.snapshots).items()}
2449
2450        missing = {
2451            k.name: v
2452            for k, v in missing_intervals(
2453                snapshots.values(), start=start, end=end, execution_time=end
2454            ).items()
2455        }
2456
2457        if select_models:
2458            selected: t.Collection[str] = self._select_models_for_run(
2459                select_models, True, snapshots.values()
2460            )
2461        else:
2462            selected = snapshots.keys()
2463
2464        results = {}
2465        execution_context = self.execution_context(snapshots=snapshots)
2466
2467        for fqn in selected:
2468            snapshot = snapshots[fqn]
2469            intervals = missing.get(fqn) or []
2470
2471            results[snapshot] = SnapshotIntervals(
2472                snapshot.snapshot_id,
2473                intervals
2474                if no_signals
2475                else snapshot.check_ready_intervals(intervals, execution_context),
2476            )
2477
2478        return results
2479
2480    @python_api_analytics
2481    def migrate(self) -> None:
2482        """Migrates SQLMesh to the current running version.
2483
2484        Please contact your SQLMesh administrator before doing this.
2485        """
2486        self.notification_target_manager.notify(NotificationEvent.MIGRATION_START)
2487        self._load_materializations()
2488        try:
2489            self._new_state_sync().migrate(
2490                promoted_snapshots_only=self.config.migration.promoted_snapshots_only,
2491            )
2492        except Exception as e:
2493            self.notification_target_manager.notify(
2494                NotificationEvent.MIGRATION_FAILURE, traceback.format_exc()
2495            )
2496            raise e
2497        self.notification_target_manager.notify(NotificationEvent.MIGRATION_END)
2498
2499    @python_api_analytics
2500    def rollback(self) -> None:
2501        """Rolls back SQLMesh to the previous migration.
2502
2503        Please contact your SQLMesh administrator before doing this. This action cannot be undone.
2504        """
2505        self._new_state_sync().rollback()
2506
2507    @python_api_analytics
2508    def create_external_models(self, strict: bool = False) -> None:
2509        """Create a file to document the schema of external models.
2510
2511        The external models file contains all columns and types of external models, allowing for more
2512        robust lineage, validation, and optimizations.
2513
2514        Args:
2515            strict: If True, raise an error if the external model is missing in the database.
2516        """
2517        if not self._models:
2518            self.load(update_schemas=False)
2519
2520        for path, config in self.configs.items():
2521            deprecated_yaml = path / c.EXTERNAL_MODELS_DEPRECATED_YAML
2522
2523            external_models_yaml = (
2524                path / c.EXTERNAL_MODELS_YAML if not deprecated_yaml.exists() else deprecated_yaml
2525            )
2526
2527            external_models_gateway: t.Optional[str] = self.gateway or self.config.default_gateway
2528            if not external_models_gateway:
2529                # can happen if there was no --gateway defined and the default_gateway is ''
2530                # which means that the single gateway syntax is being used which means there is
2531                # no named gateway which means we should not stamp `gateway:` on the external models
2532                external_models_gateway = None
2533
2534            create_external_models_file(
2535                path=external_models_yaml,
2536                models=UniqueKeyDict(
2537                    "models",
2538                    {
2539                        fqn: model
2540                        for fqn, model in self._models.items()
2541                        if self.config_for_node(model) is config
2542                    },
2543                ),
2544                adapter=self.engine_adapter,
2545                state_reader=self.state_reader,
2546                dialect=config.model_defaults.dialect,
2547                gateway=external_models_gateway,
2548                max_workers=self.concurrent_tasks,
2549                strict=strict,
2550                all_models=self._models,
2551            )
2552
2553    @python_api_analytics
2554    def print_info(
2555        self, skip_connection: bool = False, verbosity: Verbosity = Verbosity.DEFAULT
2556    ) -> None:
2557        """Prints information about connections, models, macros, etc. to the console."""
2558        self.console.log_status_update(f"Models: {len(self.models)}")
2559        self.console.log_status_update(f"Macros: {len(self._macros) - len(macro.get_registry())}")
2560
2561        if skip_connection:
2562            return
2563
2564        if verbosity >= Verbosity.VERBOSE:
2565            self.console.log_status_update("")
2566            print_config(self.config.get_connection(self.gateway), self.console, "Connection")
2567            print_config(
2568                self.config.get_test_connection(self.gateway), self.console, "Test Connection"
2569            )
2570            print_config(
2571                self.config.get_state_connection(self.gateway), self.console, "State Connection"
2572            )
2573
2574        self._try_connection("data warehouse", self.engine_adapter.ping)
2575        state_connection = self.config.get_state_connection(self.gateway)
2576        if state_connection:
2577            self._try_connection("state backend", state_connection.connection_validator())
2578
2579    @python_api_analytics
2580    def print_environment_names(self) -> None:
2581        """Prints all environment names along with expiry datetime."""
2582        result = self._new_state_sync().get_environments_summary()
2583        if not result:
2584            raise SQLMeshError(
2585                "This project has no environments. Create an environment using the `sqlmesh plan` command."
2586            )
2587        self.console.print_environments(result)
2588
2589    def close(self) -> None:
2590        """Releases all resources allocated by this context."""
2591        if self._snapshot_evaluator:
2592            self._snapshot_evaluator.close()
2593
2594        if self._state_sync:
2595            self._state_sync.close()
2596
2597    def _run(
2598        self,
2599        environment: str,
2600        *,
2601        start: t.Optional[TimeLike],
2602        end: t.Optional[TimeLike],
2603        execution_time: t.Optional[TimeLike],
2604        ignore_cron: bool,
2605        select_models: t.Optional[t.Collection[str]],
2606        circuit_breaker: t.Optional[t.Callable[[], bool]],
2607        no_auto_upstream: bool,
2608    ) -> CompletionStatus:
2609        scheduler = self.scheduler(environment=environment)
2610        snapshots = scheduler.snapshots
2611
2612        if select_models is not None:
2613            select_models = self._select_models_for_run(
2614                select_models, no_auto_upstream, snapshots.values()
2615            )
2616
2617        completion_status = scheduler.run(
2618            environment,
2619            start=start,
2620            end=end,
2621            execution_time=execution_time,
2622            ignore_cron=ignore_cron,
2623            circuit_breaker=circuit_breaker,
2624            selected_snapshots=select_models,
2625            auto_restatement_enabled=environment.lower() == c.PROD,
2626            run_environment_statements=True,
2627        )
2628
2629        if completion_status.is_nothing_to_do:
2630            next_run_ready_msg = ""
2631
2632            next_ready_interval_start = get_next_model_interval_start(snapshots.values())
2633            if next_ready_interval_start:
2634                utc_time = format_tz_datetime(next_ready_interval_start)
2635                local_time = format_tz_datetime(next_ready_interval_start, use_local_timezone=True)
2636                time_msg = local_time if local_time == utc_time else f"{local_time} ({utc_time})"
2637                next_run_ready_msg = f"\n\nNext run will be ready at {time_msg}."
2638
2639            self.console.log_status_update(
2640                f"No models are ready to run. Please wait until a model `cron` interval has elapsed.{next_run_ready_msg}"
2641            )
2642
2643        return completion_status
2644
2645    def _apply(self, plan: Plan, circuit_breaker: t.Optional[t.Callable[[], bool]]) -> None:
2646        self._scheduler.create_plan_evaluator(self).evaluate(
2647            plan.to_evaluatable(), circuit_breaker=circuit_breaker
2648        )
2649
2650    @python_api_analytics
2651    def table_name(
2652        self, model_name: str, environment: t.Optional[str] = None, prod: bool = False
2653    ) -> str:
2654        """Returns the name of the pysical table for the given model name in the target environment.
2655
2656        Args:
2657            model_name: The name of the model.
2658            environment: The environment to source the model version from.
2659            prod: If True, return the name of the physical table that will be used in production for the model version
2660                promoted in the target environment.
2661
2662        Returns:
2663            The name of the physical table.
2664        """
2665        environment = environment or self.config.default_target_environment
2666        fqn = self._node_or_snapshot_to_fqn(model_name)
2667        target_env = self.state_reader.get_environment(environment)
2668        if not target_env:
2669            raise SQLMeshError(f"Environment '{environment}' was not found.")
2670
2671        snapshot_info = None
2672        for s in target_env.snapshots:
2673            if s.name == fqn:
2674                snapshot_info = s
2675                break
2676        if not snapshot_info:
2677            raise SQLMeshError(
2678                f"Model '{model_name}' was not found in environment '{environment}'."
2679            )
2680
2681        if target_env.name == c.PROD or prod:
2682            return snapshot_info.table_name()
2683
2684        snapshots = self.state_reader.get_snapshots(target_env.snapshots)
2685        deployability_index = DeployabilityIndex.create(snapshots)
2686
2687        return snapshot_info.table_name(
2688            is_deployable=deployability_index.is_deployable(snapshot_info.snapshot_id)
2689        )
2690
2691    def clear_caches(self) -> None:
2692        paths_to_remove = [path / c.CACHE for path in self.configs]
2693        paths_to_remove.append(self.cache_dir)
2694
2695        if IS_WINDOWS:
2696            paths_to_remove = [fix_windows_path(path) for path in paths_to_remove]
2697
2698        for path in paths_to_remove:
2699            if path.exists():
2700                rmtree(path)
2701
2702        if isinstance(self._state_sync, CachingStateSync):
2703            self._state_sync.clear_cache()
2704
2705    def export_state(
2706        self,
2707        output_file: Path,
2708        environment_names: t.Optional[t.List[str]] = None,
2709        local_only: bool = False,
2710        confirm: bool = True,
2711    ) -> None:
2712        from sqlmesh.core.state_sync.export_import import export_state
2713
2714        # trigger a connection to the StateSync so we can fail early if there is a problem
2715        # note we still need to do this even if we are doing a local export so we know what 'versions' to write
2716        self.state_sync.get_versions(validate=True)
2717
2718        local_snapshots = self.snapshots if local_only else None
2719
2720        if self.console.start_state_export(
2721            output_file=output_file,
2722            gateway=self.selected_gateway,
2723            state_connection_config=self._state_connection_config,
2724            environment_names=environment_names,
2725            local_only=local_only,
2726            confirm=confirm,
2727        ):
2728            try:
2729                export_state(
2730                    state_sync=self.state_sync,
2731                    output_file=output_file,
2732                    local_snapshots=local_snapshots,
2733                    environment_names=environment_names,
2734                    console=self.console,
2735                )
2736                self.console.stop_state_export(success=True, output_file=output_file)
2737            except:
2738                self.console.stop_state_export(success=False, output_file=output_file)
2739                raise
2740
2741    def import_state(self, input_file: Path, clear: bool = False, confirm: bool = True) -> None:
2742        from sqlmesh.core.state_sync.export_import import import_state
2743
2744        if self.console.start_state_import(
2745            input_file=input_file,
2746            gateway=self.selected_gateway,
2747            state_connection_config=self._state_connection_config,
2748            clear=clear,
2749            confirm=confirm,
2750        ):
2751            try:
2752                import_state(
2753                    state_sync=self.state_sync,
2754                    input_file=input_file,
2755                    clear=clear,
2756                    console=self.console,
2757                )
2758                self.console.stop_state_import(success=True, input_file=input_file)
2759            except:
2760                self.console.stop_state_import(success=False, input_file=input_file)
2761                raise
2762
2763    def _run_tests(
2764        self, verbosity: Verbosity = Verbosity.DEFAULT
2765    ) -> t.Tuple[ModelTextTestResult, str]:
2766        test_output_io = StringIO()
2767        result = self.test(stream=test_output_io, verbosity=verbosity)
2768        return result, test_output_io.getvalue()
2769
2770    def _run_plan_tests(self, skip_tests: bool = False) -> t.Optional[ModelTextTestResult]:
2771        if not skip_tests:
2772            result = self.test()
2773            if not result.wasSuccessful():
2774                raise PlanError(
2775                    "Cannot generate plan due to failing test(s). Fix test(s) and run again."
2776                )
2777            return result
2778        return None
2779
2780    def _warn_if_virtual_catalog_rematerialization(self, plan: "Plan") -> None:
2781        """Warn when ClickHouse models appear as new snapshots solely because a virtual catalog
2782        prefix was added to their FQNs after a catalog-aware gateway joined the project.
2783
2784        This situation causes every previously-applied ClickHouse model to be treated as brand-new
2785        by SQLMesh, triggering full re-materialization and historical backfills. Emitting a warning
2786        before the plan is displayed gives users a chance to understand the cost before applying.
2787        """
2788        from sqlglot import exp
2789
2790        # Collect the set of old 2-level snapshot names from the current environment so we can
2791        # detect which new 3-level names are renames rather than genuinely new models.
2792        old_names: t.Set[str] = set()
2793        for s_id in plan.context_diff.removed_snapshots:
2794            old_names.add(s_id.name)
2795        for name in plan.context_diff.snapshots_by_name:
2796            old_names.add(name)
2797
2798        affected: t.List[t.Tuple[str, str]] = []  # (new_3level_name, old_2level_name)
2799
2800        for gateway, adapter in self.engine_adapters.items():
2801            if not adapter.supports_virtual_catalog() or not adapter._default_catalog:
2802                continue
2803            virtual_catalog = adapter._default_catalog
2804
2805            for snapshot in plan.new_snapshots:
2806                table = exp.to_table(snapshot.name)
2807                if table.catalog != virtual_catalog:
2808                    continue
2809                # Reconstruct the 2-level name that would have been used before injection.
2810                old_name = f"{table.db}.{table.name}"
2811                if old_name in old_names:
2812                    affected.append((snapshot.name, old_name))
2813
2814        if not affected:
2815            return
2816
2817        max_display = 10
2818        model_lines = "\n".join(
2819            f"  - {new_name}  (was: {old_name})" for new_name, old_name in affected[:max_display]
2820        )
2821        if len(affected) > max_display:
2822            model_lines += f"\n  ... and {len(affected) - max_display} more"
2823
2824        self.console.log_warning(
2825            "ClickHouse models are being re-materialized due to virtual catalog FQN change.\n\n"
2826            "The following ClickHouse models appear as new because their fully-qualified\n"
2827            "names changed from 2-level (db.table) to 3-level (__gateway__.db.table):\n\n"
2828            f"{model_lines}\n\n"
2829            "FULL models will be recreated once. INCREMENTAL_BY_TIME_RANGE models will\n"
2830            "require a full historical backfill from their configured start date.\n\n"
2831            "This is a one-time cost when first adding a catalog-aware gateway to an\n"
2832            "existing ClickHouse project. To proceed, run `sqlmesh apply`."
2833        )
2834
2835    @property
2836    def _model_tables(self) -> t.Dict[str, str]:
2837        """Mapping of model name to physical table name.
2838
2839        If a snapshot has not been versioned yet, its view name will be returned.
2840        """
2841        return {
2842            fqn: (
2843                snapshot.table_name()
2844                if snapshot.version
2845                else snapshot.qualified_view_name.for_environment(
2846                    EnvironmentNamingInfo.from_environment_catalog_mapping(
2847                        self.environment_catalog_mapping,
2848                        name=c.PROD,
2849                        suffix_target=self.config.environment_suffix_target,
2850                    )
2851                )
2852            )
2853            for fqn, snapshot in self.snapshots.items()
2854        }
2855
2856    @cached_property
2857    def cache_dir(self) -> Path:
2858        if self.config.cache_dir:
2859            cache_path = Path(self.config.cache_dir)
2860            if cache_path.is_absolute():
2861                return cache_path
2862            return self.path / cache_path
2863
2864        # Default to .cache directory in the project path
2865        return self.path / c.CACHE
2866
2867    @cached_property
2868    def engine_adapters(self) -> t.Dict[str, EngineAdapter]:
2869        """Returns all the engine adapters for the gateways defined in the configurations."""
2870        adapters: t.Dict[str, EngineAdapter] = {self.selected_gateway: self.engine_adapter}
2871        for config in self.configs.values():
2872            for gateway_name in config.gateways:
2873                if gateway_name not in adapters:
2874                    connection = config.get_connection(gateway_name)
2875                    adapter = connection.create_engine_adapter(
2876                        concurrent_tasks=self.concurrent_tasks,
2877                    )
2878                    adapters[gateway_name] = adapter
2879        return adapters
2880
2881    @cached_property
2882    def default_catalog_per_gateway(self) -> t.Dict[str, str]:
2883        """Returns the default catalogs for each engine adapter."""
2884        return self._scheduler.get_default_catalog_per_gateway(self)
2885
2886    @property
2887    def concurrent_tasks(self) -> int:
2888        if self._concurrent_tasks is None:
2889            self._concurrent_tasks = self.connection_config.concurrent_tasks
2890        return self._concurrent_tasks
2891
2892    @cached_property
2893    def connection_config(self) -> ConnectionConfig:
2894        return self.config.get_connection(self.selected_gateway)
2895
2896    @cached_property
2897    def test_connection_config(self) -> ConnectionConfig:
2898        return self.config.get_test_connection(
2899            self.gateway,
2900            self.default_catalog,
2901            default_catalog_dialect=self.config.dialect,
2902        )
2903
2904    @cached_property
2905    def environment_catalog_mapping(self) -> RegexKeyDict:
2906        engine_adapter = None
2907        try:
2908            engine_adapter = self.engine_adapter
2909        except Exception:
2910            pass
2911
2912        if (
2913            self.config.environment_catalog_mapping
2914            and engine_adapter
2915            and not self.engine_adapter.catalog_support.is_multi_catalog_supported
2916        ):
2917            raise SQLMeshError(
2918                "Environment catalog mapping is only supported for engine adapters that support multiple catalogs"
2919            )
2920        return self.config.environment_catalog_mapping
2921
2922    def _get_engine_adapter(self, gateway: t.Optional[str] = None) -> EngineAdapter:
2923        if gateway:
2924            if adapter := self.engine_adapters.get(gateway):
2925                return adapter
2926            raise SQLMeshError(f"Gateway '{gateway}' not found in the available engine adapters.")
2927        return self.engine_adapter
2928
2929    def _snapshots(
2930        self, models_override: t.Optional[UniqueKeyDict[str, Model]] = None
2931    ) -> t.Dict[str, Snapshot]:
2932        nodes = {**(models_override or self._models), **self._standalone_audits}
2933        snapshots = self._nodes_to_snapshots(nodes)
2934        stored_snapshots = self.state_reader.get_snapshots(snapshots.values())
2935
2936        unrestorable_snapshots = {
2937            snapshot
2938            for snapshot in stored_snapshots.values()
2939            if snapshot.name in nodes and snapshot.unrestorable
2940        }
2941        if unrestorable_snapshots:
2942            for snapshot in unrestorable_snapshots:
2943                logger.info(
2944                    "Found a unrestorable snapshot %s. Restamping the model...", snapshot.name
2945                )
2946                node = nodes[snapshot.name]
2947                nodes[snapshot.name] = node.copy(
2948                    update={"stamp": f"revert to {snapshot.identifier}"}
2949                )
2950            snapshots = self._nodes_to_snapshots(nodes)
2951            stored_snapshots = self.state_reader.get_snapshots(snapshots.values())
2952
2953        for snapshot in stored_snapshots.values():
2954            # Keep the original model instance to preserve the query cache.
2955            snapshot.node = snapshots[snapshot.name].node
2956
2957        return {name: stored_snapshots.get(s.snapshot_id, s) for name, s in snapshots.items()}
2958
2959    def _context_diff(
2960        self,
2961        environment: str,
2962        snapshots: t.Optional[t.Dict[str, Snapshot]] = None,
2963        create_from: t.Optional[str] = None,
2964        force_no_diff: bool = False,
2965        ensure_finalized_snapshots: bool = False,
2966        diff_rendered: bool = False,
2967        always_recreate_environment: bool = False,
2968    ) -> ContextDiff:
2969        environment = Environment.sanitize_name(environment)
2970        if force_no_diff:
2971            return ContextDiff.create_no_diff(environment, self.state_reader)
2972
2973        return ContextDiff.create(
2974            environment,
2975            snapshots=snapshots or self.snapshots,
2976            create_from=create_from or c.PROD,
2977            state_reader=self.state_reader,
2978            provided_requirements=self._requirements,
2979            excluded_requirements=self._excluded_requirements,
2980            ensure_finalized_snapshots=ensure_finalized_snapshots,
2981            diff_rendered=diff_rendered,
2982            environment_statements=self._environment_statements,
2983            gateway_managed_virtual_layer=self.config.gateway_managed_virtual_layer,
2984            infer_python_dependencies=self.config.infer_python_dependencies,
2985            always_recreate_environment=always_recreate_environment,
2986        )
2987
2988    def _destroy(self) -> bool:
2989        # Invalidate all environments, including prod
2990        for environment in self.state_reader.get_environments():
2991            self.state_sync.invalidate_environment(name=environment.name, protect_prod=False)
2992            self.console.log_success(f"Environment '{environment.name}' invalidated.")
2993
2994        # Run janitor to clean up all objects
2995        self._run_janitor(ignore_ttl=True)
2996
2997        # Remove state tables, including backup tables
2998        self.state_sync.remove_state(including_backup=True)
2999        self.console.log_status_update("State tables removed.")
3000
3001        # Finally clear caches
3002        self.clear_caches()
3003
3004        return True
3005
3006    def _run_janitor(
3007        self,
3008        ignore_ttl: bool = False,
3009        force_delete: bool = False,
3010        environment: t.Optional[str] = None,
3011    ) -> None:
3012        current_ts = now_timestamp()
3013        failures: t.List[str] = []
3014
3015        # Clean up expired environments by removing their views and schemas
3016        failures.extend(
3017            self._cleanup_environments(
3018                current_ts=current_ts, force_delete=force_delete, name=environment
3019            )
3020        )
3021
3022        if environment is None:
3023            failures.extend(
3024                delete_expired_snapshots(
3025                    self.state_sync,
3026                    self.snapshot_evaluator,
3027                    current_ts=current_ts,
3028                    ignore_ttl=ignore_ttl,
3029                    force_delete=force_delete,
3030                    console=self.console,
3031                    batch_size=self.config.janitor.expired_snapshots_batch_size,
3032                )
3033            )
3034            self.state_sync.compact_intervals()
3035
3036        if failures:
3037            failure_string = "\n  - ".join(failures)
3038            summary = f"Janitor completed with failures:\n  {failure_string}"
3039            if force_delete:
3040                summary += "\nState records have been deleted, but the underlying objects may still exist in the database.\nPlease investigate and clean up manually the above if necessary."
3041            if self.config.janitor.warn_on_delete_failure:
3042                self.console.log_warning(summary)
3043            else:
3044                raise SQLMeshError(summary)
3045
3046    def _cleanup_environments(
3047        self,
3048        current_ts: t.Optional[int] = None,
3049        force_delete: bool = False,
3050        name: t.Optional[str] = None,
3051    ) -> t.List[str]:
3052        current_ts = current_ts or now_timestamp()
3053        failures: t.List[str] = []
3054
3055        expired_environments_summaries = self.state_sync.get_expired_environments(
3056            current_ts=current_ts, name=name
3057        )
3058
3059        if name is not None and not expired_environments_summaries:
3060            self.console.log_warning(
3061                f"Environment '{name}' is not expired or does not exist. Nothing to clean up."
3062            )
3063
3064        for expired_env_summary in expired_environments_summaries:
3065            expired_env = self.state_reader.get_environment(expired_env_summary.name)
3066
3067            if expired_env:
3068                failures.extend(
3069                    cleanup_expired_views(
3070                        default_adapter=self.engine_adapter,
3071                        engine_adapters=self.engine_adapters,
3072                        environments=[expired_env],
3073                        console=self.console,
3074                    )
3075                )
3076
3077        # we want to retry on the next janitor pass if drops failed, unless
3078        # force_delete is set in which case we purge state records regardless
3079        if not failures or force_delete:
3080            self.state_sync.delete_expired_environments(current_ts=current_ts, name=name)
3081        return failures
3082
3083    def _try_connection(self, connection_name: str, validator: t.Callable[[], None]) -> None:
3084        connection_name = connection_name.capitalize()
3085        try:
3086            validator()
3087            self.console.log_status_update(f"{connection_name} connection [green]succeeded[/green]")
3088        except Exception as ex:
3089            self.console.log_error(f"{connection_name} connection failed. {ex}")
3090
3091    def _new_state_sync(self) -> StateSync:
3092        return self._provided_state_sync or self._scheduler.create_state_sync(self)
3093
3094    def _new_selector(
3095        self, models: t.Optional[UniqueKeyDict[str, Model]] = None, dag: t.Optional[DAG[str]] = None
3096    ) -> Selector:
3097        return self._selector_cls(
3098            self.state_reader,
3099            models=models or self._models,
3100            context_path=self.path,
3101            dag=dag,
3102            default_catalog=self.default_catalog,
3103            dialect=self.default_dialect,
3104            cache_dir=self.cache_dir,
3105        )
3106
3107    def _register_notification_targets(self) -> None:
3108        event_notifications = collections.defaultdict(set)
3109        for target in self.notification_targets:
3110            if target.is_configured:
3111                for event in target.notify_on:
3112                    event_notifications[event].add(target)
3113        user_notification_targets = {
3114            user.username: set(
3115                target for target in user.notification_targets if target.is_configured
3116            )
3117            for user in self.users
3118        }
3119        self.notification_target_manager = NotificationTargetManager(
3120            event_notifications, user_notification_targets, username=self.config.username
3121        )
3122
3123    def _load_materializations(self) -> None:
3124        if not self._loaded:
3125            for loader in self._loaders:
3126                loader.load_materializations()
3127
3128    def _select_models_for_run(
3129        self,
3130        select_models: t.Collection[str],
3131        no_auto_upstream: bool,
3132        snapshots: t.Collection[Snapshot],
3133    ) -> t.Set[str]:
3134        models: UniqueKeyDict[str, Model] = UniqueKeyDict(
3135            "models", **{s.name: s.model for s in snapshots if s.is_model}
3136        )
3137        dag: DAG[str] = DAG()
3138        for fqn, model in models.items():
3139            dag.add(fqn, model.depends_on)
3140        model_selector = self._new_selector(models=models, dag=dag)
3141        result = set(model_selector.expand_model_selections(select_models))
3142        if not no_auto_upstream:
3143            result = set(dag.subdag(*result))
3144        return result
3145
3146    @cached_property
3147    def _project_type(self) -> str:
3148        project_types = {
3149            c.DBT if loader.__class__.__name__.lower().startswith(c.DBT) else c.NATIVE
3150            for loader in self._loaders
3151        }
3152        return c.HYBRID if len(project_types) > 1 else first(project_types)
3153
3154    def _nodes_to_snapshots(self, nodes: t.Dict[str, Node]) -> t.Dict[str, Snapshot]:
3155        snapshots: t.Dict[str, Snapshot] = {}
3156        fingerprint_cache: t.Dict[str, SnapshotFingerprint] = {}
3157
3158        for node in nodes.values():
3159            kwargs: t.Dict[str, t.Any] = {}
3160            if node.project in self._projects:
3161                config = self.config_for_node(node)
3162                kwargs["ttl"] = config.snapshot_ttl
3163                kwargs["table_naming_convention"] = config.physical_table_naming_convention
3164
3165            snapshot = Snapshot.from_node(
3166                node,
3167                nodes=nodes,
3168                cache=fingerprint_cache,
3169                **kwargs,
3170            )
3171            snapshots[snapshot.name] = snapshot
3172        return snapshots
3173
3174    def _node_or_snapshot_to_fqn(self, node_or_snapshot: NodeOrSnapshot) -> str:
3175        if isinstance(node_or_snapshot, Snapshot):
3176            return node_or_snapshot.name
3177        if isinstance(node_or_snapshot, str) and not self.standalone_audits.get(node_or_snapshot):
3178            return normalize_model_name(
3179                node_or_snapshot,
3180                dialect=self.default_dialect,
3181                default_catalog=self.default_catalog,
3182            )
3183        if not isinstance(node_or_snapshot, str):
3184            return node_or_snapshot.fqn
3185        return node_or_snapshot
3186
3187    @property
3188    def _plan_preview_enabled(self) -> bool:
3189        if self.config.plan.enable_preview is not None:
3190            return self.config.plan.enable_preview
3191        # It is dangerous to enable preview by default for dbt projects that rely on engines that don't support cloning.
3192        # Enabling previews in such cases can result in unintended full refreshes because dbt incremental models rely on
3193        # the maximum timestamp value in the target table.
3194        return self._project_type == c.NATIVE or self.engine_adapter.SUPPORTS_CLONING
3195
3196    def _get_plan_default_start_end(
3197        self,
3198        snapshots: t.Dict[str, Snapshot],
3199        max_interval_end_per_model: t.Dict[str, datetime],
3200        backfill_models: t.Optional[t.Set[str]],
3201        modified_model_names: t.Set[str],
3202        execution_time: t.Optional[TimeLike] = None,
3203    ) -> t.Tuple[t.Optional[int], t.Optional[int]]:
3204        # exclude seeds so their stale interval ends does not become the default plan end date
3205        # when they're the only ones that contain intervals in this plan
3206        non_seed_interval_ends = {
3207            model_fqn: end
3208            for model_fqn, end in max_interval_end_per_model.items()
3209            if model_fqn not in snapshots or not snapshots[model_fqn].is_seed
3210        }
3211        if not non_seed_interval_ends:
3212            return None, None
3213
3214        default_end = to_timestamp(max(non_seed_interval_ends.values()))
3215        default_start: t.Optional[int] = None
3216        # Infer the default start by finding the smallest interval start that corresponds to the default end.
3217        for model_name in backfill_models or modified_model_names or max_interval_end_per_model:
3218            if model_name not in snapshots:
3219                continue
3220            node = snapshots[model_name].node
3221            interval_unit = node.interval_unit
3222            default_start = min(
3223                default_start or sys.maxsize,
3224                to_timestamp(
3225                    interval_unit.cron_prev(
3226                        interval_unit.cron_floor(
3227                            max_interval_end_per_model.get(
3228                                model_name, node.cron_floor(default_end)
3229                            ),
3230                        ),
3231                        estimate=True,
3232                    )
3233                ),
3234            )
3235
3236        if execution_time and to_timestamp(default_end) > to_timestamp(execution_time):
3237            # the end date can't be in the future, which can happen if a specific `execution_time` is set and prod intervals
3238            # are newer than it
3239            default_end = to_timestamp(execution_time)
3240
3241        return default_start, default_end
3242
3243    def _calculate_start_override_per_model(
3244        self,
3245        min_intervals: t.Optional[int],
3246        plan_start: t.Optional[TimeLike],
3247        plan_end: t.Optional[TimeLike],
3248        plan_execution_time: TimeLike,
3249        backfill_model_fqns: t.Optional[t.Set[str]],
3250        snapshots_by_model_fqn: t.Dict[str, Snapshot],
3251        end_override_per_model: t.Optional[t.Dict[str, datetime]],
3252    ) -> t.Dict[str, datetime]:
3253        if not min_intervals or not backfill_model_fqns or not plan_start:
3254            # If there are no models to backfill, there are no intervals to consider for backfill, so we dont need to consider a minimum number
3255            # If the plan doesnt have a start date, all intervals are considered already so we dont need to consider a minimum number
3256            # If we dont have a minimum number of intervals to consider, then we dont need to adjust the start date on a per-model basis
3257            return {}
3258
3259        start_overrides: t.Dict[str, datetime] = {}
3260        end_override_per_model = end_override_per_model or {}
3261
3262        plan_execution_time_dt = to_datetime(plan_execution_time)
3263        plan_start_dt = to_datetime(plan_start, relative_base=plan_execution_time_dt)
3264        plan_end_dt = to_datetime(
3265            plan_end or plan_execution_time_dt, relative_base=plan_execution_time_dt
3266        )
3267
3268        # we need to take the DAG into account so that parent models can be expanded to cover at least as much as their children
3269        # for example, A(hourly) <- B(daily)
3270        # if min_intervals=1, A would have 1 hour and B would have 1 day
3271        # but B depends on A so in order for B to have 1 valid day, A needs to be expanded to 24 hours
3272        backfill_dag: DAG[str] = DAG()
3273        for fqn in backfill_model_fqns:
3274            backfill_dag.add(
3275                fqn,
3276                [
3277                    p.name
3278                    for p in snapshots_by_model_fqn[fqn].parents
3279                    if p.name in backfill_model_fqns
3280                ],
3281            )
3282
3283        # start from the leaf nodes and work back towards the root because the min_start at the root node is determined by the calculated starts in the leaf nodes
3284        reversed_dag = backfill_dag.reversed
3285        graph = reversed_dag.graph
3286
3287        for model_fqn in reversed_dag:
3288            # Get the earliest start from all immediate children of this snapshot
3289            # this works because topological ordering guarantees that they've already been visited
3290            # and we always set a start override
3291            min_child_start = min(
3292                [start_overrides[immediate_child_fqn] for immediate_child_fqn in graph[model_fqn]],
3293                default=plan_start_dt,
3294            )
3295
3296            snapshot = snapshots_by_model_fqn.get(model_fqn)
3297
3298            if not snapshot:
3299                continue
3300
3301            starting_point = end_override_per_model.get(model_fqn, plan_end_dt)
3302            if node_end := snapshot.node.end:
3303                # if we dont do this, if the node end is a *date* (as opposed to a timestamp)
3304                # we end up incorrectly winding back an extra day
3305                node_end_dt = make_exclusive(node_end)
3306
3307                if node_end_dt < plan_end_dt:
3308                    # if the model has an end date that has already elapsed, use that as a starting point for calculating min_intervals
3309                    # instead of the plan end. If we use the plan end, we will return intervals in the future which are invalid
3310                    starting_point = node_end_dt
3311
3312            snapshot_start = snapshot.node.cron_floor(starting_point)
3313
3314            for _ in range(min_intervals):
3315                # wind back the starting point by :min_intervals intervals to arrive at the minimum snapshot start date
3316                snapshot_start = snapshot.node.cron_prev(snapshot_start)
3317
3318            start_overrides[model_fqn] = min(min_child_start, snapshot_start)
3319
3320        return start_overrides
3321
3322    def _get_max_interval_end_per_model(
3323        self, snapshots: t.Dict[str, Snapshot], backfill_models: t.Optional[t.Set[str]]
3324    ) -> t.Dict[str, datetime]:
3325        models_for_interval_end = (
3326            self._get_models_for_interval_end(snapshots, backfill_models)
3327            if backfill_models is not None
3328            else None
3329        )
3330        return {
3331            model_fqn: to_datetime(ts)
3332            for model_fqn, ts in self.state_sync.max_interval_end_per_model(
3333                c.PROD,
3334                models=models_for_interval_end,
3335                ensure_finalized_snapshots=self.config.plan.use_finalized_state,
3336            ).items()
3337        }
3338
3339    @staticmethod
3340    def _get_models_for_interval_end(
3341        snapshots: t.Dict[str, Snapshot], backfill_models: t.Set[str]
3342    ) -> t.Set[str]:
3343        models_for_interval_end = set()
3344        models_stack = list(backfill_models)
3345        while models_stack:
3346            next_model = models_stack.pop()
3347            if next_model not in snapshots:
3348                continue
3349            models_for_interval_end.add(next_model)
3350            models_stack.extend(
3351                s.name
3352                for s in snapshots[next_model].parents
3353                if s.name not in models_for_interval_end
3354            )
3355        return models_for_interval_end
3356
3357    def lint_models(
3358        self,
3359        models: t.Optional[t.Iterable[t.Union[str, Model]]] = None,
3360        raise_on_error: bool = True,
3361    ) -> t.List[AnnotatedRuleViolation]:
3362        found_error = False
3363
3364        model_list = (
3365            list(self.get_model(model, raise_if_missing=True) for model in models)
3366            if models
3367            else self.models.values()
3368        )
3369        all_violations = []
3370        for model in model_list:
3371            # Linter may be `None` if the context is not loaded yet
3372            if linter := self._linters.get(model.project):
3373                lint_violation, violations = (
3374                    linter.lint_model(model, self, console=self.console) or found_error
3375                )
3376                if lint_violation:
3377                    found_error = True
3378                all_violations.extend(violations)
3379
3380        if raise_on_error and found_error:
3381            raise LinterError(
3382                "Linter detected errors in the code. Please fix them before proceeding."
3383            )
3384
3385        return all_violations
3386
3387    def select_tests(
3388        self,
3389        tests: t.Optional[t.List[str]] = None,
3390        patterns: t.Optional[t.List[str]] = None,
3391    ) -> t.List[ModelTestMetadata]:
3392        """Filter pre-loaded test metadata based on tests and patterns."""
3393
3394        test_meta = self._model_test_metadata
3395
3396        if tests:
3397            filtered_tests = []
3398            for test in tests:
3399                if "::" in test:
3400                    if test in self._model_test_metadata_fully_qualified_name_index:
3401                        filtered_tests.append(
3402                            self._model_test_metadata_fully_qualified_name_index[test]
3403                        )
3404                else:
3405                    test_path = Path(test)
3406                    if test_path in self._model_test_metadata_path_index:
3407                        filtered_tests.extend(self._model_test_metadata_path_index[test_path])
3408
3409            test_meta = filtered_tests
3410
3411        if patterns:
3412            test_meta = filter_tests_by_patterns(test_meta, patterns)
3413
3414        return test_meta
3415
3416
3417class Context(GenericContext[Config]):
3418    CONFIG_TYPE = Config