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            # Refresh snapshot intervals to ensure that they are up to date with values reflected in the max_interval_end_per_model.
1725            self.state_sync.refresh_snapshot_intervals(context_diff.snapshots.values())
1726
1727        start_override_per_model = self._calculate_start_override_per_model(
1728            min_intervals,
1729            start or default_start,
1730            end or default_end,
1731            execution_time or now(),
1732            backfill_models,
1733            snapshots,
1734            max_interval_end_per_model,
1735        )
1736
1737        if not self.config.virtual_environment_mode.is_full:
1738            forward_only = True
1739        elif forward_only is None:
1740            forward_only = self.config.plan.forward_only
1741
1742        # When handling prod restatements, only clear intervals from other model versions if we are using full virtual environments
1743        # If we are not, then there is no point, because none of the data in dev environments can be promoted by definition
1744        restate_all_snapshots = (
1745            expanded_restate_models is not None
1746            and not is_dev
1747            and self.config.virtual_environment_mode.is_full
1748        )
1749
1750        return self.PLAN_BUILDER_TYPE(
1751            context_diff=context_diff,
1752            start=start,
1753            end=end,
1754            execution_time=execution_time,
1755            apply=self.apply,
1756            restate_models=expanded_restate_models,
1757            restate_all_snapshots=restate_all_snapshots,
1758            backfill_models=backfill_models,
1759            no_gaps=no_gaps,
1760            skip_backfill=skip_backfill,
1761            empty_backfill=empty_backfill,
1762            is_dev=is_dev,
1763            forward_only=forward_only,
1764            allow_destructive_models=expanded_destructive_models,
1765            allow_additive_models=expanded_additive_models,
1766            environment_ttl=environment_ttl,
1767            environment_suffix_target=self.config.environment_suffix_target,
1768            environment_catalog_mapping=self.environment_catalog_mapping,
1769            categorizer_config=categorizer_config or self.auto_categorize_changes,
1770            auto_categorization_enabled=not no_auto_categorization,
1771            effective_from=effective_from,
1772            include_unmodified=include_unmodified,
1773            default_start=default_start,
1774            default_end=default_end,
1775            enable_preview=(
1776                enable_preview if enable_preview is not None else self._plan_preview_enabled
1777            ),
1778            preview_start=preview_start,
1779            preview_min_intervals=preview_min_intervals or 0,
1780            end_bounded=not run,
1781            ensure_finalized_snapshots=self.config.plan.use_finalized_state,
1782            start_override_per_model=start_override_per_model,
1783            end_override_per_model=max_interval_end_per_model,
1784            console=self.console,
1785            user_provided_flags=user_provided_flags,
1786            selected_models={
1787                dbt_unique_id
1788                for model in model_selector.expand_model_selections(select_models or "*")
1789                if (dbt_unique_id := snapshots[model].node.dbt_unique_id)
1790            },
1791            explain=explain or False,
1792            ignore_cron=ignore_cron or False,
1793        )
1794
1795    def apply(
1796        self,
1797        plan: Plan,
1798        circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
1799    ) -> None:
1800        """Applies a plan by pushing snapshots and backfilling data.
1801
1802        Given a plan, it pushes snapshots into the state sync and then uses the scheduler
1803        to backfill all models.
1804
1805        Args:
1806            plan: The plan to apply.
1807            circuit_breaker: An optional handler which checks if the apply should be aborted.
1808        """
1809        if (
1810            not plan.context_diff.has_changes
1811            and not plan.requires_backfill
1812            and not plan.has_unmodified_unpromoted
1813        ):
1814            return
1815        if plan.uncategorized:
1816            raise UncategorizedPlanError("Can't apply a plan with uncategorized changes.")
1817
1818        if plan.explain:
1819            explainer = PlanExplainer(
1820                state_reader=self.state_reader,
1821                default_catalog=self.default_catalog,
1822                console=self.console,
1823            )
1824            explainer.evaluate(plan.to_evaluatable())
1825            return
1826
1827        self.notification_target_manager.notify(
1828            NotificationEvent.APPLY_START,
1829            environment=plan.environment_naming_info.name,
1830            plan_id=plan.plan_id,
1831        )
1832        try:
1833            self._apply(plan, circuit_breaker)
1834        except Exception as e:
1835            self.notification_target_manager.notify(
1836                NotificationEvent.APPLY_FAILURE,
1837                environment=plan.environment_naming_info.name,
1838                plan_id=plan.plan_id,
1839                exc=traceback.format_exc(),
1840            )
1841            logger.info("Plan application failed.", exc_info=e)
1842            raise e
1843        self.notification_target_manager.notify(
1844            NotificationEvent.APPLY_END,
1845            environment=plan.environment_naming_info.name,
1846            plan_id=plan.plan_id,
1847        )
1848
1849    @python_api_analytics
1850    def invalidate_environment(self, name: str, sync: bool = False) -> None:
1851        """Invalidates the target environment by setting its expiration timestamp to now.
1852
1853        Args:
1854            name: The name of the environment to invalidate.
1855            sync: If True, the call blocks until the environment is deleted. Otherwise, the environment will
1856                be deleted asynchronously by the janitor process.
1857        """
1858        name = Environment.sanitize_name(name)
1859        self.state_sync.invalidate_environment(name)
1860        if sync:
1861            self._cleanup_environments(name=name)
1862            self.console.log_success(f"Environment '{name}' deleted.")
1863        else:
1864            self.console.log_success(f"Environment '{name}' invalidated.")
1865
1866    @python_api_analytics
1867    def diff(self, environment: t.Optional[str] = None, detailed: bool = False) -> bool:
1868        """Show a diff of the current context with a given environment.
1869
1870        Args:
1871            environment: The environment to diff against.
1872            detailed: Show the actual SQL differences if True.
1873
1874        Returns:
1875            True if there are changes, False otherwise.
1876        """
1877        environment = environment or self.config.default_target_environment
1878        environment = Environment.sanitize_name(environment)
1879        context_diff = self._context_diff(environment)
1880        self.console.show_environment_difference_summary(
1881            context_diff,
1882            no_diff=not detailed,
1883        )
1884        if context_diff.has_changes:
1885            self.console.show_model_difference_summary(
1886                context_diff,
1887                EnvironmentNamingInfo.from_environment_catalog_mapping(
1888                    self.environment_catalog_mapping,
1889                    name=environment,
1890                    suffix_target=self.config.environment_suffix_target,
1891                    normalize_name=context_diff.normalize_environment_name,
1892                ),
1893                self.default_catalog,
1894                no_diff=not detailed,
1895            )
1896        return context_diff.has_changes
1897
1898    @python_api_analytics
1899    def table_diff(
1900        self,
1901        source: str,
1902        target: str,
1903        on: t.Optional[t.List[str] | exp.Expr] = None,
1904        skip_columns: t.Optional[t.List[str]] = None,
1905        select_models: t.Optional[t.Collection[str]] = None,
1906        where: t.Optional[str | exp.Expr] = None,
1907        limit: int = 20,
1908        show: bool = True,
1909        show_sample: bool = True,
1910        decimals: int = 3,
1911        skip_grain_check: bool = False,
1912        warn_grain_check: bool = False,
1913        temp_schema: t.Optional[str] = None,
1914        schema_diff_ignore_case: bool = False,
1915        **kwargs: t.Any,  # catch-all to prevent an 'unexpected keyword argument' error if an table_diff extension passes in some extra arguments
1916    ) -> t.List[TableDiff]:
1917        """Show a diff between two tables.
1918
1919        Args:
1920            source: The source environment or table.
1921            target: The target environment or table.
1922            on: The join condition, table aliases must be "s" and "t" for source and target.
1923                If omitted, the table's grain will be used.
1924            skip_columns: The columns to skip when computing the table diff.
1925            select_models: The models or snapshots to use when environments are passed in.
1926            where: An optional where statement to filter results.
1927            limit: The limit of the sample dataframe.
1928            show: Show the table diff output in the console.
1929            show_sample: Show the sample dataframe in the console. Requires show=True.
1930            decimals: The number of decimal places to keep when comparing floating point columns.
1931            skip_grain_check: Skip check for rows that contain null or duplicate grains.
1932            temp_schema: The schema to use for temporary tables.
1933
1934        Returns:
1935            The list of TableDiff objects containing schema and summary differences.
1936        """
1937
1938        if "|" in source or "|" in target:
1939            raise ConfigError(
1940                "Cross-database table diffing is available in Tobiko Cloud. Read more here: "
1941                "https://sqlmesh.readthedocs.io/en/stable/guides/tablediff/#diffing-tables-or-views-across-gateways"
1942            )
1943
1944        table_diffs: t.List[TableDiff] = []
1945
1946        # Diffs multiple or a single model across two environments
1947        if select_models:
1948            source_env = self.state_reader.get_environment(source)
1949            target_env = self.state_reader.get_environment(target)
1950            if not source_env:
1951                raise SQLMeshError(f"Could not find environment '{source}'")
1952            if not target_env:
1953                raise SQLMeshError(f"Could not find environment '{target}'")
1954            criteria = ", ".join(f"'{c}'" for c in select_models)
1955            try:
1956                selected_models = self._new_selector().expand_model_selections(select_models)
1957                if not selected_models:
1958                    self.console.log_status_update(
1959                        f"No models matched the selection criteria: {criteria}"
1960                    )
1961            except Exception as e:
1962                raise SQLMeshError(e)
1963
1964            models_to_diff: t.List[
1965                t.Tuple[Model, EngineAdapter, str, str, t.Optional[t.List[str] | exp.Expr]]
1966            ] = []
1967            models_without_grain: t.List[Model] = []
1968            source_snapshots_to_name = {
1969                snapshot.name: snapshot for snapshot in source_env.snapshots
1970            }
1971            target_snapshots_to_name = {
1972                snapshot.name: snapshot for snapshot in target_env.snapshots
1973            }
1974
1975            for model_fqn in selected_models:
1976                model = self._models[model_fqn]
1977                adapter = self._get_engine_adapter(model.gateway)
1978                source_snapshot = source_snapshots_to_name.get(model.fqn)
1979                target_snapshot = target_snapshots_to_name.get(model.fqn)
1980
1981                if target_snapshot and source_snapshot:
1982                    if (source_snapshot.fingerprint != target_snapshot.fingerprint) and (
1983                        (source_snapshot.version != target_snapshot.version)
1984                        or source_snapshot.is_forward_only
1985                    ):
1986                        # Compare the virtual layer instead of the physical layer because the virtual layer is guaranteed to point
1987                        # to the correct/active snapshot for the model in the specified environment, taking into account things like dev previews
1988                        source = source_snapshot.qualified_view_name.for_environment(
1989                            source_env.naming_info, adapter.dialect
1990                        )
1991                        target = target_snapshot.qualified_view_name.for_environment(
1992                            target_env.naming_info, adapter.dialect
1993                        )
1994                        model_on = on or model.on
1995                        if not model_on:
1996                            models_without_grain.append(model)
1997                        else:
1998                            models_to_diff.append((model, adapter, source, target, model_on))
1999
2000            if models_without_grain:
2001                model_names = "\n".join(
2002                    f"─ {model.name} \n  at '{model._path}'" for model in models_without_grain
2003                )
2004                message = (
2005                    "SQLMesh doesn't know how to join the tables for the following models:\n"
2006                    f"{model_names}\n\n"
2007                    "Please specify a `grain` in each model definition. It must be unique and not null."
2008                )
2009                if warn_grain_check:
2010                    self.console.log_warning(message)
2011                else:
2012                    raise SQLMeshError(message)
2013
2014            if models_to_diff:
2015                self.console.show_table_diff_details(
2016                    [model[0].name for model in models_to_diff],
2017                )
2018
2019                self.console.start_table_diff_progress(len(models_to_diff))
2020                try:
2021                    tasks_num = min(len(models_to_diff), self.concurrent_tasks)
2022                    table_diffs = concurrent_apply_to_values(
2023                        list(models_to_diff),
2024                        lambda model_info: self._model_diff(
2025                            model=model_info[0],
2026                            adapter=model_info[1],
2027                            source=model_info[2],
2028                            target=model_info[3],
2029                            on=model_info[4],
2030                            source_alias=source_env.name,
2031                            target_alias=target_env.name,
2032                            limit=limit,
2033                            decimals=decimals,
2034                            skip_columns=skip_columns,
2035                            where=where,
2036                            show=show,
2037                            temp_schema=temp_schema,
2038                            skip_grain_check=skip_grain_check,
2039                            schema_diff_ignore_case=schema_diff_ignore_case,
2040                        ),
2041                        tasks_num=tasks_num,
2042                    )
2043                    self.console.stop_table_diff_progress(success=True)
2044                except:
2045                    self.console.stop_table_diff_progress(success=False)
2046                    raise
2047            elif selected_models:
2048                self.console.log_status_update(
2049                    f"No models contain differences with the selection criteria: {criteria}"
2050                )
2051
2052        else:
2053            table_diffs = [
2054                self._table_diff(
2055                    source=source,
2056                    target=target,
2057                    source_alias=source,
2058                    target_alias=target,
2059                    limit=limit,
2060                    decimals=decimals,
2061                    adapter=self.engine_adapter,
2062                    on=on,
2063                    skip_columns=skip_columns,
2064                    where=where,
2065                    schema_diff_ignore_case=schema_diff_ignore_case,
2066                )
2067            ]
2068
2069        if show:
2070            self.console.show_table_diff(table_diffs, show_sample, skip_grain_check, temp_schema)
2071
2072        return table_diffs
2073
2074    def _model_diff(
2075        self,
2076        model: Model,
2077        adapter: EngineAdapter,
2078        source: str,
2079        target: str,
2080        source_alias: str,
2081        target_alias: str,
2082        limit: int,
2083        decimals: int,
2084        on: t.Optional[t.List[str] | exp.Expr] = None,
2085        skip_columns: t.Optional[t.List[str]] = None,
2086        where: t.Optional[str | exp.Expr] = None,
2087        show: bool = True,
2088        temp_schema: t.Optional[str] = None,
2089        skip_grain_check: bool = False,
2090        schema_diff_ignore_case: bool = False,
2091    ) -> TableDiff:
2092        self.console.start_table_diff_model_progress(model.name)
2093
2094        table_diff = self._table_diff(
2095            on=on,
2096            skip_columns=skip_columns,
2097            where=where,
2098            limit=limit,
2099            decimals=decimals,
2100            model=model,
2101            adapter=adapter,
2102            source=source,
2103            target=target,
2104            source_alias=source_alias,
2105            target_alias=target_alias,
2106            schema_diff_ignore_case=schema_diff_ignore_case,
2107        )
2108
2109        if show:
2110            # Trigger row_diff in parallel execution so it's available for ordered display later
2111            table_diff.row_diff(temp_schema=temp_schema, skip_grain_check=skip_grain_check)
2112
2113        self.console.update_table_diff_progress(model.name)
2114
2115        return table_diff
2116
2117    def _table_diff(
2118        self,
2119        source: str,
2120        target: str,
2121        source_alias: str,
2122        target_alias: str,
2123        limit: int,
2124        decimals: int,
2125        adapter: EngineAdapter,
2126        on: t.Optional[t.List[str] | exp.Expr] = None,
2127        model: t.Optional[Model] = None,
2128        skip_columns: t.Optional[t.List[str]] = None,
2129        where: t.Optional[str | exp.Expr] = None,
2130        schema_diff_ignore_case: bool = False,
2131    ) -> TableDiff:
2132        if not on:
2133            raise SQLMeshError(
2134                "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."
2135            )
2136
2137        return TableDiff(
2138            adapter=adapter.with_settings(execute_log_level=logger.getEffectiveLevel()),
2139            source=source,
2140            target=target,
2141            on=on,
2142            skip_columns=skip_columns,
2143            where=where,
2144            source_alias=source_alias,
2145            target_alias=target_alias,
2146            limit=limit,
2147            decimals=decimals,
2148            model_name=model.name if model else None,
2149            model_dialect=model.dialect if model else None,
2150            schema_diff_ignore_case=schema_diff_ignore_case,
2151        )
2152
2153    @python_api_analytics
2154    def get_dag(
2155        self, select_models: t.Optional[t.Collection[str]] = None, **options: t.Any
2156    ) -> GraphHTML:
2157        """Gets an HTML object representation of the DAG.
2158
2159        Args:
2160            select_models: A list of model selection strings that should be included in the dag.
2161        Returns:
2162            An html object that renders the dag.
2163        """
2164        dag = (
2165            self.dag.prune(*self._new_selector().expand_model_selections(select_models))
2166            if select_models
2167            else self.dag
2168        )
2169
2170        nodes = {}
2171        edges: t.List[t.Dict] = []
2172
2173        for node, deps in dag.graph.items():
2174            nodes[node] = {
2175                "id": node,
2176                "label": node.split(".")[-1],
2177                "title": f"<span>{node}</span>",
2178            }
2179            edges.extend({"from": d, "to": node} for d in deps)
2180
2181        return GraphHTML(
2182            nodes,
2183            edges,
2184            options={
2185                "height": "100%",
2186                "width": "100%",
2187                "interaction": {},
2188                "layout": {
2189                    "hierarchical": {
2190                        "enabled": True,
2191                        "nodeSpacing": 200,
2192                        "sortMethod": "directed",
2193                    },
2194                },
2195                "nodes": {
2196                    "shape": "box",
2197                },
2198                **options,
2199            },
2200        )
2201
2202    @python_api_analytics
2203    def render_dag(self, path: str, select_models: t.Optional[t.Collection[str]] = None) -> None:
2204        """Render the dag as HTML and save it to a file.
2205
2206        Args:
2207            path: filename to save the dag html to
2208            select_models: A list of model selection strings that should be included in the dag.
2209        """
2210        file_path = Path(path)
2211        suffix = file_path.suffix
2212        if suffix != ".html":
2213            if suffix:
2214                get_console().log_warning(
2215                    f"The extension {suffix} does not designate an html file. A file with a `.html` extension will be created instead."
2216                )
2217            path = str(file_path.with_suffix(".html"))
2218
2219        with open(path, "w", encoding="utf-8") as file:
2220            file.write(str(self.get_dag(select_models)))
2221
2222    @python_api_analytics
2223    def create_test(
2224        self,
2225        model: str,
2226        input_queries: t.Dict[str, str],
2227        overwrite: bool = False,
2228        variables: t.Optional[t.Dict[str, str]] = None,
2229        path: t.Optional[str] = None,
2230        name: t.Optional[str] = None,
2231        include_ctes: bool = False,
2232    ) -> None:
2233        """Generate a unit test fixture for a given model.
2234
2235        Args:
2236            model: The model to test.
2237            input_queries: Mapping of model names to queries. Each model included in this mapping
2238                will be populated in the test based on the results of the corresponding query.
2239            overwrite: Whether to overwrite the existing test in case of a file path collision.
2240                When set to False, an error will be raised if there is such a collision.
2241            variables: Key-value pairs that will define variables needed by the model.
2242            path: The file path corresponding to the fixture, relative to the test directory.
2243                By default, the fixture will be created under the test directory and the file name
2244                will be inferred from the test's name.
2245            name: The name of the test. This is inferred from the model name by default.
2246            include_ctes: When true, CTE fixtures will also be generated.
2247        """
2248        input_queries = {
2249            # The get_model here has two purposes: return normalized names & check for missing deps
2250            self.get_model(dep, raise_if_missing=True).fqn: query
2251            for dep, query in input_queries.items()
2252        }
2253
2254        try:
2255            model_to_test = self.get_model(model, raise_if_missing=True)
2256            test_adapter = self.test_connection_config.create_engine_adapter(
2257                register_comments_override=False
2258            )
2259
2260            generate_test(
2261                model=model_to_test,
2262                input_queries=input_queries,
2263                models=self._models,
2264                engine_adapter=self._get_engine_adapter(model_to_test.gateway),
2265                test_engine_adapter=test_adapter,
2266                project_path=self.path,
2267                overwrite=overwrite,
2268                variables=variables,
2269                path=path,
2270                name=name,
2271                include_ctes=include_ctes,
2272            )
2273        finally:
2274            if test_adapter:
2275                test_adapter.close()
2276
2277    @python_api_analytics
2278    def test(
2279        self,
2280        match_patterns: t.Optional[t.List[str]] = None,
2281        tests: t.Optional[t.List[str]] = None,
2282        verbosity: Verbosity = Verbosity.DEFAULT,
2283        preserve_fixtures: bool = False,
2284        stream: t.Optional[t.TextIO] = None,
2285    ) -> ModelTextTestResult:
2286        """Discover and run model tests"""
2287        if verbosity >= Verbosity.VERBOSE:
2288            import pandas as pd
2289
2290            pd.set_option("display.max_columns", None)
2291
2292        test_meta = self.select_tests(tests=tests, patterns=match_patterns)
2293
2294        result = run_tests(
2295            model_test_metadata=test_meta,
2296            models=self._models,
2297            config=self.config,
2298            selected_gateway=self.selected_gateway,
2299            dialect=self.default_dialect,
2300            verbosity=verbosity,
2301            preserve_fixtures=preserve_fixtures,
2302            stream=stream,
2303            default_catalog=self.default_catalog,
2304            default_catalog_dialect=self.config.dialect or "",
2305        )
2306
2307        self.console.log_test_results(
2308            result,
2309            self.test_connection_config._engine_adapter.DIALECT,
2310        )
2311
2312        return result
2313
2314    @python_api_analytics
2315    def audit(
2316        self,
2317        start: TimeLike,
2318        end: TimeLike,
2319        *,
2320        models: t.Optional[t.Iterator[str]] = None,
2321        execution_time: t.Optional[TimeLike] = None,
2322    ) -> bool:
2323        """Audit models.
2324
2325        Args:
2326            start: The start of the interval to audit.
2327            end: The end of the interval to audit.
2328            models: The models to audit. All models will be audited if not specified.
2329            execution_time: The date/time time reference to use for execution time. Defaults to now.
2330
2331        Returns:
2332            False if any of the audits failed, True otherwise.
2333        """
2334
2335        snapshots = (
2336            [self.get_snapshot(model, raise_if_missing=True) for model in models]
2337            if models
2338            else self.snapshots.values()
2339        )
2340
2341        num_audits = sum(len(snapshot.node.audits_with_args) for snapshot in snapshots)
2342        self.console.log_status_update(f"Found {num_audits} audit(s).")
2343
2344        errors = []
2345        skipped_count = 0
2346        for snapshot in snapshots:
2347            for audit_result in self.snapshot_evaluator.audit(
2348                snapshot=snapshot,
2349                start=start,
2350                end=end,
2351                execution_time=execution_time,
2352                snapshots=self.snapshots,
2353            ):
2354                audit_id = f"{audit_result.audit.name}"
2355                if audit_result.model:
2356                    audit_id += f" on model {audit_result.model.name}"
2357
2358                if audit_result.skipped:
2359                    self.console.log_status_update(f"{audit_id} ⏸️ SKIPPED.")
2360                    skipped_count += 1
2361                elif audit_result.count:
2362                    errors.append(audit_result)
2363                    self.console.log_status_update(
2364                        f"{audit_id} ❌ [red]FAIL [{audit_result.count}][/red]."
2365                    )
2366                else:
2367                    self.console.log_status_update(f"{audit_id} ✅ [green]PASS[/green].")
2368
2369        self.console.log_status_update(
2370            f"\nFinished with {len(errors)} audit error{'' if len(errors) == 1 else 's'} "
2371            f"and {skipped_count} audit{'' if skipped_count == 1 else 's'} skipped."
2372        )
2373        for error in errors:
2374            self.console.log_status_update(
2375                f"\nFailure in audit {error.audit.name} ({error.audit._path})."
2376            )
2377            self.console.log_status_update(f"Got {error.count} results, expected 0.")
2378            if error.query:
2379                self.console.show_sql(
2380                    f"{error.query.sql(dialect=self.snapshot_evaluator.adapter.dialect)}"
2381                )
2382
2383        self.console.log_status_update("Done.")
2384        return not errors
2385
2386    @python_api_analytics
2387    def rewrite(self, sql: str, dialect: str = "") -> exp.Expr:
2388        """Rewrite a sql expression with semantic references into an executable query.
2389
2390        https://sqlmesh.readthedocs.io/en/latest/concepts/metrics/overview/
2391
2392        Args:
2393            sql: The sql string to rewrite.
2394            dialect: The dialect of the sql string, defaults to the project dialect.
2395
2396        Returns:
2397            A SQLGlot expression with semantic references expanded.
2398        """
2399        return rewrite(
2400            sql,
2401            graph=ReferenceGraph(self.models.values()),
2402            metrics=self._metrics,
2403            dialect=dialect or self.default_dialect,
2404        )
2405
2406    @python_api_analytics
2407    def check_intervals(
2408        self,
2409        environment: t.Optional[str],
2410        no_signals: bool,
2411        select_models: t.Collection[str],
2412        start: t.Optional[TimeLike] = None,
2413        end: t.Optional[TimeLike] = None,
2414    ) -> t.Dict[Snapshot, SnapshotIntervals]:
2415        """Check intervals for a given environment.
2416
2417        Args:
2418            environment: The environment or prod if None.
2419            select_models: A list of model selection strings to show intervals for.
2420            start: The start of the intervals to check.
2421            end: The end of the intervals to check.
2422        """
2423
2424        environment = environment or c.PROD
2425        env = self.state_reader.get_environment(environment)
2426        if not env:
2427            raise SQLMeshError(f"Environment '{environment}' was not found.")
2428
2429        snapshots = {k.name: v for k, v in self.state_sync.get_snapshots(env.snapshots).items()}
2430
2431        missing = {
2432            k.name: v
2433            for k, v in missing_intervals(
2434                snapshots.values(), start=start, end=end, execution_time=end
2435            ).items()
2436        }
2437
2438        if select_models:
2439            selected: t.Collection[str] = self._select_models_for_run(
2440                select_models, True, snapshots.values()
2441            )
2442        else:
2443            selected = snapshots.keys()
2444
2445        results = {}
2446        execution_context = self.execution_context(snapshots=snapshots)
2447
2448        for fqn in selected:
2449            snapshot = snapshots[fqn]
2450            intervals = missing.get(fqn) or []
2451
2452            results[snapshot] = SnapshotIntervals(
2453                snapshot.snapshot_id,
2454                intervals
2455                if no_signals
2456                else snapshot.check_ready_intervals(intervals, execution_context),
2457            )
2458
2459        return results
2460
2461    @python_api_analytics
2462    def migrate(self) -> None:
2463        """Migrates SQLMesh to the current running version.
2464
2465        Please contact your SQLMesh administrator before doing this.
2466        """
2467        self.notification_target_manager.notify(NotificationEvent.MIGRATION_START)
2468        self._load_materializations()
2469        try:
2470            self._new_state_sync().migrate(
2471                promoted_snapshots_only=self.config.migration.promoted_snapshots_only,
2472            )
2473        except Exception as e:
2474            self.notification_target_manager.notify(
2475                NotificationEvent.MIGRATION_FAILURE, traceback.format_exc()
2476            )
2477            raise e
2478        self.notification_target_manager.notify(NotificationEvent.MIGRATION_END)
2479
2480    @python_api_analytics
2481    def rollback(self) -> None:
2482        """Rolls back SQLMesh to the previous migration.
2483
2484        Please contact your SQLMesh administrator before doing this. This action cannot be undone.
2485        """
2486        self._new_state_sync().rollback()
2487
2488    @python_api_analytics
2489    def create_external_models(self, strict: bool = False) -> None:
2490        """Create a file to document the schema of external models.
2491
2492        The external models file contains all columns and types of external models, allowing for more
2493        robust lineage, validation, and optimizations.
2494
2495        Args:
2496            strict: If True, raise an error if the external model is missing in the database.
2497        """
2498        if not self._models:
2499            self.load(update_schemas=False)
2500
2501        for path, config in self.configs.items():
2502            deprecated_yaml = path / c.EXTERNAL_MODELS_DEPRECATED_YAML
2503
2504            external_models_yaml = (
2505                path / c.EXTERNAL_MODELS_YAML if not deprecated_yaml.exists() else deprecated_yaml
2506            )
2507
2508            external_models_gateway: t.Optional[str] = self.gateway or self.config.default_gateway
2509            if not external_models_gateway:
2510                # can happen if there was no --gateway defined and the default_gateway is ''
2511                # which means that the single gateway syntax is being used which means there is
2512                # no named gateway which means we should not stamp `gateway:` on the external models
2513                external_models_gateway = None
2514
2515            create_external_models_file(
2516                path=external_models_yaml,
2517                models=UniqueKeyDict(
2518                    "models",
2519                    {
2520                        fqn: model
2521                        for fqn, model in self._models.items()
2522                        if self.config_for_node(model) is config
2523                    },
2524                ),
2525                adapter=self.engine_adapter,
2526                state_reader=self.state_reader,
2527                dialect=config.model_defaults.dialect,
2528                gateway=external_models_gateway,
2529                max_workers=self.concurrent_tasks,
2530                strict=strict,
2531                all_models=self._models,
2532            )
2533
2534    @python_api_analytics
2535    def print_info(
2536        self, skip_connection: bool = False, verbosity: Verbosity = Verbosity.DEFAULT
2537    ) -> None:
2538        """Prints information about connections, models, macros, etc. to the console."""
2539        self.console.log_status_update(f"Models: {len(self.models)}")
2540        self.console.log_status_update(f"Macros: {len(self._macros) - len(macro.get_registry())}")
2541
2542        if skip_connection:
2543            return
2544
2545        if verbosity >= Verbosity.VERBOSE:
2546            self.console.log_status_update("")
2547            print_config(self.config.get_connection(self.gateway), self.console, "Connection")
2548            print_config(
2549                self.config.get_test_connection(self.gateway), self.console, "Test Connection"
2550            )
2551            print_config(
2552                self.config.get_state_connection(self.gateway), self.console, "State Connection"
2553            )
2554
2555        self._try_connection("data warehouse", self.engine_adapter.ping)
2556        state_connection = self.config.get_state_connection(self.gateway)
2557        if state_connection:
2558            self._try_connection("state backend", state_connection.connection_validator())
2559
2560    @python_api_analytics
2561    def print_environment_names(self) -> None:
2562        """Prints all environment names along with expiry datetime."""
2563        result = self._new_state_sync().get_environments_summary()
2564        if not result:
2565            raise SQLMeshError(
2566                "This project has no environments. Create an environment using the `sqlmesh plan` command."
2567            )
2568        self.console.print_environments(result)
2569
2570    def close(self) -> None:
2571        """Releases all resources allocated by this context."""
2572        if self._snapshot_evaluator:
2573            self._snapshot_evaluator.close()
2574
2575        if self._state_sync:
2576            self._state_sync.close()
2577
2578    def _run(
2579        self,
2580        environment: str,
2581        *,
2582        start: t.Optional[TimeLike],
2583        end: t.Optional[TimeLike],
2584        execution_time: t.Optional[TimeLike],
2585        ignore_cron: bool,
2586        select_models: t.Optional[t.Collection[str]],
2587        circuit_breaker: t.Optional[t.Callable[[], bool]],
2588        no_auto_upstream: bool,
2589    ) -> CompletionStatus:
2590        scheduler = self.scheduler(environment=environment)
2591        snapshots = scheduler.snapshots
2592
2593        if select_models is not None:
2594            select_models = self._select_models_for_run(
2595                select_models, no_auto_upstream, snapshots.values()
2596            )
2597
2598        completion_status = scheduler.run(
2599            environment,
2600            start=start,
2601            end=end,
2602            execution_time=execution_time,
2603            ignore_cron=ignore_cron,
2604            circuit_breaker=circuit_breaker,
2605            selected_snapshots=select_models,
2606            auto_restatement_enabled=environment.lower() == c.PROD,
2607            run_environment_statements=True,
2608        )
2609
2610        if completion_status.is_nothing_to_do:
2611            next_run_ready_msg = ""
2612
2613            next_ready_interval_start = get_next_model_interval_start(snapshots.values())
2614            if next_ready_interval_start:
2615                utc_time = format_tz_datetime(next_ready_interval_start)
2616                local_time = format_tz_datetime(next_ready_interval_start, use_local_timezone=True)
2617                time_msg = local_time if local_time == utc_time else f"{local_time} ({utc_time})"
2618                next_run_ready_msg = f"\n\nNext run will be ready at {time_msg}."
2619
2620            self.console.log_status_update(
2621                f"No models are ready to run. Please wait until a model `cron` interval has elapsed.{next_run_ready_msg}"
2622            )
2623
2624        return completion_status
2625
2626    def _apply(self, plan: Plan, circuit_breaker: t.Optional[t.Callable[[], bool]]) -> None:
2627        self._scheduler.create_plan_evaluator(self).evaluate(
2628            plan.to_evaluatable(), circuit_breaker=circuit_breaker
2629        )
2630
2631    @python_api_analytics
2632    def table_name(
2633        self, model_name: str, environment: t.Optional[str] = None, prod: bool = False
2634    ) -> str:
2635        """Returns the name of the pysical table for the given model name in the target environment.
2636
2637        Args:
2638            model_name: The name of the model.
2639            environment: The environment to source the model version from.
2640            prod: If True, return the name of the physical table that will be used in production for the model version
2641                promoted in the target environment.
2642
2643        Returns:
2644            The name of the physical table.
2645        """
2646        environment = environment or self.config.default_target_environment
2647        fqn = self._node_or_snapshot_to_fqn(model_name)
2648        target_env = self.state_reader.get_environment(environment)
2649        if not target_env:
2650            raise SQLMeshError(f"Environment '{environment}' was not found.")
2651
2652        snapshot_info = None
2653        for s in target_env.snapshots:
2654            if s.name == fqn:
2655                snapshot_info = s
2656                break
2657        if not snapshot_info:
2658            raise SQLMeshError(
2659                f"Model '{model_name}' was not found in environment '{environment}'."
2660            )
2661
2662        if target_env.name == c.PROD or prod:
2663            return snapshot_info.table_name()
2664
2665        snapshots = self.state_reader.get_snapshots(target_env.snapshots)
2666        deployability_index = DeployabilityIndex.create(snapshots)
2667
2668        return snapshot_info.table_name(
2669            is_deployable=deployability_index.is_deployable(snapshot_info.snapshot_id)
2670        )
2671
2672    def clear_caches(self) -> None:
2673        paths_to_remove = [path / c.CACHE for path in self.configs]
2674        paths_to_remove.append(self.cache_dir)
2675
2676        if IS_WINDOWS:
2677            paths_to_remove = [fix_windows_path(path) for path in paths_to_remove]
2678
2679        for path in paths_to_remove:
2680            if path.exists():
2681                rmtree(path)
2682
2683        if isinstance(self._state_sync, CachingStateSync):
2684            self._state_sync.clear_cache()
2685
2686    def export_state(
2687        self,
2688        output_file: Path,
2689        environment_names: t.Optional[t.List[str]] = None,
2690        local_only: bool = False,
2691        confirm: bool = True,
2692    ) -> None:
2693        from sqlmesh.core.state_sync.export_import import export_state
2694
2695        # trigger a connection to the StateSync so we can fail early if there is a problem
2696        # note we still need to do this even if we are doing a local export so we know what 'versions' to write
2697        self.state_sync.get_versions(validate=True)
2698
2699        local_snapshots = self.snapshots if local_only else None
2700
2701        if self.console.start_state_export(
2702            output_file=output_file,
2703            gateway=self.selected_gateway,
2704            state_connection_config=self._state_connection_config,
2705            environment_names=environment_names,
2706            local_only=local_only,
2707            confirm=confirm,
2708        ):
2709            try:
2710                export_state(
2711                    state_sync=self.state_sync,
2712                    output_file=output_file,
2713                    local_snapshots=local_snapshots,
2714                    environment_names=environment_names,
2715                    console=self.console,
2716                )
2717                self.console.stop_state_export(success=True, output_file=output_file)
2718            except:
2719                self.console.stop_state_export(success=False, output_file=output_file)
2720                raise
2721
2722    def import_state(self, input_file: Path, clear: bool = False, confirm: bool = True) -> None:
2723        from sqlmesh.core.state_sync.export_import import import_state
2724
2725        if self.console.start_state_import(
2726            input_file=input_file,
2727            gateway=self.selected_gateway,
2728            state_connection_config=self._state_connection_config,
2729            clear=clear,
2730            confirm=confirm,
2731        ):
2732            try:
2733                import_state(
2734                    state_sync=self.state_sync,
2735                    input_file=input_file,
2736                    clear=clear,
2737                    console=self.console,
2738                )
2739                self.console.stop_state_import(success=True, input_file=input_file)
2740            except:
2741                self.console.stop_state_import(success=False, input_file=input_file)
2742                raise
2743
2744    def _run_tests(
2745        self, verbosity: Verbosity = Verbosity.DEFAULT
2746    ) -> t.Tuple[ModelTextTestResult, str]:
2747        test_output_io = StringIO()
2748        result = self.test(stream=test_output_io, verbosity=verbosity)
2749        return result, test_output_io.getvalue()
2750
2751    def _run_plan_tests(self, skip_tests: bool = False) -> t.Optional[ModelTextTestResult]:
2752        if not skip_tests:
2753            result = self.test()
2754            if not result.wasSuccessful():
2755                raise PlanError(
2756                    "Cannot generate plan due to failing test(s). Fix test(s) and run again."
2757                )
2758            return result
2759        return None
2760
2761    def _warn_if_virtual_catalog_rematerialization(self, plan: "Plan") -> None:
2762        """Warn when ClickHouse models appear as new snapshots solely because a virtual catalog
2763        prefix was added to their FQNs after a catalog-aware gateway joined the project.
2764
2765        This situation causes every previously-applied ClickHouse model to be treated as brand-new
2766        by SQLMesh, triggering full re-materialization and historical backfills. Emitting a warning
2767        before the plan is displayed gives users a chance to understand the cost before applying.
2768        """
2769        from sqlglot import exp
2770
2771        # Collect the set of old 2-level snapshot names from the current environment so we can
2772        # detect which new 3-level names are renames rather than genuinely new models.
2773        old_names: t.Set[str] = set()
2774        for s_id in plan.context_diff.removed_snapshots:
2775            old_names.add(s_id.name)
2776        for name in plan.context_diff.snapshots_by_name:
2777            old_names.add(name)
2778
2779        affected: t.List[t.Tuple[str, str]] = []  # (new_3level_name, old_2level_name)
2780
2781        for gateway, adapter in self.engine_adapters.items():
2782            if not adapter.supports_virtual_catalog() or not adapter._default_catalog:
2783                continue
2784            virtual_catalog = adapter._default_catalog
2785
2786            for snapshot in plan.new_snapshots:
2787                table = exp.to_table(snapshot.name)
2788                if table.catalog != virtual_catalog:
2789                    continue
2790                # Reconstruct the 2-level name that would have been used before injection.
2791                old_name = f"{table.db}.{table.name}"
2792                if old_name in old_names:
2793                    affected.append((snapshot.name, old_name))
2794
2795        if not affected:
2796            return
2797
2798        max_display = 10
2799        model_lines = "\n".join(
2800            f"  - {new_name}  (was: {old_name})" for new_name, old_name in affected[:max_display]
2801        )
2802        if len(affected) > max_display:
2803            model_lines += f"\n  ... and {len(affected) - max_display} more"
2804
2805        self.console.log_warning(
2806            "ClickHouse models are being re-materialized due to virtual catalog FQN change.\n\n"
2807            "The following ClickHouse models appear as new because their fully-qualified\n"
2808            "names changed from 2-level (db.table) to 3-level (__gateway__.db.table):\n\n"
2809            f"{model_lines}\n\n"
2810            "FULL models will be recreated once. INCREMENTAL_BY_TIME_RANGE models will\n"
2811            "require a full historical backfill from their configured start date.\n\n"
2812            "This is a one-time cost when first adding a catalog-aware gateway to an\n"
2813            "existing ClickHouse project. To proceed, run `sqlmesh apply`."
2814        )
2815
2816    @property
2817    def _model_tables(self) -> t.Dict[str, str]:
2818        """Mapping of model name to physical table name.
2819
2820        If a snapshot has not been versioned yet, its view name will be returned.
2821        """
2822        return {
2823            fqn: (
2824                snapshot.table_name()
2825                if snapshot.version
2826                else snapshot.qualified_view_name.for_environment(
2827                    EnvironmentNamingInfo.from_environment_catalog_mapping(
2828                        self.environment_catalog_mapping,
2829                        name=c.PROD,
2830                        suffix_target=self.config.environment_suffix_target,
2831                    )
2832                )
2833            )
2834            for fqn, snapshot in self.snapshots.items()
2835        }
2836
2837    @cached_property
2838    def cache_dir(self) -> Path:
2839        if self.config.cache_dir:
2840            cache_path = Path(self.config.cache_dir)
2841            if cache_path.is_absolute():
2842                return cache_path
2843            return self.path / cache_path
2844
2845        # Default to .cache directory in the project path
2846        return self.path / c.CACHE
2847
2848    @cached_property
2849    def engine_adapters(self) -> t.Dict[str, EngineAdapter]:
2850        """Returns all the engine adapters for the gateways defined in the configurations."""
2851        adapters: t.Dict[str, EngineAdapter] = {self.selected_gateway: self.engine_adapter}
2852        for config in self.configs.values():
2853            for gateway_name in config.gateways:
2854                if gateway_name not in adapters:
2855                    connection = config.get_connection(gateway_name)
2856                    adapter = connection.create_engine_adapter(
2857                        concurrent_tasks=self.concurrent_tasks,
2858                    )
2859                    adapters[gateway_name] = adapter
2860        return adapters
2861
2862    @cached_property
2863    def default_catalog_per_gateway(self) -> t.Dict[str, str]:
2864        """Returns the default catalogs for each engine adapter."""
2865        return self._scheduler.get_default_catalog_per_gateway(self)
2866
2867    @property
2868    def concurrent_tasks(self) -> int:
2869        if self._concurrent_tasks is None:
2870            self._concurrent_tasks = self.connection_config.concurrent_tasks
2871        return self._concurrent_tasks
2872
2873    @cached_property
2874    def connection_config(self) -> ConnectionConfig:
2875        return self.config.get_connection(self.selected_gateway)
2876
2877    @cached_property
2878    def test_connection_config(self) -> ConnectionConfig:
2879        return self.config.get_test_connection(
2880            self.gateway,
2881            self.default_catalog,
2882            default_catalog_dialect=self.config.dialect,
2883        )
2884
2885    @cached_property
2886    def environment_catalog_mapping(self) -> RegexKeyDict:
2887        engine_adapter = None
2888        try:
2889            engine_adapter = self.engine_adapter
2890        except Exception:
2891            pass
2892
2893        if (
2894            self.config.environment_catalog_mapping
2895            and engine_adapter
2896            and not self.engine_adapter.catalog_support.is_multi_catalog_supported
2897        ):
2898            raise SQLMeshError(
2899                "Environment catalog mapping is only supported for engine adapters that support multiple catalogs"
2900            )
2901        return self.config.environment_catalog_mapping
2902
2903    def _get_engine_adapter(self, gateway: t.Optional[str] = None) -> EngineAdapter:
2904        if gateway:
2905            if adapter := self.engine_adapters.get(gateway):
2906                return adapter
2907            raise SQLMeshError(f"Gateway '{gateway}' not found in the available engine adapters.")
2908        return self.engine_adapter
2909
2910    def _snapshots(
2911        self, models_override: t.Optional[UniqueKeyDict[str, Model]] = None
2912    ) -> t.Dict[str, Snapshot]:
2913        nodes = {**(models_override or self._models), **self._standalone_audits}
2914        snapshots = self._nodes_to_snapshots(nodes)
2915        stored_snapshots = self.state_reader.get_snapshots(snapshots.values())
2916
2917        unrestorable_snapshots = {
2918            snapshot
2919            for snapshot in stored_snapshots.values()
2920            if snapshot.name in nodes and snapshot.unrestorable
2921        }
2922        if unrestorable_snapshots:
2923            for snapshot in unrestorable_snapshots:
2924                logger.info(
2925                    "Found a unrestorable snapshot %s. Restamping the model...", snapshot.name
2926                )
2927                node = nodes[snapshot.name]
2928                nodes[snapshot.name] = node.copy(
2929                    update={"stamp": f"revert to {snapshot.identifier}"}
2930                )
2931            snapshots = self._nodes_to_snapshots(nodes)
2932            stored_snapshots = self.state_reader.get_snapshots(snapshots.values())
2933
2934        for snapshot in stored_snapshots.values():
2935            # Keep the original model instance to preserve the query cache.
2936            snapshot.node = snapshots[snapshot.name].node
2937
2938        return {name: stored_snapshots.get(s.snapshot_id, s) for name, s in snapshots.items()}
2939
2940    def _context_diff(
2941        self,
2942        environment: str,
2943        snapshots: t.Optional[t.Dict[str, Snapshot]] = None,
2944        create_from: t.Optional[str] = None,
2945        force_no_diff: bool = False,
2946        ensure_finalized_snapshots: bool = False,
2947        diff_rendered: bool = False,
2948        always_recreate_environment: bool = False,
2949    ) -> ContextDiff:
2950        environment = Environment.sanitize_name(environment)
2951        if force_no_diff:
2952            return ContextDiff.create_no_diff(environment, self.state_reader)
2953
2954        return ContextDiff.create(
2955            environment,
2956            snapshots=snapshots or self.snapshots,
2957            create_from=create_from or c.PROD,
2958            state_reader=self.state_reader,
2959            provided_requirements=self._requirements,
2960            excluded_requirements=self._excluded_requirements,
2961            ensure_finalized_snapshots=ensure_finalized_snapshots,
2962            diff_rendered=diff_rendered,
2963            environment_statements=self._environment_statements,
2964            gateway_managed_virtual_layer=self.config.gateway_managed_virtual_layer,
2965            infer_python_dependencies=self.config.infer_python_dependencies,
2966            always_recreate_environment=always_recreate_environment,
2967        )
2968
2969    def _destroy(self) -> bool:
2970        # Invalidate all environments, including prod
2971        for environment in self.state_reader.get_environments():
2972            self.state_sync.invalidate_environment(name=environment.name, protect_prod=False)
2973            self.console.log_success(f"Environment '{environment.name}' invalidated.")
2974
2975        # Run janitor to clean up all objects
2976        self._run_janitor(ignore_ttl=True)
2977
2978        # Remove state tables, including backup tables
2979        self.state_sync.remove_state(including_backup=True)
2980        self.console.log_status_update("State tables removed.")
2981
2982        # Finally clear caches
2983        self.clear_caches()
2984
2985        return True
2986
2987    def _run_janitor(
2988        self,
2989        ignore_ttl: bool = False,
2990        force_delete: bool = False,
2991        environment: t.Optional[str] = None,
2992    ) -> None:
2993        current_ts = now_timestamp()
2994        failures: t.List[str] = []
2995
2996        # Clean up expired environments by removing their views and schemas
2997        failures.extend(
2998            self._cleanup_environments(
2999                current_ts=current_ts, force_delete=force_delete, name=environment
3000            )
3001        )
3002
3003        if environment is None:
3004            failures.extend(
3005                delete_expired_snapshots(
3006                    self.state_sync,
3007                    self.snapshot_evaluator,
3008                    current_ts=current_ts,
3009                    ignore_ttl=ignore_ttl,
3010                    force_delete=force_delete,
3011                    console=self.console,
3012                    batch_size=self.config.janitor.expired_snapshots_batch_size,
3013                )
3014            )
3015            self.state_sync.compact_intervals()
3016
3017        if failures:
3018            failure_string = "\n  - ".join(failures)
3019            summary = f"Janitor completed with failures:\n  {failure_string}"
3020            if force_delete:
3021                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."
3022            if self.config.janitor.warn_on_delete_failure:
3023                self.console.log_warning(summary)
3024            else:
3025                raise SQLMeshError(summary)
3026
3027    def _cleanup_environments(
3028        self,
3029        current_ts: t.Optional[int] = None,
3030        force_delete: bool = False,
3031        name: t.Optional[str] = None,
3032    ) -> t.List[str]:
3033        current_ts = current_ts or now_timestamp()
3034        failures: t.List[str] = []
3035
3036        expired_environments_summaries = self.state_sync.get_expired_environments(
3037            current_ts=current_ts, name=name
3038        )
3039
3040        if name is not None and not expired_environments_summaries:
3041            self.console.log_warning(
3042                f"Environment '{name}' is not expired or does not exist. Nothing to clean up."
3043            )
3044
3045        for expired_env_summary in expired_environments_summaries:
3046            expired_env = self.state_reader.get_environment(expired_env_summary.name)
3047
3048            if expired_env:
3049                failures.extend(
3050                    cleanup_expired_views(
3051                        default_adapter=self.engine_adapter,
3052                        engine_adapters=self.engine_adapters,
3053                        environments=[expired_env],
3054                        console=self.console,
3055                    )
3056                )
3057
3058        # we want to retry on the next janitor pass if drops failed, unless
3059        # force_delete is set in which case we purge state records regardless
3060        if not failures or force_delete:
3061            self.state_sync.delete_expired_environments(current_ts=current_ts, name=name)
3062        return failures
3063
3064    def _try_connection(self, connection_name: str, validator: t.Callable[[], None]) -> None:
3065        connection_name = connection_name.capitalize()
3066        try:
3067            validator()
3068            self.console.log_status_update(f"{connection_name} connection [green]succeeded[/green]")
3069        except Exception as ex:
3070            self.console.log_error(f"{connection_name} connection failed. {ex}")
3071
3072    def _new_state_sync(self) -> StateSync:
3073        return self._provided_state_sync or self._scheduler.create_state_sync(self)
3074
3075    def _new_selector(
3076        self, models: t.Optional[UniqueKeyDict[str, Model]] = None, dag: t.Optional[DAG[str]] = None
3077    ) -> Selector:
3078        return self._selector_cls(
3079            self.state_reader,
3080            models=models or self._models,
3081            context_path=self.path,
3082            dag=dag,
3083            default_catalog=self.default_catalog,
3084            dialect=self.default_dialect,
3085            cache_dir=self.cache_dir,
3086        )
3087
3088    def _register_notification_targets(self) -> None:
3089        event_notifications = collections.defaultdict(set)
3090        for target in self.notification_targets:
3091            if target.is_configured:
3092                for event in target.notify_on:
3093                    event_notifications[event].add(target)
3094        user_notification_targets = {
3095            user.username: set(
3096                target for target in user.notification_targets if target.is_configured
3097            )
3098            for user in self.users
3099        }
3100        self.notification_target_manager = NotificationTargetManager(
3101            event_notifications, user_notification_targets, username=self.config.username
3102        )
3103
3104    def _load_materializations(self) -> None:
3105        if not self._loaded:
3106            for loader in self._loaders:
3107                loader.load_materializations()
3108
3109    def _select_models_for_run(
3110        self,
3111        select_models: t.Collection[str],
3112        no_auto_upstream: bool,
3113        snapshots: t.Collection[Snapshot],
3114    ) -> t.Set[str]:
3115        models: UniqueKeyDict[str, Model] = UniqueKeyDict(
3116            "models", **{s.name: s.model for s in snapshots if s.is_model}
3117        )
3118        dag: DAG[str] = DAG()
3119        for fqn, model in models.items():
3120            dag.add(fqn, model.depends_on)
3121        model_selector = self._new_selector(models=models, dag=dag)
3122        result = set(model_selector.expand_model_selections(select_models))
3123        if not no_auto_upstream:
3124            result = set(dag.subdag(*result))
3125        return result
3126
3127    @cached_property
3128    def _project_type(self) -> str:
3129        project_types = {
3130            c.DBT if loader.__class__.__name__.lower().startswith(c.DBT) else c.NATIVE
3131            for loader in self._loaders
3132        }
3133        return c.HYBRID if len(project_types) > 1 else first(project_types)
3134
3135    def _nodes_to_snapshots(self, nodes: t.Dict[str, Node]) -> t.Dict[str, Snapshot]:
3136        snapshots: t.Dict[str, Snapshot] = {}
3137        fingerprint_cache: t.Dict[str, SnapshotFingerprint] = {}
3138
3139        for node in nodes.values():
3140            kwargs: t.Dict[str, t.Any] = {}
3141            if node.project in self._projects:
3142                config = self.config_for_node(node)
3143                kwargs["ttl"] = config.snapshot_ttl
3144                kwargs["table_naming_convention"] = config.physical_table_naming_convention
3145
3146            snapshot = Snapshot.from_node(
3147                node,
3148                nodes=nodes,
3149                cache=fingerprint_cache,
3150                **kwargs,
3151            )
3152            snapshots[snapshot.name] = snapshot
3153        return snapshots
3154
3155    def _node_or_snapshot_to_fqn(self, node_or_snapshot: NodeOrSnapshot) -> str:
3156        if isinstance(node_or_snapshot, Snapshot):
3157            return node_or_snapshot.name
3158        if isinstance(node_or_snapshot, str) and not self.standalone_audits.get(node_or_snapshot):
3159            return normalize_model_name(
3160                node_or_snapshot,
3161                dialect=self.default_dialect,
3162                default_catalog=self.default_catalog,
3163            )
3164        if not isinstance(node_or_snapshot, str):
3165            return node_or_snapshot.fqn
3166        return node_or_snapshot
3167
3168    @property
3169    def _plan_preview_enabled(self) -> bool:
3170        if self.config.plan.enable_preview is not None:
3171            return self.config.plan.enable_preview
3172        # It is dangerous to enable preview by default for dbt projects that rely on engines that don't support cloning.
3173        # Enabling previews in such cases can result in unintended full refreshes because dbt incremental models rely on
3174        # the maximum timestamp value in the target table.
3175        return self._project_type == c.NATIVE or self.engine_adapter.SUPPORTS_CLONING
3176
3177    def _get_plan_default_start_end(
3178        self,
3179        snapshots: t.Dict[str, Snapshot],
3180        max_interval_end_per_model: t.Dict[str, datetime],
3181        backfill_models: t.Optional[t.Set[str]],
3182        modified_model_names: t.Set[str],
3183        execution_time: t.Optional[TimeLike] = None,
3184    ) -> t.Tuple[t.Optional[int], t.Optional[int]]:
3185        # exclude seeds so their stale interval ends does not become the default plan end date
3186        # when they're the only ones that contain intervals in this plan
3187        non_seed_interval_ends = {
3188            model_fqn: end
3189            for model_fqn, end in max_interval_end_per_model.items()
3190            if model_fqn not in snapshots or not snapshots[model_fqn].is_seed
3191        }
3192        if not non_seed_interval_ends:
3193            return None, None
3194
3195        default_end = to_timestamp(max(non_seed_interval_ends.values()))
3196        default_start: t.Optional[int] = None
3197        # Infer the default start by finding the smallest interval start that corresponds to the default end.
3198        for model_name in backfill_models or modified_model_names or max_interval_end_per_model:
3199            if model_name not in snapshots:
3200                continue
3201            node = snapshots[model_name].node
3202            interval_unit = node.interval_unit
3203            default_start = min(
3204                default_start or sys.maxsize,
3205                to_timestamp(
3206                    interval_unit.cron_prev(
3207                        interval_unit.cron_floor(
3208                            max_interval_end_per_model.get(
3209                                model_name, node.cron_floor(default_end)
3210                            ),
3211                        ),
3212                        estimate=True,
3213                    )
3214                ),
3215            )
3216
3217        if execution_time and to_timestamp(default_end) > to_timestamp(execution_time):
3218            # the end date can't be in the future, which can happen if a specific `execution_time` is set and prod intervals
3219            # are newer than it
3220            default_end = to_timestamp(execution_time)
3221
3222        return default_start, default_end
3223
3224    def _calculate_start_override_per_model(
3225        self,
3226        min_intervals: t.Optional[int],
3227        plan_start: t.Optional[TimeLike],
3228        plan_end: t.Optional[TimeLike],
3229        plan_execution_time: TimeLike,
3230        backfill_model_fqns: t.Optional[t.Set[str]],
3231        snapshots_by_model_fqn: t.Dict[str, Snapshot],
3232        end_override_per_model: t.Optional[t.Dict[str, datetime]],
3233    ) -> t.Dict[str, datetime]:
3234        if not min_intervals or not backfill_model_fqns or not plan_start:
3235            # If there are no models to backfill, there are no intervals to consider for backfill, so we dont need to consider a minimum number
3236            # If the plan doesnt have a start date, all intervals are considered already so we dont need to consider a minimum number
3237            # 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
3238            return {}
3239
3240        start_overrides: t.Dict[str, datetime] = {}
3241        end_override_per_model = end_override_per_model or {}
3242
3243        plan_execution_time_dt = to_datetime(plan_execution_time)
3244        plan_start_dt = to_datetime(plan_start, relative_base=plan_execution_time_dt)
3245        plan_end_dt = to_datetime(
3246            plan_end or plan_execution_time_dt, relative_base=plan_execution_time_dt
3247        )
3248
3249        # we need to take the DAG into account so that parent models can be expanded to cover at least as much as their children
3250        # for example, A(hourly) <- B(daily)
3251        # if min_intervals=1, A would have 1 hour and B would have 1 day
3252        # but B depends on A so in order for B to have 1 valid day, A needs to be expanded to 24 hours
3253        backfill_dag: DAG[str] = DAG()
3254        for fqn in backfill_model_fqns:
3255            backfill_dag.add(
3256                fqn,
3257                [
3258                    p.name
3259                    for p in snapshots_by_model_fqn[fqn].parents
3260                    if p.name in backfill_model_fqns
3261                ],
3262            )
3263
3264        # 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
3265        reversed_dag = backfill_dag.reversed
3266        graph = reversed_dag.graph
3267
3268        for model_fqn in reversed_dag:
3269            # Get the earliest start from all immediate children of this snapshot
3270            # this works because topological ordering guarantees that they've already been visited
3271            # and we always set a start override
3272            min_child_start = min(
3273                [start_overrides[immediate_child_fqn] for immediate_child_fqn in graph[model_fqn]],
3274                default=plan_start_dt,
3275            )
3276
3277            snapshot = snapshots_by_model_fqn.get(model_fqn)
3278
3279            if not snapshot:
3280                continue
3281
3282            starting_point = end_override_per_model.get(model_fqn, plan_end_dt)
3283            if node_end := snapshot.node.end:
3284                # if we dont do this, if the node end is a *date* (as opposed to a timestamp)
3285                # we end up incorrectly winding back an extra day
3286                node_end_dt = make_exclusive(node_end)
3287
3288                if node_end_dt < plan_end_dt:
3289                    # if the model has an end date that has already elapsed, use that as a starting point for calculating min_intervals
3290                    # instead of the plan end. If we use the plan end, we will return intervals in the future which are invalid
3291                    starting_point = node_end_dt
3292
3293            snapshot_start = snapshot.node.cron_floor(starting_point)
3294
3295            for _ in range(min_intervals):
3296                # wind back the starting point by :min_intervals intervals to arrive at the minimum snapshot start date
3297                snapshot_start = snapshot.node.cron_prev(snapshot_start)
3298
3299            start_overrides[model_fqn] = min(min_child_start, snapshot_start)
3300
3301        return start_overrides
3302
3303    def _get_max_interval_end_per_model(
3304        self, snapshots: t.Dict[str, Snapshot], backfill_models: t.Optional[t.Set[str]]
3305    ) -> t.Dict[str, datetime]:
3306        models_for_interval_end = (
3307            self._get_models_for_interval_end(snapshots, backfill_models)
3308            if backfill_models is not None
3309            else None
3310        )
3311        return {
3312            model_fqn: to_datetime(ts)
3313            for model_fqn, ts in self.state_sync.max_interval_end_per_model(
3314                c.PROD,
3315                models=models_for_interval_end,
3316                ensure_finalized_snapshots=self.config.plan.use_finalized_state,
3317            ).items()
3318        }
3319
3320    @staticmethod
3321    def _get_models_for_interval_end(
3322        snapshots: t.Dict[str, Snapshot], backfill_models: t.Set[str]
3323    ) -> t.Set[str]:
3324        models_for_interval_end = set()
3325        models_stack = list(backfill_models)
3326        while models_stack:
3327            next_model = models_stack.pop()
3328            if next_model not in snapshots:
3329                continue
3330            models_for_interval_end.add(next_model)
3331            models_stack.extend(
3332                s.name
3333                for s in snapshots[next_model].parents
3334                if s.name not in models_for_interval_end
3335            )
3336        return models_for_interval_end
3337
3338    def lint_models(
3339        self,
3340        models: t.Optional[t.Iterable[t.Union[str, Model]]] = None,
3341        raise_on_error: bool = True,
3342    ) -> t.List[AnnotatedRuleViolation]:
3343        found_error = False
3344
3345        model_list = (
3346            list(self.get_model(model, raise_if_missing=True) for model in models)
3347            if models
3348            else self.models.values()
3349        )
3350        all_violations = []
3351        for model in model_list:
3352            # Linter may be `None` if the context is not loaded yet
3353            if linter := self._linters.get(model.project):
3354                lint_violation, violations = (
3355                    linter.lint_model(model, self, console=self.console) or found_error
3356                )
3357                if lint_violation:
3358                    found_error = True
3359                all_violations.extend(violations)
3360
3361        if raise_on_error and found_error:
3362            raise LinterError(
3363                "Linter detected errors in the code. Please fix them before proceeding."
3364            )
3365
3366        return all_violations
3367
3368    def select_tests(
3369        self,
3370        tests: t.Optional[t.List[str]] = None,
3371        patterns: t.Optional[t.List[str]] = None,
3372    ) -> t.List[ModelTestMetadata]:
3373        """Filter pre-loaded test metadata based on tests and patterns."""
3374
3375        test_meta = self._model_test_metadata
3376
3377        if tests:
3378            filtered_tests = []
3379            for test in tests:
3380                if "::" in test:
3381                    if test in self._model_test_metadata_fully_qualified_name_index:
3382                        filtered_tests.append(
3383                            self._model_test_metadata_fully_qualified_name_index[test]
3384                        )
3385                else:
3386                    test_path = Path(test)
3387                    if test_path in self._model_test_metadata_path_index:
3388                        filtered_tests.extend(self._model_test_metadata_path_index[test_path])
3389
3390            test_meta = filtered_tests
3391
3392        if patterns:
3393            test_meta = filter_tests_by_patterns(test_meta, patterns)
3394
3395        return test_meta
3396
3397
3398class Context(GenericContext[Config]):
3399    CONFIG_TYPE = Config