sqlmesh.core.console

   1from __future__ import annotations
   2
   3import abc
   4import datetime
   5import typing as t
   6import unittest
   7import uuid
   8import logging
   9import textwrap
  10from humanize import metric, naturalsize
  11from itertools import zip_longest
  12from pathlib import Path
  13from hyperscript import h
  14from rich.console import Console as RichConsole
  15from rich.live import Live
  16from rich.progress import (
  17    BarColumn,
  18    Progress,
  19    SpinnerColumn,
  20    TaskID,
  21    TextColumn,
  22    TimeElapsedColumn,
  23)
  24from rich.prompt import Confirm, Prompt
  25from rich.status import Status
  26from rich.syntax import Syntax
  27from rich.table import Table
  28from rich.tree import Tree
  29from sqlglot import exp
  30
  31from sqlmesh.core.schema_diff import TableAlterOperation
  32from sqlmesh.core.test.result import ModelTextTestResult
  33from sqlmesh.core.environment import EnvironmentNamingInfo, EnvironmentSummary
  34from sqlmesh.core.linter.rule import RuleViolation
  35from sqlmesh.core.model import Model
  36from sqlmesh.core.snapshot import (
  37    Snapshot,
  38    SnapshotChangeCategory,
  39    SnapshotId,
  40    SnapshotInfoLike,
  41)
  42from sqlmesh.core.snapshot.definition import Interval, Intervals, SnapshotTableInfo
  43from sqlmesh.core.snapshot.execution_tracker import QueryExecutionStats
  44from sqlmesh.core.test import ModelTest
  45from sqlmesh.utils import rich as srich
  46from sqlmesh.utils import Verbosity
  47from sqlmesh.utils.concurrency import NodeExecutionFailedError
  48from sqlmesh.utils.date import time_like_to_str, to_date, yesterday_ds, to_ds, make_inclusive
  49from sqlmesh.utils.errors import (
  50    PythonModelEvalError,
  51    NodeAuditsErrors,
  52    format_destructive_change_msg,
  53    format_additive_change_msg,
  54)
  55from sqlmesh.utils.rich import strip_ansi_codes
  56
  57if t.TYPE_CHECKING:
  58    import ipywidgets as widgets
  59
  60    from sqlglot import exp
  61    from sqlglot.dialects.dialect import DialectType
  62    from sqlmesh.core.context_diff import ContextDiff
  63    from sqlmesh.core.plan import Plan, EvaluatablePlan, PlanBuilder, SnapshotIntervals
  64    from sqlmesh.core.table_diff import TableDiff, RowDiff, SchemaDiff
  65    from sqlmesh.core.config.connection import ConnectionConfig
  66    from sqlmesh.core.state_sync import Versions
  67
  68    LayoutWidget = t.TypeVar("LayoutWidget", bound=t.Union[widgets.VBox, widgets.HBox])
  69
  70
  71logger = logging.getLogger(__name__)
  72
  73
  74SNAPSHOT_CHANGE_CATEGORY_STR = {
  75    None: "Unknown",
  76    SnapshotChangeCategory.BREAKING: "Breaking",
  77    SnapshotChangeCategory.NON_BREAKING: "Non-breaking",
  78    SnapshotChangeCategory.FORWARD_ONLY: "Forward-only",
  79    SnapshotChangeCategory.INDIRECT_BREAKING: "Indirect Breaking",
  80    SnapshotChangeCategory.INDIRECT_NON_BREAKING: "Indirect Non-breaking",
  81    SnapshotChangeCategory.METADATA: "Metadata",
  82}
  83
  84PROGRESS_BAR_WIDTH = 40
  85LINE_WRAP_WIDTH = 100
  86
  87
  88class LinterConsole(abc.ABC):
  89    """Console for displaying linter violations"""
  90
  91    @abc.abstractmethod
  92    def show_linter_violations(
  93        self, violations: t.List[RuleViolation], model: Model, is_error: bool = False
  94    ) -> None:
  95        """Prints all linter violations depending on their severity"""
  96
  97
  98class StateExporterConsole(abc.ABC):
  99    """Console for describing a state export"""
 100
 101    @abc.abstractmethod
 102    def start_state_export(
 103        self,
 104        output_file: Path,
 105        gateway: t.Optional[str] = None,
 106        state_connection_config: t.Optional[ConnectionConfig] = None,
 107        environment_names: t.Optional[t.List[str]] = None,
 108        local_only: bool = False,
 109        confirm: bool = True,
 110    ) -> bool:
 111        """State a state export"""
 112
 113    @abc.abstractmethod
 114    def update_state_export_progress(
 115        self,
 116        version_count: t.Optional[int] = None,
 117        versions_complete: bool = False,
 118        snapshot_count: t.Optional[int] = None,
 119        snapshots_complete: bool = False,
 120        environment_count: t.Optional[int] = None,
 121        environments_complete: bool = False,
 122    ) -> None:
 123        """Update the state export progress"""
 124
 125    @abc.abstractmethod
 126    def stop_state_export(self, success: bool, output_file: Path) -> None:
 127        """Finish a state export"""
 128
 129
 130class StateImporterConsole(abc.ABC):
 131    """Console for describing a state import"""
 132
 133    @abc.abstractmethod
 134    def start_state_import(
 135        self,
 136        input_file: Path,
 137        gateway: str,
 138        state_connection_config: ConnectionConfig,
 139        clear: bool = False,
 140        confirm: bool = True,
 141    ) -> bool:
 142        """Start a state import"""
 143
 144    @abc.abstractmethod
 145    def update_state_import_progress(
 146        self,
 147        timestamp: t.Optional[str] = None,
 148        state_file_version: t.Optional[int] = None,
 149        versions: t.Optional[Versions] = None,
 150        snapshot_count: t.Optional[int] = None,
 151        snapshots_complete: bool = False,
 152        environment_count: t.Optional[int] = None,
 153        environments_complete: bool = False,
 154    ) -> None:
 155        """Update the state import process"""
 156
 157    @abc.abstractmethod
 158    def stop_state_import(self, success: bool, input_file: Path) -> None:
 159        """Finish a state import"""
 160
 161
 162class JanitorConsole(abc.ABC):
 163    """Console for describing a janitor / snapshot cleanup run"""
 164
 165    @abc.abstractmethod
 166    def start_cleanup(self, ignore_ttl: bool) -> bool:
 167        """Start a janitor / snapshot cleanup run.
 168
 169        Args:
 170            ignore_ttl: Indicates that the user wants to ignore the snapshot TTL and clean up everything not promoted to an environment
 171
 172        Returns:
 173            Whether or not the cleanup run should proceed
 174        """
 175
 176    @abc.abstractmethod
 177    def update_cleanup_progress(self, object_name: str) -> None:
 178        """Update the snapshot cleanup progress."""
 179
 180    @abc.abstractmethod
 181    def stop_cleanup(self, success: bool = True) -> None:
 182        """Indicates the janitor / snapshot cleanup run has ended
 183
 184        Args:
 185            success: Whether or not the cleanup completed successfully
 186        """
 187
 188
 189class DestroyConsole(abc.ABC):
 190    """Console for describing a destroy operation"""
 191
 192    @abc.abstractmethod
 193    def start_destroy(
 194        self,
 195        schemas_to_delete: t.Optional[t.Set[str]] = None,
 196        views_to_delete: t.Optional[t.Set[str]] = None,
 197        tables_to_delete: t.Optional[t.Set[str]] = None,
 198    ) -> bool:
 199        """Start a destroy operation.
 200
 201        Args:
 202            schemas_to_delete: Set of schemas that will be deleted
 203            views_to_delete: Set of views that will be deleted
 204            tables_to_delete: Set of tables that will be deleted
 205
 206        Returns:
 207            Whether or not the destroy operation should proceed
 208        """
 209
 210    @abc.abstractmethod
 211    def stop_destroy(self, success: bool = True) -> None:
 212        """Indicates the destroy operation has ended
 213
 214        Args:
 215            success: Whether or not the cleanup completed successfully
 216        """
 217
 218
 219class EnvironmentsConsole(abc.ABC):
 220    """Console for displaying environments"""
 221
 222    @abc.abstractmethod
 223    def print_environments(self, environments_summary: t.List[EnvironmentSummary]) -> None:
 224        """Prints all environment names along with expiry datetime."""
 225
 226    @abc.abstractmethod
 227    def show_intervals(self, snapshot_intervals: t.Dict[Snapshot, SnapshotIntervals]) -> None:
 228        """Show ready intervals"""
 229
 230
 231class DifferenceConsole(abc.ABC):
 232    """Console for displaying environment differences"""
 233
 234    @abc.abstractmethod
 235    def show_environment_difference_summary(
 236        self,
 237        context_diff: ContextDiff,
 238        no_diff: bool = True,
 239    ) -> None:
 240        """Displays a summary of differences for the environment."""
 241
 242    @abc.abstractmethod
 243    def show_model_difference_summary(
 244        self,
 245        context_diff: ContextDiff,
 246        environment_naming_info: EnvironmentNamingInfo,
 247        default_catalog: t.Optional[str],
 248        no_diff: bool = True,
 249    ) -> None:
 250        """Displays a summary of differences for the given models."""
 251
 252
 253class TableDiffConsole(abc.ABC):
 254    """Console for displaying table differences"""
 255
 256    @abc.abstractmethod
 257    def show_table_diff(
 258        self,
 259        table_diffs: t.List[TableDiff],
 260        show_sample: bool = True,
 261        skip_grain_check: bool = False,
 262        temp_schema: t.Optional[str] = None,
 263    ) -> None:
 264        """Display the table diff between two or multiple tables."""
 265
 266    @abc.abstractmethod
 267    def update_table_diff_progress(self, model: str) -> None:
 268        """Update table diff progress bar"""
 269
 270    @abc.abstractmethod
 271    def start_table_diff_progress(self, models_to_diff: int) -> None:
 272        """Start table diff progress bar"""
 273
 274    @abc.abstractmethod
 275    def start_table_diff_model_progress(self, model: str) -> None:
 276        """Start table diff model progress"""
 277
 278    @abc.abstractmethod
 279    def stop_table_diff_progress(self, success: bool) -> None:
 280        """Stop table diff progress bar"""
 281
 282    @abc.abstractmethod
 283    def show_table_diff_details(
 284        self,
 285        models_to_diff: t.List[str],
 286    ) -> None:
 287        """Display information about which tables are going to be diffed"""
 288
 289    @abc.abstractmethod
 290    def show_table_diff_summary(self, table_diff: TableDiff) -> None:
 291        """Display information about the tables being diffed and how they are being joined"""
 292
 293    @abc.abstractmethod
 294    def show_schema_diff(self, schema_diff: SchemaDiff) -> None:
 295        """Show table schema diff."""
 296
 297    @abc.abstractmethod
 298    def show_row_diff(
 299        self, row_diff: RowDiff, show_sample: bool = True, skip_grain_check: bool = False
 300    ) -> None:
 301        """Show table summary diff."""
 302
 303
 304class BaseConsole(abc.ABC):
 305    @abc.abstractmethod
 306    def log_error(self, message: str, *args: t.Any, **kwargs: t.Any) -> None:
 307        """Display error info to the user."""
 308
 309    @abc.abstractmethod
 310    def log_warning(
 311        self,
 312        short_message: str,
 313        long_message: t.Optional[str] = None,
 314        *args: t.Any,
 315        **kwargs: t.Any,
 316    ) -> None:
 317        """Display warning info to the user.
 318
 319        Args:
 320            short_message: The warning message to print to console.
 321            long_message: The warning message to log to file. If not provided, `short_message` is used.
 322        """
 323
 324    @abc.abstractmethod
 325    def log_success(self, message: str) -> None:
 326        """Display a general successful message to the user."""
 327
 328
 329class PlanBuilderConsole(BaseConsole, abc.ABC):
 330    @abc.abstractmethod
 331    def log_destructive_change(
 332        self,
 333        snapshot_name: str,
 334        alter_operations: t.List[TableAlterOperation],
 335        dialect: str,
 336        error: bool = True,
 337    ) -> None:
 338        """Display a destructive change error or warning to the user."""
 339
 340    @abc.abstractmethod
 341    def log_additive_change(
 342        self,
 343        snapshot_name: str,
 344        alter_operations: t.List[TableAlterOperation],
 345        dialect: str,
 346        error: bool = True,
 347    ) -> None:
 348        """Display an additive change error or warning to the user."""
 349
 350
 351class UnitTestConsole(abc.ABC):
 352    @abc.abstractmethod
 353    def log_test_results(self, result: ModelTextTestResult, target_dialect: str) -> None:
 354        """Display the test result and output.
 355
 356        Args:
 357            result: The unittest test result that contains metrics like num success, fails, ect.
 358            target_dialect: The dialect that tests were run against. Assumes all tests run against the same dialect.
 359        """
 360
 361
 362class SignalConsole(abc.ABC):
 363    @abc.abstractmethod
 364    def start_signal_progress(
 365        self,
 366        snapshot: Snapshot,
 367        default_catalog: t.Optional[str],
 368        environment_naming_info: EnvironmentNamingInfo,
 369    ) -> None:
 370        """Indicates that signal checking has begun for a snapshot."""
 371
 372    @abc.abstractmethod
 373    def update_signal_progress(
 374        self,
 375        snapshot: Snapshot,
 376        signal_name: str,
 377        signal_idx: int,
 378        total_signals: int,
 379        ready_intervals: Intervals,
 380        check_intervals: Intervals,
 381        duration: float,
 382    ) -> None:
 383        """Updates the signal checking progress."""
 384
 385    @abc.abstractmethod
 386    def stop_signal_progress(self) -> None:
 387        """Indicates that signal checking has completed for a snapshot."""
 388
 389
 390class Console(
 391    SignalConsole,
 392    PlanBuilderConsole,
 393    LinterConsole,
 394    StateExporterConsole,
 395    StateImporterConsole,
 396    JanitorConsole,
 397    DestroyConsole,
 398    EnvironmentsConsole,
 399    DifferenceConsole,
 400    TableDiffConsole,
 401    BaseConsole,
 402    UnitTestConsole,
 403    abc.ABC,
 404):
 405    """Abstract base class for defining classes used for displaying information to the user and also interact
 406    with them when their input is needed."""
 407
 408    INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD = 10
 409
 410    @abc.abstractmethod
 411    def start_plan_evaluation(self, plan: EvaluatablePlan) -> None:
 412        """Indicates that a new evaluation has begun."""
 413
 414    @abc.abstractmethod
 415    def stop_plan_evaluation(self) -> None:
 416        """Indicates that the evaluation has ended."""
 417
 418    @abc.abstractmethod
 419    def start_evaluation_progress(
 420        self,
 421        batched_intervals: t.Dict[Snapshot, Intervals],
 422        environment_naming_info: EnvironmentNamingInfo,
 423        default_catalog: t.Optional[str],
 424        audit_only: bool = False,
 425    ) -> None:
 426        """Indicates that a new snapshot evaluation/auditing progress has begun."""
 427
 428    @abc.abstractmethod
 429    def start_snapshot_evaluation_progress(
 430        self, snapshot: Snapshot, audit_only: bool = False
 431    ) -> None:
 432        """Starts the snapshot evaluation progress."""
 433
 434    @abc.abstractmethod
 435    def update_snapshot_evaluation_progress(
 436        self,
 437        snapshot: Snapshot,
 438        interval: Interval,
 439        batch_idx: int,
 440        duration_ms: t.Optional[int],
 441        num_audits_passed: int,
 442        num_audits_failed: int,
 443        audit_only: bool = False,
 444        execution_stats: t.Optional[QueryExecutionStats] = None,
 445        auto_restatement_triggers: t.Optional[t.List[SnapshotId]] = None,
 446    ) -> None:
 447        """Updates the snapshot evaluation progress."""
 448
 449    @abc.abstractmethod
 450    def stop_evaluation_progress(self, success: bool = True) -> None:
 451        """Stops the snapshot evaluation progress."""
 452
 453    @abc.abstractmethod
 454    def start_creation_progress(
 455        self,
 456        snapshots: t.List[Snapshot],
 457        environment_naming_info: EnvironmentNamingInfo,
 458        default_catalog: t.Optional[str],
 459    ) -> None:
 460        """Indicates that a new snapshot creation progress has begun."""
 461
 462    @abc.abstractmethod
 463    def update_creation_progress(self, snapshot: SnapshotInfoLike) -> None:
 464        """Update the snapshot creation progress."""
 465
 466    @abc.abstractmethod
 467    def stop_creation_progress(self, success: bool = True) -> None:
 468        """Stop the snapshot creation progress."""
 469
 470    @abc.abstractmethod
 471    def start_promotion_progress(
 472        self,
 473        snapshots: t.List[SnapshotTableInfo],
 474        environment_naming_info: EnvironmentNamingInfo,
 475        default_catalog: t.Optional[str],
 476    ) -> None:
 477        """Indicates that a new snapshot promotion progress has begun."""
 478
 479    @abc.abstractmethod
 480    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
 481        """Update the snapshot promotion progress."""
 482
 483    @abc.abstractmethod
 484    def stop_promotion_progress(self, success: bool = True) -> None:
 485        """Stop the snapshot promotion progress."""
 486
 487    @abc.abstractmethod
 488    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
 489        """Indicates that a new snapshot migration progress has begun."""
 490
 491    @abc.abstractmethod
 492    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
 493        """Update the snapshot migration progress."""
 494
 495    @abc.abstractmethod
 496    def log_migration_status(self, success: bool = True) -> None:
 497        """Log the finished migration status."""
 498
 499    @abc.abstractmethod
 500    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
 501        """Stop the snapshot migration progress."""
 502
 503    @abc.abstractmethod
 504    def start_env_migration_progress(self, total_tasks: int) -> None:
 505        """Indicates that a new environment migration progress has begun."""
 506
 507    @abc.abstractmethod
 508    def update_env_migration_progress(self, num_tasks: int) -> None:
 509        """Update the environment migration progress."""
 510
 511    @abc.abstractmethod
 512    def stop_env_migration_progress(self, success: bool = True) -> None:
 513        """Stop the environment migration progress."""
 514
 515    @abc.abstractmethod
 516    def plan(
 517        self,
 518        plan_builder: PlanBuilder,
 519        auto_apply: bool,
 520        default_catalog: t.Optional[str],
 521        no_diff: bool = False,
 522        no_prompts: bool = False,
 523    ) -> None:
 524        """The main plan flow.
 525
 526        The console should present the user with choices on how to backfill and version the snapshots
 527        of a plan.
 528
 529        Args:
 530            plan: The plan to make choices for.
 531            auto_apply: Whether to automatically apply the plan after all choices have been made.
 532            no_diff: Hide text differences for changed models.
 533            no_prompts: Whether to disable interactive prompts for the backfill time range. Please note that
 534                if this flag is set to true and there are uncategorized changes the plan creation will
 535                fail. Default: False
 536        """
 537
 538    @abc.abstractmethod
 539    def show_sql(self, sql: str) -> None:
 540        """Display to the user SQL."""
 541
 542    @abc.abstractmethod
 543    def log_status_update(self, message: str) -> None:
 544        """Display general status update to the user."""
 545
 546    @abc.abstractmethod
 547    def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
 548        """Display list of models skipped during evaluation to the user."""
 549
 550    @abc.abstractmethod
 551    def log_failed_models(self, errors: t.List[NodeExecutionFailedError]) -> None:
 552        """Display list of models that failed during evaluation to the user."""
 553
 554    @abc.abstractmethod
 555    def log_models_updated_during_restatement(
 556        self,
 557        snapshots: t.List[t.Tuple[SnapshotTableInfo, SnapshotTableInfo]],
 558        environment_naming_info: EnvironmentNamingInfo,
 559        default_catalog: t.Optional[str],
 560    ) -> None:
 561        """Display a list of models where new versions got deployed to the specified :environment while we were restating data the old versions
 562
 563        Args:
 564            snapshots: a list of (snapshot_we_restated, snapshot_it_got_replaced_with_during_restatement) tuples
 565            environment: which environment got updated while we were restating models
 566            environment_naming_info: how snapshots are named in that :environment (for display name purposes)
 567            default_catalog: the configured default catalog (for display name purposes)
 568        """
 569
 570    @abc.abstractmethod
 571    def loading_start(self, message: t.Optional[str] = None) -> uuid.UUID:
 572        """Starts loading and returns a unique ID that can be used to stop the loading. Optionally can display a message."""
 573
 574    @abc.abstractmethod
 575    def loading_stop(self, id: uuid.UUID) -> None:
 576        """Stop loading for the given id."""
 577
 578
 579class NoopConsole(Console):
 580    def start_plan_evaluation(self, plan: EvaluatablePlan) -> None:
 581        pass
 582
 583    def stop_plan_evaluation(self) -> None:
 584        pass
 585
 586    def start_evaluation_progress(
 587        self,
 588        batched_intervals: t.Dict[Snapshot, Intervals],
 589        environment_naming_info: EnvironmentNamingInfo,
 590        default_catalog: t.Optional[str],
 591        audit_only: bool = False,
 592    ) -> None:
 593        pass
 594
 595    def start_snapshot_evaluation_progress(
 596        self, snapshot: Snapshot, audit_only: bool = False
 597    ) -> None:
 598        pass
 599
 600    def update_snapshot_evaluation_progress(
 601        self,
 602        snapshot: Snapshot,
 603        interval: Interval,
 604        batch_idx: int,
 605        duration_ms: t.Optional[int],
 606        num_audits_passed: int,
 607        num_audits_failed: int,
 608        audit_only: bool = False,
 609        execution_stats: t.Optional[QueryExecutionStats] = None,
 610        auto_restatement_triggers: t.Optional[t.List[SnapshotId]] = None,
 611    ) -> None:
 612        pass
 613
 614    def stop_evaluation_progress(self, success: bool = True) -> None:
 615        pass
 616
 617    def start_signal_progress(
 618        self,
 619        snapshot: Snapshot,
 620        default_catalog: t.Optional[str],
 621        environment_naming_info: EnvironmentNamingInfo,
 622    ) -> None:
 623        pass
 624
 625    def update_signal_progress(
 626        self,
 627        snapshot: Snapshot,
 628        signal_name: str,
 629        signal_idx: int,
 630        total_signals: int,
 631        ready_intervals: Intervals,
 632        check_intervals: Intervals,
 633        duration: float,
 634    ) -> None:
 635        pass
 636
 637    def stop_signal_progress(self) -> None:
 638        pass
 639
 640    def start_creation_progress(
 641        self,
 642        snapshots: t.List[Snapshot],
 643        environment_naming_info: EnvironmentNamingInfo,
 644        default_catalog: t.Optional[str],
 645    ) -> None:
 646        pass
 647
 648    def update_creation_progress(self, snapshot: SnapshotInfoLike) -> None:
 649        pass
 650
 651    def stop_creation_progress(self, success: bool = True) -> None:
 652        pass
 653
 654    def start_cleanup(self, ignore_ttl: bool) -> bool:
 655        return True
 656
 657    def update_cleanup_progress(self, object_name: str) -> None:
 658        pass
 659
 660    def stop_cleanup(self, success: bool = True) -> None:
 661        pass
 662
 663    def start_promotion_progress(
 664        self,
 665        snapshots: t.List[SnapshotTableInfo],
 666        environment_naming_info: EnvironmentNamingInfo,
 667        default_catalog: t.Optional[str],
 668    ) -> None:
 669        pass
 670
 671    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
 672        pass
 673
 674    def stop_promotion_progress(self, success: bool = True) -> None:
 675        pass
 676
 677    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
 678        pass
 679
 680    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
 681        pass
 682
 683    def log_migration_status(self, success: bool = True) -> None:
 684        pass
 685
 686    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
 687        pass
 688
 689    def start_env_migration_progress(self, total_tasks: int) -> None:
 690        pass
 691
 692    def update_env_migration_progress(self, num_tasks: int) -> None:
 693        pass
 694
 695    def stop_env_migration_progress(self, success: bool = True) -> None:
 696        pass
 697
 698    def start_state_export(
 699        self,
 700        output_file: Path,
 701        gateway: t.Optional[str] = None,
 702        state_connection_config: t.Optional[ConnectionConfig] = None,
 703        environment_names: t.Optional[t.List[str]] = None,
 704        local_only: bool = False,
 705        confirm: bool = True,
 706    ) -> bool:
 707        return confirm
 708
 709    def update_state_export_progress(
 710        self,
 711        version_count: t.Optional[int] = None,
 712        versions_complete: bool = False,
 713        snapshot_count: t.Optional[int] = None,
 714        snapshots_complete: bool = False,
 715        environment_count: t.Optional[int] = None,
 716        environments_complete: bool = False,
 717    ) -> None:
 718        pass
 719
 720    def stop_state_export(self, success: bool, output_file: Path) -> None:
 721        pass
 722
 723    def start_state_import(
 724        self,
 725        input_file: Path,
 726        gateway: str,
 727        state_connection_config: ConnectionConfig,
 728        clear: bool = False,
 729        confirm: bool = True,
 730    ) -> bool:
 731        return confirm
 732
 733    def update_state_import_progress(
 734        self,
 735        timestamp: t.Optional[str] = None,
 736        state_file_version: t.Optional[int] = None,
 737        versions: t.Optional[Versions] = None,
 738        snapshot_count: t.Optional[int] = None,
 739        snapshots_complete: bool = False,
 740        environment_count: t.Optional[int] = None,
 741        environments_complete: bool = False,
 742    ) -> None:
 743        pass
 744
 745    def stop_state_import(self, success: bool, input_file: Path) -> None:
 746        pass
 747
 748    def show_environment_difference_summary(
 749        self,
 750        context_diff: ContextDiff,
 751        no_diff: bool = True,
 752    ) -> None:
 753        pass
 754
 755    def show_model_difference_summary(
 756        self,
 757        context_diff: ContextDiff,
 758        environment_naming_info: EnvironmentNamingInfo,
 759        default_catalog: t.Optional[str],
 760        no_diff: bool = True,
 761    ) -> None:
 762        pass
 763
 764    def plan(
 765        self,
 766        plan_builder: PlanBuilder,
 767        auto_apply: bool,
 768        default_catalog: t.Optional[str],
 769        no_diff: bool = False,
 770        no_prompts: bool = False,
 771    ) -> None:
 772        if auto_apply:
 773            plan_builder.apply()
 774
 775    def log_test_results(self, result: ModelTextTestResult, target_dialect: str) -> None:
 776        pass
 777
 778    def show_sql(self, sql: str) -> None:
 779        pass
 780
 781    def log_status_update(self, message: str) -> None:
 782        pass
 783
 784    def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
 785        pass
 786
 787    def log_failed_models(self, errors: t.List[NodeExecutionFailedError]) -> None:
 788        pass
 789
 790    def log_models_updated_during_restatement(
 791        self,
 792        snapshots: t.List[t.Tuple[SnapshotTableInfo, SnapshotTableInfo]],
 793        environment_naming_info: EnvironmentNamingInfo,
 794        default_catalog: t.Optional[str],
 795    ) -> None:
 796        pass
 797
 798    def log_destructive_change(
 799        self,
 800        snapshot_name: str,
 801        alter_operations: t.List[TableAlterOperation],
 802        dialect: str,
 803        error: bool = True,
 804    ) -> None:
 805        pass
 806
 807    def log_additive_change(
 808        self,
 809        snapshot_name: str,
 810        alter_operations: t.List[TableAlterOperation],
 811        dialect: str,
 812        error: bool = True,
 813    ) -> None:
 814        pass
 815
 816    def log_error(self, message: str) -> None:
 817        pass
 818
 819    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
 820        logger.warning(long_message or short_message)
 821
 822    def log_success(self, message: str) -> None:
 823        pass
 824
 825    def loading_start(self, message: t.Optional[str] = None) -> uuid.UUID:
 826        return uuid.uuid4()
 827
 828    def loading_stop(self, id: uuid.UUID) -> None:
 829        pass
 830
 831    def show_table_diff(
 832        self,
 833        table_diffs: t.List[TableDiff],
 834        show_sample: bool = True,
 835        skip_grain_check: bool = False,
 836        temp_schema: t.Optional[str] = None,
 837    ) -> None:
 838        for table_diff in table_diffs:
 839            self.show_table_diff_summary(table_diff)
 840            self.show_schema_diff(table_diff.schema_diff())
 841            self.show_row_diff(
 842                table_diff.row_diff(temp_schema=temp_schema, skip_grain_check=skip_grain_check),
 843                show_sample=show_sample,
 844                skip_grain_check=skip_grain_check,
 845            )
 846
 847    def update_table_diff_progress(self, model: str) -> None:
 848        pass
 849
 850    def start_table_diff_progress(self, models_to_diff: int) -> None:
 851        pass
 852
 853    def start_table_diff_model_progress(self, model: str) -> None:
 854        pass
 855
 856    def stop_table_diff_progress(self, success: bool) -> None:
 857        pass
 858
 859    def show_table_diff_details(
 860        self,
 861        models_to_diff: t.List[str],
 862    ) -> None:
 863        pass
 864
 865    def show_table_diff_summary(self, table_diff: TableDiff) -> None:
 866        pass
 867
 868    def show_schema_diff(self, schema_diff: SchemaDiff) -> None:
 869        pass
 870
 871    def show_row_diff(
 872        self, row_diff: RowDiff, show_sample: bool = True, skip_grain_check: bool = False
 873    ) -> None:
 874        pass
 875
 876    def print_environments(self, environments_summary: t.List[EnvironmentSummary]) -> None:
 877        pass
 878
 879    def show_intervals(self, snapshot_intervals: t.Dict[Snapshot, SnapshotIntervals]) -> None:
 880        pass
 881
 882    def show_linter_violations(
 883        self, violations: t.List[RuleViolation], model: Model, is_error: bool = False
 884    ) -> None:
 885        pass
 886
 887    def print_connection_config(
 888        self, config: ConnectionConfig, title: t.Optional[str] = "Connection"
 889    ) -> None:
 890        pass
 891
 892    def start_destroy(
 893        self,
 894        schemas_to_delete: t.Optional[t.Set[str]] = None,
 895        views_to_delete: t.Optional[t.Set[str]] = None,
 896        tables_to_delete: t.Optional[t.Set[str]] = None,
 897    ) -> bool:
 898        return True
 899
 900    def stop_destroy(self, success: bool = True) -> None:
 901        pass
 902
 903
 904def make_progress_bar(
 905    message: str,
 906    console: t.Optional[RichConsole] = None,
 907    justify: t.Literal["default", "left", "center", "right", "full"] = "right",
 908) -> Progress:
 909    return Progress(
 910        TextColumn(f"[bold blue]{message}", justify=justify),
 911        BarColumn(bar_width=PROGRESS_BAR_WIDTH),
 912        "[progress.percentage]{task.percentage:>3.1f}%",
 913        "•",
 914        srich.BatchColumn(),
 915        "•",
 916        TimeElapsedColumn(),
 917        console=console,
 918    )
 919
 920
 921class TerminalConsole(Console):
 922    """A rich based implementation of the console."""
 923
 924    TABLE_DIFF_SOURCE_BLUE = "#0248ff"
 925    TABLE_DIFF_TARGET_GREEN = "green"
 926    AUDIT_PASS_MARK = "\u2714"
 927    GREEN_AUDIT_PASS_MARK = f"[green]{AUDIT_PASS_MARK}[/green]"
 928    AUDIT_FAIL_MARK = "\u274c"
 929    AUDIT_PADDING = 0
 930    CHECK_MARK = f"{AUDIT_PASS_MARK} "
 931
 932    def __init__(
 933        self,
 934        console: t.Optional[RichConsole] = None,
 935        verbosity: Verbosity = Verbosity.DEFAULT,
 936        dialect: DialectType = None,
 937        ignore_warnings: bool = False,
 938        **kwargs: t.Any,
 939    ) -> None:
 940        self.console: RichConsole = console or srich.console
 941
 942        self.evaluation_progress_live: t.Optional[Live] = None
 943        self.evaluation_total_progress: t.Optional[Progress] = None
 944        self.evaluation_total_task: t.Optional[TaskID] = None
 945        self.evaluation_model_progress: t.Optional[Progress] = None
 946        self.evaluation_model_tasks: t.Dict[str, TaskID] = {}
 947        self.evaluation_model_batch_sizes: t.Dict[Snapshot, int] = {}
 948        self.evaluation_column_widths: t.Dict[str, int] = {}
 949
 950        # Put in temporary values that are replaced when evaluating
 951        self.environment_naming_info = EnvironmentNamingInfo()
 952        self.default_catalog: t.Optional[str] = None
 953
 954        self.creation_progress: t.Optional[Progress] = None
 955        self.creation_column_widths: t.Dict[str, int] = {}
 956        self.creation_task: t.Optional[TaskID] = None
 957
 958        self.promotion_progress: t.Optional[Progress] = None
 959        self.promotion_column_widths: t.Dict[str, int] = {}
 960        self.promotion_task: t.Optional[TaskID] = None
 961
 962        self.migration_progress: t.Optional[Progress] = None
 963        self.migration_task: t.Optional[TaskID] = None
 964
 965        self.env_migration_progress: t.Optional[Progress] = None
 966        self.env_migration_task: t.Optional[TaskID] = None
 967
 968        self.loading_status: t.Dict[uuid.UUID, Status] = {}
 969
 970        self.state_export_progress: t.Optional[Progress] = None
 971        self.state_export_version_task: t.Optional[TaskID] = None
 972        self.state_export_snapshot_task: t.Optional[TaskID] = None
 973        self.state_export_environment_task: t.Optional[TaskID] = None
 974
 975        self.state_import_progress: t.Optional[Progress] = None
 976        self.state_import_version_task: t.Optional[TaskID] = None
 977        self.state_import_snapshot_task: t.Optional[TaskID] = None
 978        self.state_import_environment_task: t.Optional[TaskID] = None
 979
 980        self.table_diff_progress: t.Optional[Progress] = None
 981        self.table_diff_model_progress: t.Optional[Progress] = None
 982        self.table_diff_model_tasks: t.Dict[str, TaskID] = {}
 983        self.table_diff_progress_live: t.Optional[Live] = None
 984
 985        self.signal_progress_logged = False
 986        self.signal_status_tree: t.Optional[Tree] = None
 987
 988        self.verbosity = verbosity
 989        self.dialect = dialect
 990        self.ignore_warnings = ignore_warnings
 991
 992    def _limit_model_names(self, tree: Tree, verbosity: Verbosity = Verbosity.DEFAULT) -> Tree:
 993        """Trim long indirectly modified model lists below threshold."""
 994        modified_length = len(tree.children)
 995        if (
 996            verbosity < Verbosity.VERY_VERBOSE
 997            and modified_length > self.INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD
 998        ):
 999            tree.children = [
1000                tree.children[0],
1001                Tree(f".... {modified_length - 2} more ...."),
1002                tree.children[-1],
1003            ]
1004        return tree
1005
1006    def _print(self, value: t.Any, **kwargs: t.Any) -> None:
1007        self.console.print(value, **kwargs)
1008
1009    def _prompt(self, message: str, **kwargs: t.Any) -> t.Any:
1010        return Prompt.ask(message, console=self.console, **kwargs)
1011
1012    def _confirm(self, message: str, **kwargs: t.Any) -> bool:
1013        return Confirm.ask(message, console=self.console, **kwargs)
1014
1015    def start_plan_evaluation(self, plan: EvaluatablePlan) -> None:
1016        pass
1017
1018    def stop_plan_evaluation(self) -> None:
1019        pass
1020
1021    def start_evaluation_progress(
1022        self,
1023        batched_intervals: t.Dict[Snapshot, Intervals],
1024        environment_naming_info: EnvironmentNamingInfo,
1025        default_catalog: t.Optional[str],
1026        audit_only: bool = False,
1027    ) -> None:
1028        """Indicates that a new snapshot evaluation/auditing progress has begun."""
1029        # Add a newline to separate signal checking from evaluation
1030        if self.signal_progress_logged:
1031            self._print("")
1032
1033        if not self.evaluation_progress_live:
1034            self.evaluation_total_progress = make_progress_bar(
1035                "Executing model batches" if not audit_only else "Auditing models", self.console
1036            )
1037
1038            self.evaluation_model_progress = Progress(
1039                TextColumn("{task.fields[view_name]}", justify="right"),
1040                SpinnerColumn(spinner_name="simpleDots"),
1041                console=self.console,
1042            )
1043
1044            progress_table = Table.grid()
1045            progress_table.add_row(self.evaluation_total_progress)
1046            progress_table.add_row(self.evaluation_model_progress)
1047
1048            self.evaluation_progress_live = Live(
1049                progress_table, console=self.console, refresh_per_second=10
1050            )
1051            self.evaluation_progress_live.start()
1052
1053            batch_sizes = {
1054                snapshot: len(intervals) for snapshot, intervals in batched_intervals.items()
1055            }
1056            message = "Executing" if not audit_only else "Auditing"
1057            self.evaluation_total_task = self.evaluation_total_progress.add_task(
1058                f"{message} models...", total=sum(batch_sizes.values())
1059            )
1060
1061            # determine column widths
1062            self.evaluation_column_widths["annotation"] = (
1063                _calculate_annotation_str_len(
1064                    batched_intervals, self.AUDIT_PADDING, len(" (123.4m rows, 123.4 KiB)")
1065                )
1066                + 3  # brackets and opening escape backslash
1067            )
1068            self.evaluation_column_widths["name"] = max(
1069                len(
1070                    snapshot.display_name(
1071                        environment_naming_info,
1072                        default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1073                        dialect=self.dialect,
1074                    )
1075                )
1076                for snapshot in batched_intervals
1077            )
1078            largest_batch_size = max(batch_sizes.values())
1079            self.evaluation_column_widths["batch"] = len(str(largest_batch_size)) * 2 + 3  # [X/X]
1080            self.evaluation_column_widths["duration"] = 8
1081
1082            self.evaluation_model_batch_sizes = batch_sizes
1083            self.environment_naming_info = environment_naming_info
1084            self.default_catalog = default_catalog
1085
1086    def start_snapshot_evaluation_progress(
1087        self, snapshot: Snapshot, audit_only: bool = False
1088    ) -> None:
1089        if self.evaluation_model_progress and snapshot.name not in self.evaluation_model_tasks:
1090            display_name = snapshot.display_name(
1091                self.environment_naming_info,
1092                self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1093                dialect=self.dialect,
1094            )
1095            self.evaluation_model_tasks[snapshot.name] = self.evaluation_model_progress.add_task(
1096                f"{'Evaluating' if not audit_only else 'Auditing'} {display_name}...",
1097                view_name=display_name,
1098                total=self.evaluation_model_batch_sizes[snapshot],
1099            )
1100
1101    def update_snapshot_evaluation_progress(
1102        self,
1103        snapshot: Snapshot,
1104        interval: Interval,
1105        batch_idx: int,
1106        duration_ms: t.Optional[int],
1107        num_audits_passed: int,
1108        num_audits_failed: int,
1109        audit_only: bool = False,
1110        execution_stats: t.Optional[QueryExecutionStats] = None,
1111        auto_restatement_triggers: t.Optional[t.List[SnapshotId]] = None,
1112    ) -> None:
1113        """Update the snapshot evaluation progress."""
1114        if (
1115            self.evaluation_total_progress
1116            and self.evaluation_model_progress
1117            and self.evaluation_progress_live
1118        ):
1119            total_batches = self.evaluation_model_batch_sizes[snapshot]
1120            batch_num = str(batch_idx + 1).rjust(len(str(total_batches)))
1121            batch = f"[{batch_num}/{total_batches}]".ljust(self.evaluation_column_widths["batch"])
1122
1123            if duration_ms:
1124                display_name = snapshot.display_name(
1125                    self.environment_naming_info,
1126                    self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1127                    dialect=self.dialect,
1128                ).ljust(self.evaluation_column_widths["name"])
1129
1130                annotation = _create_evaluation_model_annotation(
1131                    snapshot, _format_evaluation_model_interval(snapshot, interval), execution_stats
1132                )
1133                audits_str = ""
1134                if num_audits_passed:
1135                    audits_str += f" {self.AUDIT_PASS_MARK}{num_audits_passed}"
1136                if num_audits_failed:
1137                    audits_str += f" {self.AUDIT_FAIL_MARK}{num_audits_failed}"
1138                audits_str = f", audits{audits_str}" if audits_str else ""
1139                annotation_len = self.evaluation_column_widths["annotation"]
1140                # don't adjust the annotation_len if we're using AUDIT_PADDING
1141                annotation = f"\\[{annotation + audits_str}]".ljust(
1142                    annotation_len - 1
1143                    if num_audits_failed and self.AUDIT_PADDING == 0
1144                    else annotation_len
1145                )
1146
1147                duration = f"{(duration_ms / 1000.0):.2f}s".ljust(
1148                    self.evaluation_column_widths["duration"]
1149                )
1150
1151                msg = f"{f'{batch} ' if not audit_only else ''}{display_name}   {annotation}   {duration}".replace(
1152                    self.AUDIT_PASS_MARK, self.GREEN_AUDIT_PASS_MARK
1153                )
1154
1155                self.evaluation_progress_live.console.print(msg)
1156
1157            self.evaluation_total_progress.update(
1158                self.evaluation_total_task or TaskID(0), refresh=True, advance=1
1159            )
1160
1161            model_task_id = self.evaluation_model_tasks[snapshot.name]
1162            self.evaluation_model_progress.update(model_task_id, refresh=True, advance=1)
1163            if (
1164                self.evaluation_model_progress._tasks[model_task_id].completed >= total_batches
1165                or audit_only
1166            ):
1167                self.evaluation_model_progress.remove_task(model_task_id)
1168
1169    def stop_evaluation_progress(self, success: bool = True) -> None:
1170        """Stop the snapshot evaluation progress."""
1171        if self.evaluation_progress_live:
1172            self.evaluation_progress_live.stop()
1173            if success:
1174                self.log_success(f"{self.CHECK_MARK}Model batches executed")
1175
1176        self.evaluation_progress_live = None
1177        self.evaluation_total_progress = None
1178        self.evaluation_total_task = None
1179        self.evaluation_model_progress = None
1180        self.evaluation_model_tasks = {}
1181        self.evaluation_model_batch_sizes = {}
1182        self.evaluation_column_widths = {}
1183        self.environment_naming_info = EnvironmentNamingInfo()
1184        self.default_catalog = None
1185
1186    def start_signal_progress(
1187        self,
1188        snapshot: Snapshot,
1189        default_catalog: t.Optional[str],
1190        environment_naming_info: EnvironmentNamingInfo,
1191    ) -> None:
1192        """Indicates that signal checking has begun for a snapshot."""
1193        display_name = snapshot.display_name(
1194            environment_naming_info,
1195            default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1196            dialect=self.dialect,
1197        )
1198        self.signal_status_tree = Tree(f"Checking signals for {display_name}")
1199
1200    def update_signal_progress(
1201        self,
1202        snapshot: Snapshot,
1203        signal_name: str,
1204        signal_idx: int,
1205        total_signals: int,
1206        ready_intervals: Intervals,
1207        check_intervals: Intervals,
1208        duration: float,
1209    ) -> None:
1210        """Updates the signal checking progress."""
1211        tree = Tree(f"[{signal_idx + 1}/{total_signals}] {signal_name} {duration:.2f}s")
1212
1213        formatted_check_intervals = [_format_signal_interval(snapshot, i) for i in check_intervals]
1214        formatted_ready_intervals = [_format_signal_interval(snapshot, i) for i in ready_intervals]
1215
1216        if not formatted_check_intervals:
1217            formatted_check_intervals = ["no intervals"]
1218        if not formatted_ready_intervals:
1219            formatted_ready_intervals = ["no intervals"]
1220
1221        # Color coding to help detect partial interval ranges quickly
1222        if ready_intervals == check_intervals:
1223            msg = "All ready"
1224            color = "green"
1225        elif ready_intervals:
1226            msg = "Some ready"
1227            color = "yellow"
1228        else:
1229            msg = "None ready"
1230            color = "red"
1231
1232        if self.verbosity < Verbosity.VERY_VERBOSE:
1233            num_check_intervals = len(formatted_check_intervals)
1234            if num_check_intervals > 3:
1235                formatted_check_intervals = formatted_check_intervals[:3]
1236                formatted_check_intervals.append(f"... and {num_check_intervals - 3} more")
1237
1238            num_ready_intervals = len(formatted_ready_intervals)
1239            if num_ready_intervals > 3:
1240                formatted_ready_intervals = formatted_ready_intervals[:3]
1241                formatted_ready_intervals.append(f"... and {num_ready_intervals - 3} more")
1242
1243            check = ", ".join(formatted_check_intervals)
1244            tree.add(f"Check: {check}")
1245
1246            ready = ", ".join(formatted_ready_intervals)
1247            tree.add(f"[{color}]{msg}: {ready}[/{color}]")
1248        else:
1249            check_tree = Tree("Check")
1250            tree.add(check_tree)
1251            for interval in formatted_check_intervals:
1252                check_tree.add(interval)
1253
1254            ready_tree = Tree(f"[{color}]{msg}[/{color}]")
1255            tree.add(ready_tree)
1256            for interval in formatted_ready_intervals:
1257                ready_tree.add(f"[{color}]{interval}[/{color}]")
1258
1259        if self.signal_status_tree is not None:
1260            self.signal_status_tree.add(tree)
1261
1262    def stop_signal_progress(self) -> None:
1263        """Indicates that signal checking has completed for a snapshot."""
1264        if self.signal_status_tree is not None:
1265            self._print(self.signal_status_tree)
1266            self.signal_status_tree = None
1267            self.signal_progress_logged = True
1268
1269    def start_creation_progress(
1270        self,
1271        snapshots: t.List[Snapshot],
1272        environment_naming_info: EnvironmentNamingInfo,
1273        default_catalog: t.Optional[str],
1274    ) -> None:
1275        """Indicates that a new creation progress has begun."""
1276        if self.creation_progress is None:
1277            self.creation_progress = make_progress_bar("Updating physical layer", self.console)
1278
1279            self._print("")
1280            self.creation_progress.start()
1281            self.creation_task = self.creation_progress.add_task(
1282                "Updating physical layer...",
1283                total=len(snapshots),
1284            )
1285
1286            # determine name column widths if we're printing name
1287            if self.verbosity >= Verbosity.VERBOSE:
1288                self.creation_column_widths["name"] = max(
1289                    len(
1290                        snapshot.display_name(
1291                            environment_naming_info,
1292                            default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1293                            dialect=self.dialect,
1294                        )
1295                    )
1296                    for snapshot in snapshots
1297                )
1298
1299            self.environment_naming_info = environment_naming_info
1300            self.default_catalog = default_catalog
1301
1302    def update_creation_progress(self, snapshot: SnapshotInfoLike) -> None:
1303        """Update the snapshot creation progress."""
1304        if self.creation_progress is not None and self.creation_task is not None:
1305            if self.verbosity >= Verbosity.VERBOSE:
1306                msg = snapshot.display_name(
1307                    self.environment_naming_info,
1308                    self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1309                    dialect=self.dialect,
1310                ).ljust(self.creation_column_widths["name"])
1311                self.creation_progress.live.console.print(msg + "  [green]created[/green]")
1312            self.creation_progress.update(self.creation_task, refresh=True, advance=1)
1313
1314    def stop_creation_progress(self, success: bool = True) -> None:
1315        """Stop the snapshot creation progress."""
1316        self.creation_task = None
1317        if self.creation_progress is not None:
1318            self.creation_progress.stop()
1319            self.creation_progress = None
1320            if success:
1321                self.log_success(f"\n{self.CHECK_MARK}Physical layer updated")
1322
1323        self.environment_naming_info = EnvironmentNamingInfo()
1324        self.default_catalog = None
1325        self.creation_column_widths = {}
1326
1327    def start_cleanup(self, ignore_ttl: bool) -> bool:
1328        if ignore_ttl:
1329            self._print(
1330                "Are you sure you want to delete all snapshots that are not referenced in any environment?"
1331            )
1332            self._print(
1333                "Note that this may cause a race condition if there are any concurrently running plans."
1334            )
1335            self._print(
1336                "It may also confuse users who were expecting to be able to rollback changes in their development environments."
1337            )
1338            if not self._confirm("Proceed?"):
1339                self.log_error("Cleanup aborted")
1340                return False
1341        return True
1342
1343    def update_cleanup_progress(self, object_name: str) -> None:
1344        """Update the snapshot cleanup progress."""
1345        self._print(f"Deleted object {object_name}")
1346
1347    def stop_cleanup(self, success: bool = False) -> None:
1348        if success:
1349            self.log_success("Cleanup complete.")
1350        else:
1351            self.log_error("Cleanup failed!")
1352
1353    def start_destroy(
1354        self,
1355        schemas_to_delete: t.Optional[t.Set[str]] = None,
1356        views_to_delete: t.Optional[t.Set[str]] = None,
1357        tables_to_delete: t.Optional[t.Set[str]] = None,
1358    ) -> bool:
1359        self.log_warning(
1360            "This will permanently delete all engine-managed objects, state tables and SQLMesh cache.\n"
1361            "The operation may disrupt any currently running or scheduled plans.\n"
1362        )
1363
1364        if schemas_to_delete or views_to_delete or tables_to_delete:
1365            if schemas_to_delete:
1366                self.log_error("Schemas to be deleted:")
1367                for schema in sorted(schemas_to_delete):
1368                    self.log_error(f"  • {schema}")
1369
1370            if views_to_delete:
1371                self.log_error("\nEnvironment views to be deleted:")
1372                for view in sorted(views_to_delete):
1373                    self.log_error(f"  • {view}")
1374
1375            if tables_to_delete:
1376                self.log_error("\nSnapshot tables to be deleted:")
1377                for table in sorted(tables_to_delete):
1378                    self.log_error(f"  • {table}")
1379
1380            self.log_error(
1381                "\nThis action will DELETE ALL the above resources managed by SQLMesh AND\n"
1382                "potentially external resources created by other tools in these schemas.\n"
1383            )
1384
1385        if not self._confirm("Are you ABSOLUTELY SURE you want to proceed with deletion?"):
1386            self.log_error("Destroy operation cancelled.")
1387            return False
1388        return True
1389
1390    def stop_destroy(self, success: bool = False) -> None:
1391        if success:
1392            self.log_success("Destroy completed successfully.")
1393        else:
1394            self.log_error("Destroy failed!")
1395
1396    def start_promotion_progress(
1397        self,
1398        snapshots: t.List[SnapshotTableInfo],
1399        environment_naming_info: EnvironmentNamingInfo,
1400        default_catalog: t.Optional[str],
1401    ) -> None:
1402        """Indicates that a new snapshot promotion progress has begun."""
1403        if snapshots and self.promotion_progress is None:
1404            self.promotion_progress = make_progress_bar(
1405                "Updating virtual layer ", self.console, justify="left"
1406            )
1407
1408            snapshots_with_virtual_views = [
1409                s for s in snapshots if s.is_model and not s.is_symbolic
1410            ]
1411            self.promotion_progress.start()
1412            self.promotion_task = self.promotion_progress.add_task(
1413                f"Virtually updating {environment_naming_info.name}...",
1414                total=len(snapshots_with_virtual_views),
1415            )
1416
1417            # determine name column widths if we're printing names
1418            if self.verbosity >= Verbosity.VERBOSE:
1419                self.promotion_column_widths["name"] = max(
1420                    len(
1421                        snapshot.display_name(
1422                            environment_naming_info,
1423                            default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1424                            dialect=self.dialect,
1425                        )
1426                    )
1427                    for snapshot in snapshots_with_virtual_views
1428                )
1429
1430            self.environment_naming_info = environment_naming_info
1431            self.default_catalog = default_catalog
1432
1433    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
1434        """Update the snapshot promotion progress."""
1435        if (
1436            self.promotion_progress is not None
1437            and self.promotion_task is not None
1438            and snapshot.is_model
1439            and not snapshot.is_symbolic
1440        ):
1441            if self.verbosity >= Verbosity.VERBOSE:
1442                display_name = snapshot.display_name(
1443                    self.environment_naming_info,
1444                    self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1445                    dialect=self.dialect,
1446                ).ljust(self.promotion_column_widths["name"])
1447                action_str = ""
1448                if promoted:
1449                    action_str = (
1450                        "[yellow]updated[/yellow]"
1451                        if snapshot.previous_version
1452                        else "[green]created[/green]"
1453                    )
1454                action_str = action_str or "[red]dropped[/red]"
1455                self.promotion_progress.live.console.print(f"{display_name}  {action_str}")
1456            self.promotion_progress.update(self.promotion_task, refresh=True, advance=1)
1457
1458    def stop_promotion_progress(self, success: bool = True) -> None:
1459        """Stop the snapshot promotion progress."""
1460        self.promotion_task = None
1461        if self.promotion_progress is not None:
1462            self.promotion_progress.stop()
1463            self.promotion_progress = None
1464            if success:
1465                self.log_success(f"\n{self.CHECK_MARK}Virtual layer updated")
1466
1467        self.environment_naming_info = EnvironmentNamingInfo()
1468        self.default_catalog = None
1469        self.promotion_column_widths = {}
1470
1471    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
1472        """Indicates that a new snapshot migration progress has begun."""
1473        if self.migration_progress is None:
1474            self.migration_progress = make_progress_bar("Migrating snapshots", self.console)
1475
1476            self.migration_progress.start()
1477            self.migration_task = self.migration_progress.add_task(
1478                "Migrating snapshots...",
1479                total=total_tasks,
1480            )
1481
1482    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
1483        """Update the migration progress."""
1484        if self.migration_progress is not None and self.migration_task is not None:
1485            self.migration_progress.update(self.migration_task, refresh=True, advance=num_tasks)
1486
1487    def log_migration_status(self, success: bool = True) -> None:
1488        """Log the migration status."""
1489        if self.migration_progress is not None:
1490            self.migration_progress = None
1491            if success:
1492                self.log_success("Migration completed successfully")
1493
1494    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
1495        """Stop the migration progress."""
1496        self.migration_task = None
1497        if self.migration_progress is not None:
1498            self.migration_progress.stop()
1499            if success:
1500                self.log_success("Snapshots migrated successfully")
1501
1502    def start_env_migration_progress(self, total_tasks: int) -> None:
1503        """Indicates that a new environment migration has begun."""
1504        if self.env_migration_progress is None:
1505            self.env_migration_progress = make_progress_bar("Migrating environments", self.console)
1506            self.env_migration_progress.start()
1507            self.env_migration_task = self.env_migration_progress.add_task(
1508                "Migrating environments...",
1509                total=total_tasks,
1510            )
1511
1512    def update_env_migration_progress(self, num_tasks: int) -> None:
1513        """Update the environment migration progress."""
1514        if self.env_migration_progress is not None and self.env_migration_task is not None:
1515            self.env_migration_progress.update(
1516                self.env_migration_task, refresh=True, advance=num_tasks
1517            )
1518
1519    def stop_env_migration_progress(self, success: bool = True) -> None:
1520        """Stop the environment migration progress."""
1521        self.env_migration_task = None
1522        if self.env_migration_progress is not None:
1523            self.env_migration_progress.stop()
1524            self.env_migration_progress = None
1525            if success:
1526                self.log_success("Environments migrated successfully")
1527
1528    def start_state_export(
1529        self,
1530        output_file: Path,
1531        gateway: t.Optional[str] = None,
1532        state_connection_config: t.Optional[ConnectionConfig] = None,
1533        environment_names: t.Optional[t.List[str]] = None,
1534        local_only: bool = False,
1535        confirm: bool = True,
1536    ) -> bool:
1537        self.state_export_progress = None
1538
1539        if local_only:
1540            self.log_status_update(f"Exporting [b]local[/b] state to '{output_file.as_posix()}'\n")
1541            self.log_warning(
1542                "Local state exports just contain the model versions in your local context. Therefore, the resulting file cannot be imported."
1543            )
1544        else:
1545            self.log_status_update(
1546                f"Exporting state to '{output_file.as_posix()}' from the following connection:\n"
1547            )
1548            if gateway:
1549                self.log_status_update(f"[b]Gateway[/b]: [green]{gateway}[/green]")
1550            if state_connection_config:
1551                self.print_connection_config(state_connection_config, title="State Connection")
1552            if environment_names:
1553                heading = "Environments" if len(environment_names) > 1 else "Environment"
1554                self.log_status_update(
1555                    f"[b]{heading}[/b]: [yellow]{', '.join(environment_names)}[/yellow]"
1556                )
1557
1558        should_continue = True
1559        if confirm:
1560            should_continue = self._confirm("\nContinue?")
1561            self.log_status_update("")
1562
1563        if should_continue:
1564            self.state_export_progress = make_progress_bar("{task.description}", self.console)
1565            assert isinstance(self.state_export_progress, Progress)
1566
1567            self.state_export_version_task = self.state_export_progress.add_task(
1568                "Exporting versions", start=False
1569            )
1570            self.state_export_snapshot_task = self.state_export_progress.add_task(
1571                "Exporting snapshots", start=False
1572            )
1573            self.state_export_environment_task = self.state_export_progress.add_task(
1574                "Exporting environments", start=False
1575            )
1576
1577            self.state_export_progress.start()
1578
1579        return should_continue
1580
1581    def update_state_export_progress(
1582        self,
1583        version_count: t.Optional[int] = None,
1584        versions_complete: bool = False,
1585        snapshot_count: t.Optional[int] = None,
1586        snapshots_complete: bool = False,
1587        environment_count: t.Optional[int] = None,
1588        environments_complete: bool = False,
1589    ) -> None:
1590        if self.state_export_progress:
1591            if self.state_export_version_task is not None:
1592                if version_count is not None:
1593                    self.state_export_progress.start_task(self.state_export_version_task)
1594                    self.state_export_progress.update(
1595                        self.state_export_version_task,
1596                        total=version_count,
1597                        completed=version_count,
1598                        refresh=True,
1599                    )
1600                if versions_complete:
1601                    self.state_export_progress.stop_task(self.state_export_version_task)
1602
1603            if self.state_export_snapshot_task is not None:
1604                if snapshot_count is not None:
1605                    self.state_export_progress.start_task(self.state_export_snapshot_task)
1606                    self.state_export_progress.update(
1607                        self.state_export_snapshot_task,
1608                        total=snapshot_count,
1609                        completed=snapshot_count,
1610                        refresh=True,
1611                    )
1612                if snapshots_complete:
1613                    self.state_export_progress.stop_task(self.state_export_snapshot_task)
1614
1615            if self.state_export_environment_task is not None:
1616                if environment_count is not None:
1617                    self.state_export_progress.start_task(self.state_export_environment_task)
1618                    self.state_export_progress.update(
1619                        self.state_export_environment_task,
1620                        total=environment_count,
1621                        completed=environment_count,
1622                        refresh=True,
1623                    )
1624                if environments_complete:
1625                    self.state_export_progress.stop_task(self.state_export_environment_task)
1626
1627    def stop_state_export(self, success: bool, output_file: Path) -> None:
1628        if self.state_export_progress:
1629            self.state_export_progress.stop()
1630            self.state_export_progress = None
1631
1632            self.log_status_update("")
1633
1634            if success:
1635                self.log_success(f"State exported successfully to '{output_file.as_posix()}'")
1636            else:
1637                self.log_error("State export failed!")
1638
1639    def start_state_import(
1640        self,
1641        input_file: Path,
1642        gateway: str,
1643        state_connection_config: ConnectionConfig,
1644        clear: bool = False,
1645        confirm: bool = True,
1646    ) -> bool:
1647        self.log_status_update(
1648            f"Loading state from '{input_file.as_posix()}' into the following connection:\n"
1649        )
1650        self.log_status_update(f"[b]Gateway[/b]: [green]{gateway}[/green]")
1651        self.print_connection_config(state_connection_config, title="State Connection")
1652        self.log_status_update("")
1653
1654        if clear:
1655            self.log_warning(
1656                f"This [b]destructive[/b] operation will delete all existing state against the '{gateway}' gateway \n"
1657                f"and replace it with what's in the '{input_file.as_posix()}' file.\n"
1658            )
1659        else:
1660            self.log_warning(
1661                f"This operation will [b]merge[/b] the contents of the state file to the state located at the '{gateway}' gateway.\n"
1662                "Matching snapshots or environments will be replaced.\n"
1663                "Non-matching snapshots or environments will be ignored.\n"
1664            )
1665
1666        should_continue = True
1667        if confirm:
1668            should_continue = self._confirm("[red]Are you sure?[/red]")
1669            self.log_status_update("")
1670
1671        if should_continue:
1672            self.state_import_progress = make_progress_bar("{task.description}", self.console)
1673
1674            self.state_import_info = Tree("[bold]State File Information:")
1675
1676            self.state_import_version_task = self.state_import_progress.add_task(
1677                "Importing versions", start=False
1678            )
1679            self.state_import_snapshot_task = self.state_import_progress.add_task(
1680                "Importing snapshots", start=False
1681            )
1682            self.state_import_environment_task = self.state_import_progress.add_task(
1683                "Importing environments", start=False
1684            )
1685
1686            self.state_import_progress.start()
1687
1688        return should_continue
1689
1690    def update_state_import_progress(
1691        self,
1692        timestamp: t.Optional[str] = None,
1693        state_file_version: t.Optional[int] = None,
1694        versions: t.Optional[Versions] = None,
1695        snapshot_count: t.Optional[int] = None,
1696        snapshots_complete: bool = False,
1697        environment_count: t.Optional[int] = None,
1698        environments_complete: bool = False,
1699    ) -> None:
1700        if self.state_import_progress:
1701            if self.state_import_info:
1702                if timestamp:
1703                    self.state_import_info.add(f"Creation Timestamp: {timestamp}")
1704                if state_file_version:
1705                    self.state_import_info.add(f"File Version: {state_file_version}")
1706                if versions:
1707                    self.state_import_info.add(f"SQLMesh version: {versions.sqlmesh_version}")
1708                    self.state_import_info.add(
1709                        f"SQLMesh migration version: {versions.schema_version}"
1710                    )
1711                    self.state_import_info.add(f"SQLGlot version: {versions.sqlglot_version}\n")
1712
1713                    self._print(self.state_import_info)
1714
1715                    version_count = len(versions.model_dump())
1716
1717                    if self.state_import_version_task is not None:
1718                        self.state_import_progress.start_task(self.state_import_version_task)
1719                        self.state_import_progress.update(
1720                            self.state_import_version_task,
1721                            total=version_count,
1722                            completed=version_count,
1723                        )
1724                        self.state_import_progress.stop_task(self.state_import_version_task)
1725
1726            if self.state_import_snapshot_task is not None:
1727                if snapshot_count is not None:
1728                    self.state_import_progress.start_task(self.state_import_snapshot_task)
1729                    self.state_import_progress.update(
1730                        self.state_import_snapshot_task,
1731                        completed=snapshot_count,
1732                        total=snapshot_count,
1733                        refresh=True,
1734                    )
1735
1736                if snapshots_complete:
1737                    self.state_import_progress.stop_task(self.state_import_snapshot_task)
1738
1739            if self.state_import_environment_task is not None:
1740                if environment_count is not None:
1741                    self.state_import_progress.start_task(self.state_import_environment_task)
1742                    self.state_import_progress.update(
1743                        self.state_import_environment_task,
1744                        completed=environment_count,
1745                        total=environment_count,
1746                        refresh=True,
1747                    )
1748
1749                if environments_complete:
1750                    self.state_import_progress.stop_task(self.state_import_environment_task)
1751
1752    def stop_state_import(self, success: bool, input_file: Path) -> None:
1753        if self.state_import_progress:
1754            self.state_import_progress.stop()
1755            self.state_import_progress = None
1756
1757            self.log_status_update("")
1758
1759            if success:
1760                self.log_success(f"State imported successfully from '{input_file.as_posix()}'")
1761            else:
1762                self.log_error("State import failed!")
1763
1764    def show_environment_difference_summary(
1765        self,
1766        context_diff: ContextDiff,
1767        no_diff: bool = True,
1768    ) -> None:
1769        """Shows a summary of the environment differences.
1770
1771        Args:
1772            context_diff: The context diff to use to print the summary
1773            no_diff: Hide the actual environment statement differences.
1774        """
1775        if context_diff.is_new_environment:
1776            msg = (
1777                f"\n`{context_diff.environment}` environment will be initialized"
1778                if not context_diff.create_from_env_exists
1779                else f"\nNew environment `{context_diff.environment}` will be created from `{context_diff.create_from}`"
1780            )
1781            self._print(Tree(f"[bold]{msg}\n"))
1782            if not context_diff.has_snapshot_changes:
1783                return
1784
1785        if not context_diff.has_changes:
1786            # This is only reached when the plan is against an existing environment, so we use the environment
1787            #   name instead of the create_from name. The equivalent message for new environments happens in
1788            #   the PlanBuilder.
1789            self._print(
1790                Tree(
1791                    f"\n[bold]No changes to plan: project files match the `{context_diff.environment}` environment\n"
1792                )
1793            )
1794            return
1795
1796        if not context_diff.is_new_environment or (
1797            context_diff.is_new_environment and context_diff.create_from_env_exists
1798        ):
1799            self._print(
1800                Tree(
1801                    f"\n[bold]Differences from the `{context_diff.create_from if context_diff.is_new_environment else context_diff.environment}` environment:\n"
1802                )
1803            )
1804
1805        if context_diff.has_requirement_changes:
1806            self._print(f"[bold]Requirements:\n{context_diff.requirements_diff()}")
1807
1808        if context_diff.has_environment_statements_changes and not no_diff:
1809            self._print("[bold]Environment statements:\n")
1810            for type, diff in context_diff.environment_statements_diff(
1811                include_python_env=not context_diff.is_new_environment
1812            ):
1813                self._print(Syntax(diff, type, line_numbers=False))
1814
1815    def show_model_difference_summary(
1816        self,
1817        context_diff: ContextDiff,
1818        environment_naming_info: EnvironmentNamingInfo,
1819        default_catalog: t.Optional[str],
1820        no_diff: bool = True,
1821    ) -> None:
1822        """Shows a summary of the model differences.
1823
1824        Args:
1825            context_diff: The context diff to use to print the summary
1826            environment_naming_info: The environment naming info to reference when printing model names
1827            default_catalog: The default catalog to reference when deciding to remove catalog from display names
1828            no_diff: Hide the actual SQL differences.
1829        """
1830        self._show_summary_tree_for(
1831            context_diff,
1832            "Models",
1833            lambda x: x.is_model,
1834            environment_naming_info,
1835            default_catalog,
1836            no_diff=no_diff,
1837        )
1838        self._show_summary_tree_for(
1839            context_diff,
1840            "Standalone Audits",
1841            lambda x: x.is_audit,
1842            environment_naming_info,
1843            default_catalog,
1844            no_diff=no_diff,
1845        )
1846
1847    def plan(
1848        self,
1849        plan_builder: PlanBuilder,
1850        auto_apply: bool,
1851        default_catalog: t.Optional[str],
1852        no_diff: bool = False,
1853        no_prompts: bool = False,
1854    ) -> None:
1855        """The main plan flow.
1856
1857        The console should present the user with choices on how to backfill and version the snapshots
1858        of a plan.
1859
1860        Args:
1861            plan: The plan to make choices for.
1862            auto_apply: Whether to automatically apply the plan after all choices have been made.
1863            default_catalog: The default catalog to reference when deciding to remove catalog from display names
1864            no_diff: Hide text differences for changed models.
1865            no_prompts: Whether to disable interactive prompts for the backfill time range. Please note that
1866                if this flag is set to true and there are uncategorized changes the plan creation will
1867                fail. Default: False
1868        """
1869        self._prompt_categorize(
1870            plan_builder,
1871            auto_apply,
1872            no_diff=no_diff,
1873            no_prompts=no_prompts,
1874            default_catalog=default_catalog,
1875        )
1876
1877        self._show_options_after_categorization(
1878            plan_builder, auto_apply, default_catalog=default_catalog, no_prompts=no_prompts
1879        )
1880
1881        if auto_apply:
1882            plan_builder.apply()
1883
1884    def _show_summary_tree_for(
1885        self,
1886        context_diff: ContextDiff,
1887        header: str,
1888        snapshot_selector: t.Callable[[SnapshotInfoLike], bool],
1889        environment_naming_info: EnvironmentNamingInfo,
1890        default_catalog: t.Optional[str],
1891        no_diff: bool = True,
1892    ) -> None:
1893        added_snapshot_ids = {
1894            s_id for s_id in context_diff.added if snapshot_selector(context_diff.snapshots[s_id])
1895        }
1896        removed_snapshot_ids = {
1897            s_id
1898            for s_id, snapshot in context_diff.removed_snapshots.items()
1899            if snapshot_selector(snapshot)
1900        }
1901        modified_snapshot_ids = {
1902            current_snapshot.snapshot_id
1903            for _, (current_snapshot, _) in context_diff.modified_snapshots.items()
1904            if snapshot_selector(current_snapshot)
1905        }
1906
1907        tree_sets = (
1908            added_snapshot_ids,
1909            removed_snapshot_ids,
1910            modified_snapshot_ids,
1911        )
1912        if all(not s_ids for s_ids in tree_sets):
1913            return
1914
1915        tree = Tree(f"[bold]{header}:")
1916        if added_snapshot_ids:
1917            added_tree = Tree("[bold][added]Added:")
1918            for s_id in sorted(added_snapshot_ids):
1919                snapshot = context_diff.snapshots[s_id]
1920                added_tree.add(
1921                    f"[added]{snapshot.display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}"
1922                )
1923            tree.add(self._limit_model_names(added_tree, self.verbosity))
1924        if removed_snapshot_ids:
1925            removed_tree = Tree("[bold][removed]Removed:")
1926            for s_id in sorted(removed_snapshot_ids):
1927                snapshot_table_info = context_diff.removed_snapshots[s_id]
1928                removed_tree.add(
1929                    f"[removed]{snapshot_table_info.display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}"
1930                )
1931            tree.add(self._limit_model_names(removed_tree, self.verbosity))
1932        if modified_snapshot_ids:
1933            tree = self._add_modified_models(
1934                context_diff,
1935                modified_snapshot_ids,
1936                tree,
1937                environment_naming_info,
1938                default_catalog,
1939                no_diff,
1940            )
1941
1942        self._print(tree)
1943
1944    def _add_modified_models(
1945        self,
1946        context_diff: ContextDiff,
1947        modified_snapshot_ids: t.Set[SnapshotId],
1948        tree: Tree,
1949        environment_naming_info: EnvironmentNamingInfo,
1950        default_catalog: t.Optional[str] = None,
1951        no_diff: bool = True,
1952    ) -> Tree:
1953        direct = Tree("[bold][direct]Directly Modified:")
1954        indirect = Tree("[bold][indirect]Indirectly Modified:")
1955        metadata = Tree("[bold][metadata]Metadata Updated:")
1956        for s_id in modified_snapshot_ids:
1957            name = s_id.name
1958            display_name = context_diff.snapshots[s_id].display_name(
1959                environment_naming_info,
1960                default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
1961                dialect=self.dialect,
1962            )
1963            if context_diff.directly_modified(name):
1964                direct.add(
1965                    f"[direct]{display_name}"
1966                    if no_diff
1967                    else Syntax(f"{display_name}\n{context_diff.text_diff(name)}", "sql")
1968                )
1969            elif context_diff.indirectly_modified(name):
1970                indirect.add(f"[indirect]{display_name}")
1971            elif context_diff.metadata_updated(name):
1972                metadata.add(
1973                    f"[metadata]{display_name}"
1974                    if no_diff
1975                    else Syntax(f"{display_name}\n{context_diff.text_diff(name)}", "sql")
1976                )
1977        if direct.children:
1978            tree.add(direct)
1979        if indirect.children:
1980            tree.add(self._limit_model_names(indirect, self.verbosity))
1981        if metadata.children:
1982            tree.add(metadata)
1983        return tree
1984
1985    def _show_options_after_categorization(
1986        self,
1987        plan_builder: PlanBuilder,
1988        auto_apply: bool,
1989        default_catalog: t.Optional[str],
1990        no_prompts: bool,
1991    ) -> None:
1992        plan = plan_builder.build()
1993        if not no_prompts and plan.forward_only and plan.new_snapshots:
1994            self._prompt_effective_from(plan_builder, auto_apply, default_catalog)
1995
1996        if plan.requires_backfill:
1997            self._show_missing_dates(plan_builder.build(), default_catalog)
1998
1999            if not no_prompts:
2000                self._prompt_backfill(plan_builder, auto_apply, default_catalog)
2001
2002            backfill_or_preview = "preview" if plan.is_dev and plan.forward_only else "backfill"
2003            if not auto_apply and self._confirm(
2004                f"Apply - {backfill_or_preview.capitalize()} Tables"
2005            ):
2006                plan_builder.apply()
2007        elif plan.has_changes and not auto_apply:
2008            self._prompt_promote(plan_builder)
2009        elif plan.has_unmodified_unpromoted and not auto_apply:
2010            self.log_status_update("\n[bold]Virtually updating unmodified models\n")
2011            self._prompt_promote(plan_builder)
2012
2013    def _prompt_categorize(
2014        self,
2015        plan_builder: PlanBuilder,
2016        auto_apply: bool,
2017        no_diff: bool,
2018        no_prompts: bool,
2019        default_catalog: t.Optional[str],
2020    ) -> None:
2021        """Get the user's change category for the directly modified models."""
2022        plan = plan_builder.build()
2023
2024        if plan.restatements:
2025            # A plan can have restatements for the following reasons:
2026            # - The user specifically called `sqlmesh plan` with --restate-model.
2027            #   This creates a "restatement plan" which disallows all other changes and simply force-backfills
2028            #   the selected models and their downstream dependencies using the versions of the models stored in state.
2029            # - There are no specific restatements (so changes are allowed) AND dev previews need to be computed.
2030            #   The "restatements" feature is currently reused for dev previews.
2031            if plan.selected_models_to_restate:
2032                # There were legitimate restatements, no dev previews
2033                tree = Tree(
2034                    "[bold]Models selected for restatement:[/bold]\n"
2035                    "This causes backfill of the model itself as well as affected downstream models"
2036                )
2037                model_fqn_to_snapshot = {s.name: s for s in plan.snapshots.values()}
2038                for model_fqn in plan.selected_models_to_restate:
2039                    snapshot = model_fqn_to_snapshot[model_fqn]
2040                    display_name = snapshot.display_name(
2041                        plan.environment_naming_info,
2042                        default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
2043                        dialect=self.dialect,
2044                    )
2045                    tree.add(
2046                        display_name
2047                    )  # note: we deliberately dont show any intervals here; they get shown in the backfill section
2048                self._print(tree)
2049            else:
2050                # We are computing dev previews, do not confuse the user by printing out something to do
2051                # with restatements. Dev previews are already highlighted in the backfill step
2052                pass
2053        else:
2054            self.show_environment_difference_summary(
2055                plan.context_diff,
2056                no_diff=no_diff,
2057            )
2058
2059        if plan.context_diff.has_changes:
2060            self.show_model_difference_summary(
2061                plan.context_diff,
2062                plan.environment_naming_info,
2063                default_catalog=default_catalog,
2064            )
2065
2066        if not no_diff:
2067            self._show_categorized_snapshots(plan, default_catalog)
2068
2069        for snapshot in plan.uncategorized:
2070            if snapshot.is_model and snapshot.model.forward_only:
2071                continue
2072            if not no_diff:
2073                self.show_sql(plan.context_diff.text_diff(snapshot.name))
2074            tree = Tree(
2075                f"[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}"
2076            )
2077            indirect_tree = None
2078
2079            for child_sid in sorted(plan.indirectly_modified.get(snapshot.snapshot_id, set())):
2080                child_snapshot = plan.context_diff.snapshots[child_sid]
2081                if not indirect_tree:
2082                    indirect_tree = Tree("[indirect]Indirectly Modified Children:")
2083                    tree.add(indirect_tree)
2084                indirect_tree.add(
2085                    f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}"
2086                )
2087            if indirect_tree:
2088                indirect_tree = self._limit_model_names(indirect_tree, self.verbosity)
2089
2090            self._print(tree)
2091            if not no_prompts:
2092                self._get_snapshot_change_category(
2093                    snapshot, plan_builder, auto_apply, default_catalog
2094                )
2095
2096    def _show_categorized_snapshots(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
2097        context_diff = plan.context_diff
2098
2099        for snapshot in plan.categorized:
2100            if context_diff.directly_modified(snapshot.name):
2101                category_str = SNAPSHOT_CHANGE_CATEGORY_STR[snapshot.change_category]
2102                tree = Tree(
2103                    f"\n[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)} ({category_str})"
2104                )
2105                indirect_tree = None
2106                for child_sid in sorted(plan.indirectly_modified.get(snapshot.snapshot_id, set())):
2107                    child_snapshot = context_diff.snapshots[child_sid]
2108                    if not indirect_tree:
2109                        indirect_tree = Tree("[indirect]Indirectly Modified Children:")
2110                        tree.add(indirect_tree)
2111                    child_category_str = SNAPSHOT_CHANGE_CATEGORY_STR[
2112                        child_snapshot.change_category
2113                    ]
2114                    indirect_tree.add(
2115                        f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)} ({child_category_str})"
2116                    )
2117                if indirect_tree:
2118                    indirect_tree = self._limit_model_names(indirect_tree, self.verbosity)
2119            elif context_diff.metadata_updated(snapshot.name):
2120                tree = Tree(
2121                    f"\n[bold][metadata]Metadata Updated: {snapshot.display_name(plan.environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}"
2122                )
2123            else:
2124                continue
2125
2126            text_diff = context_diff.text_diff(snapshot.name)
2127            if text_diff:
2128                self._print("")
2129                self._print(Syntax(text_diff, "sql", word_wrap=True))
2130            self._print(tree)
2131
2132    def _show_missing_dates(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
2133        """Displays the models with missing dates."""
2134        missing_intervals = plan.missing_intervals
2135        if not missing_intervals:
2136            return
2137        backfill = Tree("[bold]Models needing backfill:[/bold]")
2138        for missing in missing_intervals:
2139            snapshot = plan.context_diff.snapshots[missing.snapshot_id]
2140            if not snapshot.is_model:
2141                continue
2142
2143            preview_modifier = ""
2144            if not plan.deployability_index.is_deployable(snapshot):
2145                preview_modifier = " ([orange1]preview[/orange1])"
2146
2147            display_name = snapshot.display_name(
2148                plan.environment_naming_info,
2149                default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
2150                dialect=self.dialect,
2151            )
2152            backfill.add(
2153                f"{display_name}: \\[{_format_missing_intervals(snapshot, missing)}]{preview_modifier}"
2154            )
2155
2156        if backfill:
2157            backfill = self._limit_model_names(backfill, self.verbosity)
2158        self._print(backfill)
2159
2160    def _prompt_effective_from(
2161        self, plan_builder: PlanBuilder, auto_apply: bool, default_catalog: t.Optional[str]
2162    ) -> None:
2163        if not plan_builder.build().effective_from:
2164            effective_from = self._prompt(
2165                "Enter the effective date (eg. '1 year', '2020-01-01') to apply forward-only changes retroactively or blank to only apply them going forward once changes are deployed to prod"
2166            )
2167            if effective_from:
2168                plan_builder.set_effective_from(effective_from)
2169
2170    def _prompt_backfill(
2171        self, plan_builder: PlanBuilder, auto_apply: bool, default_catalog: t.Optional[str]
2172    ) -> None:
2173        plan = plan_builder.build()
2174        is_forward_only_dev = plan.is_dev and plan.forward_only
2175        backfill_or_preview = "preview" if is_forward_only_dev else "backfill"
2176
2177        if plan_builder.is_start_and_end_allowed:
2178            if not plan_builder.override_start:
2179                if is_forward_only_dev:
2180                    if plan.effective_from:
2181                        blank_meaning = f"to preview starting from the effective date ('{time_like_to_str(plan.effective_from)}')"
2182                        default_start = plan.effective_from
2183                    else:
2184                        blank_meaning = "to preview starting from yesterday"
2185                        default_start = yesterday_ds()
2186                else:
2187                    if plan.provided_start:
2188                        blank_meaning = f"starting from '{time_like_to_str(plan.provided_start)}'"
2189                    else:
2190                        blank_meaning = "from the beginning of history"
2191                    default_start = None
2192
2193                start = self._prompt(
2194                    f"Enter the {backfill_or_preview} start date (eg. '1 year', '2020-01-01') or blank to backfill {blank_meaning}",
2195                )
2196                if start:
2197                    plan_builder.set_start(start)
2198                elif default_start:
2199                    plan_builder.set_start(default_start)
2200
2201            if not plan_builder.override_end:
2202                if plan.provided_end:
2203                    blank_meaning = f"'{time_like_to_str(plan.provided_end)}'"
2204                elif plan.end_override_per_model:
2205                    max_end = max(plan.end_override_per_model.values())
2206                    blank_meaning = f"'{time_like_to_str(max_end)}'"
2207                else:
2208                    blank_meaning = "now"
2209                end = self._prompt(
2210                    f"Enter the {backfill_or_preview} end date (eg. '1 month ago', '2020-01-01') or blank to {backfill_or_preview} up until {blank_meaning}",
2211                )
2212                if end:
2213                    plan_builder.set_end(end)
2214
2215            plan = plan_builder.build()
2216
2217    def _prompt_promote(self, plan_builder: PlanBuilder) -> None:
2218        if self._confirm(
2219            "Apply - Virtual Update",
2220        ):
2221            plan_builder.apply()
2222
2223    def log_test_results(self, result: ModelTextTestResult, target_dialect: str) -> None:
2224        # We don't log the test results if no tests were ran
2225        if not result.testsRun:
2226            return
2227
2228        divider_length = 70
2229
2230        self._log_test_details(result)
2231
2232        message = (
2233            f"Ran {result.testsRun} tests against {target_dialect} in {result.duration} seconds."
2234        )
2235        if result.wasSuccessful():
2236            self._print("=" * divider_length)
2237            self._print(
2238                f"Successfully {message}",
2239                style="green",
2240            )
2241            self._print("-" * divider_length)
2242        else:
2243            self._print("-" * divider_length)
2244            self._print("Test Failure Summary", style="red")
2245            self._print("=" * divider_length)
2246            fail_and_error_tests = result.get_fail_and_error_tests()
2247            self._print(f"{message} \n")
2248
2249            self._print(f"Failed tests ({len(fail_and_error_tests)}):")
2250            for test in fail_and_error_tests:
2251                self._print(f" • {test.path}::{test.test_name}")
2252            self._print("=" * divider_length, end="\n\n")
2253
2254    def _captured_unit_test_results(self, result: ModelTextTestResult) -> str:
2255        with self.console.capture() as capture:
2256            self._log_test_details(result)
2257        return strip_ansi_codes(capture.get())
2258
2259    def show_sql(self, sql: str) -> None:
2260        self._print(Syntax(sql, "sql", word_wrap=True), crop=False)
2261
2262    def log_status_update(self, message: str) -> None:
2263        self._print(message)
2264
2265    def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
2266        if snapshot_names:
2267            msg = "  " + "\n  ".join(snapshot_names)
2268            self._print(f"[dark_orange3]Skipped models[/dark_orange3]\n\n{msg}")
2269
2270    def log_failed_models(self, errors: t.List[NodeExecutionFailedError]) -> None:
2271        if errors:
2272            self._print("\n[red]Failed models[/red]\n")
2273
2274            error_messages = _format_node_errors(errors)
2275
2276            for node_name, msg in error_messages.items():
2277                self._print(f"  [red]{node_name}[/red]\n\n{msg}")
2278
2279    def log_models_updated_during_restatement(
2280        self,
2281        snapshots: t.List[t.Tuple[SnapshotTableInfo, SnapshotTableInfo]],
2282        environment_naming_info: EnvironmentNamingInfo,
2283        default_catalog: t.Optional[str] = None,
2284    ) -> None:
2285        if snapshots:
2286            tree = Tree(
2287                f"[yellow]The following models had new versions deployed while data was being restated:[/yellow]"
2288            )
2289
2290            for restated_snapshot, updated_snapshot in snapshots:
2291                display_name = restated_snapshot.display_name(
2292                    environment_naming_info,
2293                    default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
2294                    dialect=self.dialect,
2295                )
2296                current_branch = tree.add(display_name)
2297                current_branch.add(f"restated version: '{restated_snapshot.version}'")
2298                current_branch.add(f"currently active version: '{updated_snapshot.version}'")
2299
2300            self._print(tree)
2301            self._print("")  # newline spacer
2302
2303    def log_destructive_change(
2304        self,
2305        snapshot_name: str,
2306        alter_operations: t.List[TableAlterOperation],
2307        dialect: str,
2308        error: bool = True,
2309    ) -> None:
2310        if error:
2311            self._print(format_destructive_change_msg(snapshot_name, alter_operations, dialect))
2312        else:
2313            self.log_warning(
2314                format_destructive_change_msg(snapshot_name, alter_operations, dialect, error)
2315            )
2316
2317    def log_additive_change(
2318        self,
2319        snapshot_name: str,
2320        alter_operations: t.List[TableAlterOperation],
2321        dialect: str,
2322        error: bool = True,
2323    ) -> None:
2324        if error:
2325            self._print(format_additive_change_msg(snapshot_name, alter_operations, dialect))
2326        else:
2327            self.log_warning(
2328                format_additive_change_msg(snapshot_name, alter_operations, dialect, error)
2329            )
2330
2331    def log_error(self, message: str) -> None:
2332        self._print(f"[red]{message}[/red]")
2333
2334    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
2335        logger.warning(long_message or short_message)
2336        if not self.ignore_warnings:
2337            if long_message:
2338                file_path = None
2339                for handler in logger.root.handlers:
2340                    if isinstance(handler, logging.FileHandler):
2341                        file_path = handler.baseFilename
2342                        break
2343                file_path_msg = f" Learn more in logs: {file_path}\n" if file_path else ""
2344                short_message = f"{short_message}{file_path_msg}"
2345            message_lstrip = short_message.lstrip()
2346            leading_ws = short_message[: -len(message_lstrip)]
2347            message_formatted = f"{leading_ws}[yellow]\\[WARNING] {message_lstrip}[/yellow]"
2348            self._print(message_formatted)
2349
2350    def log_success(self, message: str) -> None:
2351        self._print(f"[green]{message}[/green]\n")
2352
2353    def loading_start(self, message: t.Optional[str] = None) -> uuid.UUID:
2354        id = uuid.uuid4()
2355        self.loading_status[id] = Status(message or "", console=self.console, spinner="line")
2356        self.loading_status[id].start()
2357        return id
2358
2359    def loading_stop(self, id: uuid.UUID) -> None:
2360        self.loading_status[id].stop()
2361        del self.loading_status[id]
2362
2363    def show_table_diff_details(
2364        self,
2365        models_to_diff: t.List[str],
2366    ) -> None:
2367        """Display information about which tables are going to be diffed"""
2368
2369        if models_to_diff:
2370            m_tree = Tree("\n[b]Models to compare:")
2371            for m in models_to_diff:
2372                m_tree.add(f"[{self.TABLE_DIFF_SOURCE_BLUE}]{m}[/{self.TABLE_DIFF_SOURCE_BLUE}]")
2373            self._print(m_tree)
2374            self._print("")
2375
2376    def start_table_diff_progress(self, models_to_diff: int) -> None:
2377        if not self.table_diff_progress:
2378            self.table_diff_progress = make_progress_bar(
2379                "Calculating model differences", self.console
2380            )
2381            self.table_diff_model_progress = Progress(
2382                TextColumn("{task.fields[view_name]}", justify="right"),
2383                SpinnerColumn(spinner_name="simpleDots"),
2384                console=self.console,
2385            )
2386
2387            progress_table = Table.grid()
2388            progress_table.add_row(self.table_diff_progress)
2389            progress_table.add_row(self.table_diff_model_progress)
2390
2391            self.table_diff_progress_live = Live(progress_table, refresh_per_second=10)
2392            self.table_diff_progress_live.start()
2393
2394            self.table_diff_model_task = self.table_diff_progress.add_task(
2395                "Diffing", total=models_to_diff
2396            )
2397
2398    def start_table_diff_model_progress(self, model: str) -> None:
2399        if self.table_diff_model_progress and model not in self.table_diff_model_tasks:
2400            self.table_diff_model_tasks[model] = self.table_diff_model_progress.add_task(
2401                f"Diffing {model}...",
2402                view_name=model,
2403                total=1,
2404            )
2405
2406    def update_table_diff_progress(self, model: str) -> None:
2407        if self.table_diff_progress:
2408            self.table_diff_progress.update(self.table_diff_model_task, refresh=True, advance=1)
2409        if self.table_diff_model_progress and model in self.table_diff_model_tasks:
2410            model_task_id = self.table_diff_model_tasks[model]
2411            self.table_diff_model_progress.remove_task(model_task_id)
2412
2413    def stop_table_diff_progress(self, success: bool) -> None:
2414        if self.table_diff_progress_live:
2415            self.table_diff_progress_live.stop()
2416            self.table_diff_progress_live = None
2417            self.log_status_update("")
2418
2419            if success:
2420                self.log_success(f"Table diff completed successfully!")
2421            else:
2422                self.log_error("Table diff failed!")
2423
2424        self.table_diff_progress = None
2425        self.table_diff_model_progress = None
2426        self.table_diff_model_tasks = {}
2427
2428    def show_table_diff_summary(self, table_diff: TableDiff) -> None:
2429        tree = Tree("\n[b]Table Diff")
2430
2431        if table_diff.model_name:
2432            model = Tree("Model:")
2433            model.add(f"[blue]{table_diff.model_name}[/blue]")
2434
2435            tree.add(model)
2436
2437            envs = Tree("Environment:")
2438            source = Tree(
2439                f"Source: [{self.TABLE_DIFF_SOURCE_BLUE}]{table_diff.source_alias}[/{self.TABLE_DIFF_SOURCE_BLUE}]"
2440            )
2441            envs.add(source)
2442
2443            target = Tree(
2444                f"Target: [{self.TABLE_DIFF_TARGET_GREEN}]{table_diff.target_alias}[/{self.TABLE_DIFF_TARGET_GREEN}]"
2445            )
2446            envs.add(target)
2447
2448            tree.add(envs)
2449
2450        tables = Tree("Tables:")
2451
2452        tables.add(
2453            f"Source: [{self.TABLE_DIFF_SOURCE_BLUE}]{table_diff.source}[/{self.TABLE_DIFF_SOURCE_BLUE}]"
2454        )
2455        tables.add(
2456            f"Target: [{self.TABLE_DIFF_TARGET_GREEN}]{table_diff.target}[/{self.TABLE_DIFF_TARGET_GREEN}]"
2457        )
2458
2459        tree.add(tables)
2460
2461        join = Tree("Join On:")
2462        _, _, key_column_names = table_diff.key_columns
2463        for col_name in key_column_names:
2464            join.add(f"[yellow]{col_name}[/yellow]")
2465
2466        tree.add(join)
2467
2468        self._print(tree)
2469
2470    def show_schema_diff(self, schema_diff: SchemaDiff) -> None:
2471        source_name = schema_diff.source
2472        if schema_diff.source_alias:
2473            source_name = schema_diff.source_alias.upper()
2474        target_name = schema_diff.target
2475        if schema_diff.target_alias:
2476            target_name = schema_diff.target_alias.upper()
2477
2478        first_line = f"\n[b]Schema Diff Between '[{self.TABLE_DIFF_SOURCE_BLUE}]{source_name}[/{self.TABLE_DIFF_SOURCE_BLUE}]' and '[{self.TABLE_DIFF_TARGET_GREEN}]{target_name}[/{self.TABLE_DIFF_TARGET_GREEN}]'"
2479        if schema_diff.model_name:
2480            first_line = (
2481                first_line + f" environments for model '[blue]{schema_diff.model_name}[/blue]'"
2482            )
2483
2484        tree = Tree(first_line + ":")
2485
2486        if any([schema_diff.added, schema_diff.removed, schema_diff.modified]):
2487            if schema_diff.added:
2488                added = Tree("[green]Added Columns:")
2489                for c, t in schema_diff.added:
2490                    added.add(f"[green]{c} ({t})")
2491                tree.add(added)
2492
2493            if schema_diff.removed:
2494                removed = Tree("[red]Removed Columns:")
2495                for c, t in schema_diff.removed:
2496                    removed.add(f"[red]{c} ({t})")
2497                tree.add(removed)
2498
2499            if schema_diff.modified:
2500                modified = Tree("[magenta]Modified Columns:")
2501                for c, (ft, tt) in schema_diff.modified.items():
2502                    modified.add(f"[magenta]{c} ({ft} -> {tt})")
2503                tree.add(modified)
2504        else:
2505            tree.add("[b]Schemas match")
2506
2507        self.console.print(tree)
2508
2509    def show_row_diff(
2510        self, row_diff: RowDiff, show_sample: bool = True, skip_grain_check: bool = False
2511    ) -> None:
2512        if row_diff.empty:
2513            self.console.print(
2514                "\n[b][red]Neither the source nor the target table contained any records[/red][/b]"
2515            )
2516            return
2517
2518        source_name = row_diff.source
2519        if row_diff.source_alias:
2520            source_name = row_diff.source_alias.upper()
2521        target_name = row_diff.target
2522        if row_diff.target_alias:
2523            target_name = row_diff.target_alias.upper()
2524
2525        if row_diff.stats["null_grain_count"] > 0 or (
2526            not skip_grain_check
2527            and (
2528                row_diff.stats["distinct_count_s"] != row_diff.stats["s_count"]
2529                or row_diff.stats["distinct_count_t"] != row_diff.stats["t_count"]
2530            )
2531        ):
2532            self.console.print(
2533                "[b][red]\nGrain should have unique and not-null audits for accurate results.[/red][/b]"
2534            )
2535
2536        tree = Tree("[b]Row Counts:[/b]")
2537        if row_diff.full_match_count:
2538            tree.add(
2539                f" [b][cyan]FULL MATCH[/cyan]:[/b] {row_diff.full_match_count} rows ({row_diff.full_match_pct}%)"
2540            )
2541        if row_diff.partial_match_count:
2542            tree.add(
2543                f" [b][blue]PARTIAL MATCH[/blue]:[/b] {row_diff.partial_match_count} rows ({row_diff.partial_match_pct}%)"
2544            )
2545        if row_diff.s_only_count:
2546            tree.add(
2547                f" [b][yellow]{source_name} ONLY[/yellow]:[/b] {row_diff.s_only_count} rows ({row_diff.s_only_pct}%)"
2548            )
2549        if row_diff.t_only_count:
2550            tree.add(
2551                f" [b][green]{target_name} ONLY[/green]:[/b] {row_diff.t_only_count} rows ({row_diff.t_only_pct}%)"
2552            )
2553        self.console.print("\n", tree)
2554
2555        self.console.print("\n[b][blue]COMMON ROWS[/blue] column comparison stats:[/b]")
2556        if row_diff.column_stats.shape[0] > 0:
2557            self.console.print(row_diff.column_stats.to_string(index=True), end="\n\n")
2558        else:
2559            self.console.print("  No columns with same name and data type in both tables")
2560
2561        if show_sample:
2562            sample = row_diff.joined_sample
2563            self.console.print("\n[b][blue]COMMON ROWS[/blue] sample data differences:[/b]")
2564            if sample.shape[0] > 0:
2565                keys: list[str] = []
2566                columns: dict[str, list[str]] = {}
2567                source_prefix, source_name = (
2568                    (f"{source_name}__", source_name)
2569                    if source_name.lower() != row_diff.source.lower()
2570                    else ("s__", "SOURCE")
2571                )
2572                target_prefix, target_name = (
2573                    (f"{target_name}__", target_name)
2574                    if target_name.lower() != row_diff.target.lower()
2575                    else ("t__", "TARGET")
2576                )
2577
2578                # Extract key and column names from the joined sample
2579                for column in row_diff.joined_sample.columns:
2580                    if source_prefix in column:
2581                        column_name = "__".join(column.split(source_prefix)[1:])
2582                        columns[column_name] = [column, target_prefix + column_name]
2583                    elif target_prefix not in column:
2584                        keys.append(column)
2585
2586                column_styles = {
2587                    source_name: self.TABLE_DIFF_SOURCE_BLUE,
2588                    target_name: self.TABLE_DIFF_TARGET_GREEN,
2589                }
2590
2591                for column, [source_column, target_column] in columns.items():
2592                    # Create a table with the joined keys and comparison columns
2593                    column_table = row_diff.joined_sample[keys + [source_column, target_column]]
2594
2595                    # Filter to retain non identical-valued rows
2596                    column_table = column_table[
2597                        column_table.apply(
2598                            lambda row: not _cells_match(row[source_column], row[target_column]),
2599                            axis=1,
2600                        )
2601                    ]
2602
2603                    # Rename the column headers for readability
2604                    column_table = column_table.rename(
2605                        columns={
2606                            source_column: source_name,
2607                            target_column: target_name,
2608                        }
2609                    )
2610
2611                    table = Table(show_header=True)
2612                    for column_name in column_table.columns:
2613                        style = column_styles.get(column_name, "")
2614                        table.add_column(column_name, style=style, header_style=style)
2615
2616                    for _, row in column_table.iterrows():
2617                        table.add_row(
2618                            *[
2619                                str(
2620                                    round(cell, row_diff.decimals)
2621                                    if isinstance(cell, float)
2622                                    else cell
2623                                )
2624                                for cell in row
2625                            ]
2626                        )
2627
2628                    self.console.print(
2629                        f"Column: [underline][bold cyan]{column}[/bold cyan][/underline]",
2630                        table,
2631                        end="\n",
2632                    )
2633
2634            else:
2635                self.console.print("  All joined rows match")
2636
2637            if row_diff.s_sample.shape[0] > 0:
2638                self.console.print(f"\n[b][yellow]{source_name} ONLY[/yellow] sample rows:[/b]")
2639                self.console.print(row_diff.s_sample.to_string(index=False), end="\n\n")
2640
2641            if row_diff.t_sample.shape[0] > 0:
2642                self.console.print(f"\n[b][green]{target_name} ONLY[/green] sample rows:[/b]")
2643                self.console.print(row_diff.t_sample.to_string(index=False), end="\n\n")
2644
2645    def show_table_diff(
2646        self,
2647        table_diffs: t.List[TableDiff],
2648        show_sample: bool = True,
2649        skip_grain_check: bool = False,
2650        temp_schema: t.Optional[str] = None,
2651    ) -> None:
2652        """
2653        Display the table diff between all mismatched tables.
2654        """
2655        if len(table_diffs) > 1:
2656            mismatched_tables = []
2657            fully_matched = []
2658            for table_diff in table_diffs:
2659                if (
2660                    table_diff.schema_diff().source_schema == table_diff.schema_diff().target_schema
2661                ) and (
2662                    table_diff.row_diff(
2663                        temp_schema=temp_schema, skip_grain_check=skip_grain_check
2664                    ).full_match_pct
2665                    == 100
2666                ):
2667                    fully_matched.append(table_diff)
2668                else:
2669                    mismatched_tables.append(table_diff)
2670            table_diffs = mismatched_tables if mismatched_tables else []
2671            if fully_matched:
2672                m_tree = Tree("\n[b]Identical Tables")
2673                for m in fully_matched:
2674                    m_tree.add(
2675                        f"[{self.TABLE_DIFF_SOURCE_BLUE}]{m.source}[/{self.TABLE_DIFF_SOURCE_BLUE}] - [{self.TABLE_DIFF_TARGET_GREEN}]{m.target}[/{self.TABLE_DIFF_TARGET_GREEN}]"
2676                    )
2677                self._print(m_tree)
2678
2679            if mismatched_tables:
2680                m_tree = Tree("\n[b]Mismatched Tables")
2681                for m in mismatched_tables:
2682                    m_tree.add(
2683                        f"[{self.TABLE_DIFF_SOURCE_BLUE}]{m.source}[/{self.TABLE_DIFF_SOURCE_BLUE}] - [{self.TABLE_DIFF_TARGET_GREEN}]{m.target}[/{self.TABLE_DIFF_TARGET_GREEN}]"
2684                    )
2685                self._print(m_tree)
2686
2687        for table_diff in table_diffs:
2688            self.show_table_diff_summary(table_diff)
2689            self.show_schema_diff(table_diff.schema_diff())
2690            self.show_row_diff(
2691                table_diff.row_diff(temp_schema=temp_schema, skip_grain_check=skip_grain_check),
2692                show_sample=show_sample,
2693                skip_grain_check=skip_grain_check,
2694            )
2695
2696    def print_environments(self, environments_summary: t.List[EnvironmentSummary]) -> None:
2697        """Prints all environment names along with expiry datetime."""
2698        output = [
2699            f"{summary.name} - {time_like_to_str(summary.expiration_ts)}"
2700            if summary.expiration_ts
2701            else f"{summary.name} - No Expiry"
2702            for summary in environments_summary
2703        ]
2704        output_str = "\n".join([str(len(output)), *output])
2705        self.log_status_update(f"Number of SQLMesh environments are: {output_str}")
2706
2707    def show_intervals(self, snapshot_intervals: t.Dict[Snapshot, SnapshotIntervals]) -> None:
2708        complete = Tree(f"[b]Complete Intervals[/b]")
2709        incomplete = Tree(f"[b]Missing Intervals[/b]")
2710
2711        for snapshot, intervals in sorted(snapshot_intervals.items(), key=lambda s: s[0].node.name):
2712            if intervals.intervals:
2713                incomplete.add(
2714                    f"{snapshot.node.name}: [{intervals.format_intervals(snapshot.node.interval_unit)}]"
2715                )
2716            else:
2717                complete.add(snapshot.node.name)
2718
2719        if complete.children:
2720            self._print(complete)
2721
2722        if incomplete.children:
2723            self._print(incomplete)
2724
2725    def print_connection_config(self, config: ConnectionConfig, title: str = "Connection") -> None:
2726        tree = Tree(f"[b]{title}:[/b]")
2727        tree.add(f"Type: [bold cyan]{config.type_}[/bold cyan]")
2728        tree.add(f"Catalog: [bold cyan]{config.get_catalog()}[/bold cyan]")
2729
2730        try:
2731            engine_adapter_type = config._engine_adapter
2732            tree.add(f"Dialect: [bold cyan]{engine_adapter_type.DIALECT}[/bold cyan]")
2733        except NotImplementedError:
2734            # not all ConnectionConfig's have an engine adapter associated. The CloudConnectionConfig has a HTTP client instead
2735            pass
2736
2737        self._print(tree)
2738
2739    def _get_snapshot_change_category(
2740        self,
2741        snapshot: Snapshot,
2742        plan_builder: PlanBuilder,
2743        auto_apply: bool,
2744        default_catalog: t.Optional[str],
2745    ) -> None:
2746        choices = self._snapshot_change_choices(
2747            snapshot, plan_builder.environment_naming_info, default_catalog
2748        )
2749        response = self._prompt(
2750            "\n".join([f"[{i + 1}] {choice}" for i, choice in enumerate(choices.values())]),
2751            show_choices=False,
2752            choices=[f"{i + 1}" for i in range(len(choices))],
2753        )
2754        choice = list(choices)[int(response) - 1]
2755        plan_builder.set_choice(snapshot, choice)
2756
2757    def _snapshot_change_choices(
2758        self,
2759        snapshot: Snapshot,
2760        environment_naming_info: EnvironmentNamingInfo,
2761        default_catalog: t.Optional[str],
2762        use_rich_formatting: bool = True,
2763    ) -> t.Dict[SnapshotChangeCategory, str]:
2764        direct = snapshot.display_name(
2765            environment_naming_info,
2766            default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
2767            dialect=self.dialect,
2768        )
2769        if use_rich_formatting:
2770            direct = f"[direct]{direct}[/direct]"
2771        indirect = "indirectly modified children"
2772        if use_rich_formatting:
2773            indirect = f"[indirect]{indirect}[/indirect]"
2774        if snapshot.is_view:
2775            choices = {
2776                SnapshotChangeCategory.BREAKING: f"Update {direct} and backfill {indirect}",
2777                SnapshotChangeCategory.NON_BREAKING: f"Update {direct} but don't backfill {indirect}",
2778            }
2779        elif snapshot.is_symbolic:
2780            choices = {
2781                SnapshotChangeCategory.BREAKING: f"Backfill {indirect}",
2782                SnapshotChangeCategory.NON_BREAKING: f"Don't backfill {indirect}",
2783            }
2784        else:
2785            choices = {
2786                SnapshotChangeCategory.BREAKING: f"Backfill {direct} and {indirect}",
2787                SnapshotChangeCategory.NON_BREAKING: f"Backfill {direct} but not {indirect}",
2788            }
2789        labeled_choices = {
2790            k: f"[{SNAPSHOT_CHANGE_CATEGORY_STR[k]}] {v}" for k, v in choices.items()
2791        }
2792        return labeled_choices
2793
2794    def show_linter_violations(
2795        self, violations: t.List[RuleViolation], model: Model, is_error: bool = False
2796    ) -> None:
2797        severity = "errors" if is_error else "warnings"
2798
2799        # Sort violations by line, then alphabetically the name of the violation
2800        # Violations with no range go first
2801        sorted_violations = sorted(
2802            violations,
2803            key=lambda v: (
2804                v.violation_range.start.line if v.violation_range else -1,
2805                v.rule.name.lower(),
2806            ),
2807        )
2808        violations_text = [
2809            (
2810                f" - Line {v.violation_range.start.line + 1}: {v.rule.name} - {v.violation_msg}"
2811                if v.violation_range
2812                else f" - {v.rule.name}: {v.violation_msg}"
2813            )
2814            for v in sorted_violations
2815        ]
2816        violations_msg = "\n".join(violations_text)
2817        msg = f"Linter {severity} for {model._path}:\n{violations_msg}"
2818
2819        if is_error:
2820            self.log_error(msg)
2821        else:
2822            self.log_warning(msg)
2823
2824    def _log_test_details(
2825        self, result: ModelTextTestResult, unittest_char_separator: bool = True
2826    ) -> None:
2827        """
2828        This is a helper method that encapsulates the logic for logging the relevant unittest for the result.
2829        The top level method (`log_test_results`) reuses `_log_test_details` differently based on the console.
2830
2831        Args:
2832            result: The unittest test result that contains metrics like num success, fails, ect.
2833        """
2834        if result.wasSuccessful():
2835            self._print("\n", end="")
2836            return
2837
2838        if unittest_char_separator:
2839            self._print(f"\n{unittest.TextTestResult.separator1}\n\n", end="")
2840
2841        for (test_case, failure), test_failure_tables in zip_longest(  # type: ignore
2842            result.failures, result.failure_tables
2843        ):
2844            self._print(unittest.TextTestResult.separator2)
2845            self._print(f"FAIL: {test_case}")
2846
2847            if test_description := test_case.shortDescription():
2848                self._print(test_description)
2849            self._print(f"{unittest.TextTestResult.separator2}")
2850
2851            if not test_failure_tables:
2852                self._print(failure)
2853            else:
2854                for failure_table in test_failure_tables:
2855                    self._print(failure_table)
2856                    self._print("\n", end="")
2857
2858        for test_case, error in result.errors:
2859            self._print(unittest.TextTestResult.separator2)
2860            self._print(f"ERROR: {test_case}")
2861            self._print(f"{unittest.TextTestResult.separator2}")
2862            self._print(error)
2863
2864
2865def _cells_match(x: t.Any, y: t.Any) -> bool:
2866    """Helper function to compare two cells and returns true if they're equal, handling array objects."""
2867    import pandas as pd
2868    import numpy as np
2869
2870    # Convert array-like objects to list for consistent comparison
2871    def _normalize(val: t.Any) -> t.Any:
2872        # Convert Pandas null to Python null for the purposes of comparison to prevent errors like the following on boolean fields:
2873        # - TypeError: boolean value of NA is ambiguous
2874        # note pd.isnull() returns either a bool or a ndarray[bool] depending on if the input
2875        # is scalar or an array
2876        isnull = pd.isnull(val)
2877
2878        if isinstance(isnull, bool):  # scalar
2879            if isnull:
2880                val = None
2881        elif all(isnull):  # array
2882            val = None
2883
2884        return list(val) if isinstance(val, (pd.Series, np.ndarray)) else val
2885
2886    return _normalize(x) == _normalize(y)
2887
2888
2889def add_to_layout_widget(target_widget: LayoutWidget, *widgets: widgets.Widget) -> LayoutWidget:
2890    """Helper function to add a widget to a layout widget.
2891
2892    Args:
2893        target_widget: The layout widget to add the other widget(s) to.
2894        *widgets: The widgets to add to the layout widget.
2895
2896    Returns:
2897        The layout widget with the children added.
2898    """
2899    target_widget.children += tuple(widgets)
2900    return target_widget
2901
2902
2903class NotebookMagicConsole(TerminalConsole):
2904    """
2905    Console to be used when using the magic notebook interface (`%<command>`).
2906    Generally reuses the Terminal console when possible by either directly outputing what it provides
2907    or capturing it and converting it into a widget.
2908    """
2909
2910    def __init__(
2911        self,
2912        display: t.Optional[t.Callable] = None,
2913        console: t.Optional[RichConsole] = None,
2914        dialect: DialectType = None,
2915        **kwargs: t.Any,
2916    ) -> None:
2917        import ipywidgets as widgets
2918        from IPython import get_ipython
2919        from IPython.display import display as ipython_display
2920
2921        super().__init__(console, **kwargs)
2922
2923        self.display = display or get_ipython().user_ns.get("display", ipython_display)
2924        self.missing_dates_output = widgets.Output()
2925        self.dynamic_options_after_categorization_output = widgets.VBox()
2926
2927        self.dialect = dialect
2928
2929    def _show_missing_dates(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
2930        self._add_to_dynamic_options(self.missing_dates_output)
2931        self.missing_dates_output.outputs = ()
2932        with self.missing_dates_output:
2933            super()._show_missing_dates(plan, default_catalog)
2934
2935    def _apply(self, button: widgets.Button) -> None:
2936        button.disabled = True
2937        with button.output:
2938            button.plan_builder.apply()
2939
2940    def _prompt_promote(self, plan_builder: PlanBuilder) -> None:
2941        import ipywidgets as widgets
2942
2943        button = widgets.Button(
2944            description="Apply - Virtual Update",
2945            disabled=False,
2946            button_style="success",
2947            # Auto will make the button really large.
2948            # Likely changing this soon anyways to be just `Apply` with description above
2949            layout={"width": "10rem"},
2950        )
2951        self._add_to_dynamic_options(button)
2952        output = widgets.Output()
2953        self._add_to_dynamic_options(output)
2954
2955        button.plan_builder = plan_builder
2956        button.on_click(self._apply)
2957        button.output = output
2958
2959    def _prompt_effective_from(
2960        self, plan_builder: PlanBuilder, auto_apply: bool, default_catalog: t.Optional[str]
2961    ) -> None:
2962        import ipywidgets as widgets
2963
2964        prompt = widgets.VBox()
2965
2966        def effective_from_change_callback(change: t.Dict[str, datetime.datetime]) -> None:
2967            plan_builder.set_effective_from(change["new"])
2968            self._show_options_after_categorization(
2969                plan_builder, auto_apply, default_catalog, no_prompts=False
2970            )
2971
2972        def going_forward_change_callback(change: t.Dict[str, bool]) -> None:
2973            checked = change["new"]
2974            plan_builder.set_effective_from(None if checked else yesterday_ds())
2975            self._show_options_after_categorization(
2976                plan_builder,
2977                auto_apply=auto_apply,
2978                default_catalog=default_catalog,
2979                no_prompts=False,
2980            )
2981
2982        date_picker = widgets.DatePicker(
2983            disabled=plan_builder.build().effective_from is None,
2984            value=to_date(plan_builder.build().effective_from or yesterday_ds()),
2985            layout={"width": "auto"},
2986        )
2987        date_picker.observe(effective_from_change_callback, "value")
2988
2989        going_forward_checkbox = widgets.Checkbox(
2990            value=plan_builder.build().effective_from is None,
2991            description="Apply Going Forward Once Deployed To Prod",
2992            disabled=False,
2993            indent=False,
2994        )
2995        going_forward_checkbox.observe(going_forward_change_callback, "value")
2996
2997        add_to_layout_widget(
2998            prompt,
2999            widgets.HBox(
3000                [
3001                    widgets.Label("Effective From Date:", layout={"width": "8rem"}),
3002                    date_picker,
3003                    going_forward_checkbox,
3004                ]
3005            ),
3006        )
3007
3008        self._add_to_dynamic_options(prompt)
3009
3010    def _prompt_backfill(
3011        self, plan_builder: PlanBuilder, auto_apply: bool, default_catalog: t.Optional[str]
3012    ) -> None:
3013        import ipywidgets as widgets
3014
3015        prompt = widgets.VBox()
3016
3017        backfill_or_preview = (
3018            "Preview"
3019            if plan_builder.build().is_dev and plan_builder.build().forward_only
3020            else "Backfill"
3021        )
3022
3023        def _date_picker(
3024            plan_builder: PlanBuilder, value: t.Any, on_change: t.Callable, disabled: bool = False
3025        ) -> widgets.DatePicker:
3026            picker = widgets.DatePicker(
3027                disabled=disabled,
3028                value=value,
3029                layout={"width": "auto"},
3030            )
3031
3032            picker.observe(on_change, "value")
3033            return picker
3034
3035        def start_change_callback(change: t.Dict[str, datetime.datetime]) -> None:
3036            plan_builder.set_start(change["new"])
3037            self._show_options_after_categorization(
3038                plan_builder, auto_apply, default_catalog, no_prompts=False
3039            )
3040
3041        def end_change_callback(change: t.Dict[str, datetime.datetime]) -> None:
3042            plan_builder.set_end(change["new"])
3043            self._show_options_after_categorization(
3044                plan_builder, auto_apply, default_catalog, no_prompts=False
3045            )
3046
3047        if plan_builder.is_start_and_end_allowed:
3048            add_to_layout_widget(
3049                prompt,
3050                widgets.HBox(
3051                    [
3052                        widgets.Label(
3053                            f"Start {backfill_or_preview} Date:", layout={"width": "8rem"}
3054                        ),
3055                        _date_picker(
3056                            plan_builder, to_date(plan_builder.build().start), start_change_callback
3057                        ),
3058                    ]
3059                ),
3060            )
3061
3062            add_to_layout_widget(
3063                prompt,
3064                widgets.HBox(
3065                    [
3066                        widgets.Label(f"End {backfill_or_preview} Date:", layout={"width": "8rem"}),
3067                        _date_picker(
3068                            plan_builder,
3069                            to_date(plan_builder.build().end),
3070                            end_change_callback,
3071                        ),
3072                    ]
3073                ),
3074            )
3075
3076        self._add_to_dynamic_options(prompt)
3077
3078        if not auto_apply:
3079            button = widgets.Button(
3080                description=f"Apply - {backfill_or_preview} Tables",
3081                disabled=False,
3082                button_style="success",
3083            )
3084            self._add_to_dynamic_options(button)
3085            output = widgets.Output()
3086            self._add_to_dynamic_options(output)
3087
3088            button.plan_builder = plan_builder
3089            button.on_click(self._apply)
3090            button.output = output
3091
3092    def _show_options_after_categorization(
3093        self,
3094        plan_builder: PlanBuilder,
3095        auto_apply: bool,
3096        default_catalog: t.Optional[str],
3097        no_prompts: bool,
3098    ) -> None:
3099        self.dynamic_options_after_categorization_output.children = ()
3100        self.display(self.dynamic_options_after_categorization_output)
3101        super()._show_options_after_categorization(
3102            plan_builder, auto_apply, default_catalog, no_prompts
3103        )
3104
3105    def _add_to_dynamic_options(self, *widgets: widgets.Widget) -> None:
3106        add_to_layout_widget(self.dynamic_options_after_categorization_output, *widgets)
3107
3108    def _get_snapshot_change_category(
3109        self,
3110        snapshot: Snapshot,
3111        plan_builder: PlanBuilder,
3112        auto_apply: bool,
3113        default_catalog: t.Optional[str],
3114    ) -> None:
3115        import ipywidgets as widgets
3116
3117        choice_mapping = self._snapshot_change_choices(
3118            snapshot,
3119            plan_builder.environment_naming_info,
3120            default_catalog,
3121            use_rich_formatting=False,
3122        )
3123        choices = list(choice_mapping)
3124        plan_builder.set_choice(snapshot, choices[0])
3125
3126        def radio_button_selected(change: t.Dict[str, t.Any]) -> None:
3127            plan_builder.set_choice(snapshot, choices[change["owner"].index])
3128            self._show_options_after_categorization(
3129                plan_builder, auto_apply, default_catalog, no_prompts=False
3130            )
3131
3132        radio = widgets.RadioButtons(
3133            options=choice_mapping.values(),
3134            layout={"width": "max-content"},
3135            disabled=False,
3136        )
3137        radio.observe(
3138            radio_button_selected,
3139            "value",
3140        )
3141        self.display(radio)
3142
3143    def log_test_results(self, result: ModelTextTestResult, target_dialect: str) -> None:
3144        # We don't log the test results if no tests were ran
3145        if not result.testsRun:
3146            return
3147
3148        import ipywidgets as widgets
3149
3150        divider_length = 70
3151        shared_style = {
3152            "font-size": "11px",
3153            "font-weight": "bold",
3154            "font-family": "Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace",
3155        }
3156
3157        message = (
3158            f"Ran {result.testsRun} tests against {target_dialect} in {result.duration} seconds."
3159        )
3160
3161        if result.wasSuccessful():
3162            success_color = {"color": "#008000"}
3163            header = str(h("span", {"style": shared_style}, "-" * divider_length))
3164            message = str(
3165                h(
3166                    "span",
3167                    {"style": {**shared_style, **success_color}},
3168                    f"Successfully {message}",
3169                )
3170            )
3171            footer = str(h("span", {"style": shared_style}, "=" * divider_length))
3172            self.display(widgets.HTML("<br>".join([header, message, footer])))
3173        else:
3174            output = self._captured_unit_test_results(result)
3175
3176            fail_color = {"color": "#db3737"}
3177            fail_shared_style = {**shared_style, **fail_color}
3178            header = str(h("span", {"style": fail_shared_style}, "-" * divider_length))
3179            message = str(h("span", {"style": fail_shared_style}, "Test Failure Summary"))
3180            fail_and_error_tests = result.get_fail_and_error_tests()
3181            failed_tests = [
3182                str(
3183                    h(
3184                        "span",
3185                        {"style": fail_shared_style},
3186                        f"Failed tests ({len(fail_and_error_tests)}):",
3187                    )
3188                )
3189            ]
3190
3191            for test in fail_and_error_tests:
3192                failed_tests.append(
3193                    str(
3194                        h(
3195                            "span",
3196                            {"style": fail_shared_style},
3197                            f" • {test.model.name}::{test.test_name}",
3198                        )
3199                    )
3200                )
3201            failures = "<br>".join(failed_tests)
3202            footer = str(h("span", {"style": fail_shared_style}, "=" * divider_length))
3203            error_output = widgets.Textarea(output, layout={"height": "300px", "width": "100%"})
3204            test_info = widgets.HTML("<br>".join([header, message, footer, failures, footer]))
3205            self.display(widgets.VBox(children=[test_info, error_output], layout={"width": "100%"}))
3206
3207
3208class CaptureTerminalConsole(TerminalConsole):
3209    """
3210    Captures the output of the terminal console so that it can be extracted out and displayed within other interfaces.
3211    The captured output is cleared out after it is retrieved.
3212
3213    Note: `_prompt` and `_confirm` need to also be overriden to work with the custom interface if you want to use
3214    this console interactively.
3215    """
3216
3217    def __init__(self, console: t.Optional[RichConsole] = None, **kwargs: t.Any) -> None:
3218        super().__init__(console=console, **kwargs)
3219        self._captured_outputs: t.List[str] = []
3220        self._warnings: t.List[str] = []
3221        self._errors: t.List[str] = []
3222
3223    @property
3224    def captured_output(self) -> str:
3225        return "".join(self._captured_outputs)
3226
3227    @property
3228    def captured_warnings(self) -> str:
3229        return "".join(self._warnings)
3230
3231    @property
3232    def captured_errors(self) -> str:
3233        return "".join(self._errors)
3234
3235    def consume_captured_output(self) -> str:
3236        try:
3237            return self.captured_output
3238        finally:
3239            self._captured_outputs = []
3240
3241    def consume_captured_warnings(self) -> str:
3242        try:
3243            return self.captured_warnings
3244        finally:
3245            self._warnings = []
3246
3247    def consume_captured_errors(self) -> str:
3248        try:
3249            return self.captured_errors
3250        finally:
3251            self._errors = []
3252
3253    def log_warning(
3254        self,
3255        short_message: str,
3256        long_message: t.Optional[str] = None,
3257        *args: t.Any,
3258        **kwargs: t.Any,
3259    ) -> None:
3260        if short_message not in self._warnings:
3261            self._warnings.append(short_message)
3262        if kwargs.pop("print", True):
3263            super().log_warning(short_message, long_message)
3264
3265    def log_error(self, message: str, *args: t.Any, **kwargs: t.Any) -> None:
3266        if message not in self._errors:
3267            self._errors.append(message)
3268        if kwargs.pop("print", True):
3269            super().log_error(message)
3270
3271    def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
3272        if snapshot_names:
3273            self._captured_outputs.append(
3274                "\n".join([f"SKIPPED snapshot {skipped}\n" for skipped in snapshot_names])
3275            )
3276            super().log_skipped_models(snapshot_names)
3277
3278    def log_failed_models(self, errors: t.List[NodeExecutionFailedError]) -> None:
3279        self._errors.extend([str(ex) for ex in errors if str(ex) not in self._errors])
3280        super().log_failed_models(errors)
3281
3282    def _print(self, value: t.Any, **kwargs: t.Any) -> None:
3283        with self.console.capture() as capture:
3284            self.console.print(value, **kwargs)
3285        self._captured_outputs.append(capture.get())
3286
3287
3288class MarkdownConsole(CaptureTerminalConsole):
3289    """
3290    A console that outputs markdown. Currently this is only configured for non-interactive use so for use cases
3291    where you want to display a plan or test results in markdown.
3292    """
3293
3294    CHECK_MARK = ""
3295    AUDIT_PASS_MARK = "passed "
3296    GREEN_AUDIT_PASS_MARK = AUDIT_PASS_MARK
3297    AUDIT_FAIL_MARK = "failed "
3298    AUDIT_PADDING = 7
3299
3300    def __init__(self, **kwargs: t.Any) -> None:
3301        self.alert_block_max_content_length = int(kwargs.pop("alert_block_max_content_length", 500))
3302        self.alert_block_collapsible_threshold = int(
3303            kwargs.pop("alert_block_collapsible_threshold", 200)
3304        )
3305
3306        # capture_only = True: capture but dont print to console
3307        # capture_only = False: capture and also print to console
3308        self.warning_capture_only = kwargs.pop("warning_capture_only", False)
3309        self.error_capture_only = kwargs.pop("error_capture_only", False)
3310
3311        super().__init__(
3312            **{**kwargs, "console": RichConsole(no_color=True, width=kwargs.pop("width", None))}
3313        )
3314
3315    def show_environment_difference_summary(
3316        self,
3317        context_diff: ContextDiff,
3318        no_diff: bool = True,
3319    ) -> None:
3320        """Shows a summary of the environment differences.
3321
3322        Args:
3323            context_diff: The context diff to use to print the summary.
3324            no_diff: Hide the actual environment statements differences.
3325        """
3326        if context_diff.is_new_environment:
3327            msg = (
3328                f"\n**`{context_diff.environment}` environment will be initialized**"
3329                if not context_diff.create_from_env_exists
3330                else f"\n**New environment `{context_diff.environment}` will be created from `{context_diff.create_from}`**"
3331            )
3332            self._print(msg)
3333            if not context_diff.has_snapshot_changes:
3334                return
3335
3336        if not context_diff.has_changes:
3337            self._print(
3338                f"\n**No changes to plan: project files match the `{context_diff.environment}` environment**\n"
3339            )
3340            return
3341
3342        self._print(f"\n**Summary of differences from `{context_diff.environment}`:**")
3343
3344        if context_diff.has_requirement_changes:
3345            self._print(f"\nRequirements:\n{context_diff.requirements_diff()}")
3346
3347        if context_diff.has_environment_statements_changes and not no_diff:
3348            self._print("\nEnvironment statements:\n")
3349            for _, diff in context_diff.environment_statements_diff(
3350                include_python_env=not context_diff.is_new_environment
3351            ):
3352                self._print(diff)
3353
3354    def show_model_difference_summary(
3355        self,
3356        context_diff: ContextDiff,
3357        environment_naming_info: EnvironmentNamingInfo,
3358        default_catalog: t.Optional[str],
3359        no_diff: bool = True,
3360    ) -> None:
3361        """Shows a summary of the model differences.
3362
3363        Args:
3364            context_diff: The context diff to use to print the summary.
3365            environment_naming_info: The environment naming info to reference when printing model names
3366            default_catalog: The default catalog to reference when deciding to remove catalog from display names
3367            no_diff: Hide the actual SQL differences.
3368        """
3369        added_snapshots = {context_diff.snapshots[s_id] for s_id in context_diff.added}
3370        if added_snapshots:
3371            self._print("\n**Added Models:**")
3372            self._print_models_with_threshold(
3373                environment_naming_info, {s for s in added_snapshots if s.is_model}, default_catalog
3374            )
3375
3376        added_snapshot_audits = {s for s in added_snapshots if s.is_audit}
3377        if added_snapshot_audits:
3378            self._print("\n**Added Standalone Audits:**")
3379            for snapshot in sorted(added_snapshot_audits):
3380                self._print(
3381                    f"- `{snapshot.display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}`"
3382                )
3383
3384        removed_snapshot_table_infos = set(context_diff.removed_snapshots.values())
3385        if removed_snapshot_table_infos:
3386            self._print("\n**Removed Models:**")
3387            self._print_models_with_threshold(
3388                environment_naming_info,
3389                {s for s in removed_snapshot_table_infos if s.is_model},
3390                default_catalog,
3391            )
3392
3393        removed_audit_snapshot_table_infos = {s for s in removed_snapshot_table_infos if s.is_audit}
3394        if removed_audit_snapshot_table_infos:
3395            self._print("\n**Removed Standalone Audits:**")
3396            for snapshot_table_info in sorted(removed_audit_snapshot_table_infos):
3397                self._print(
3398                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}`"
3399                )
3400
3401        modified_snapshots = {
3402            current_snapshot for current_snapshot, _ in context_diff.modified_snapshots.values()
3403        }
3404        if modified_snapshots:
3405            self._print_modified_models(
3406                context_diff, modified_snapshots, environment_naming_info, default_catalog, no_diff
3407            )
3408
3409    def _print_models_with_threshold(
3410        self,
3411        environment_naming_info: EnvironmentNamingInfo,
3412        snapshot_table_infos: t.Set[SnapshotInfoLike],
3413        default_catalog: t.Optional[str] = None,
3414    ) -> None:
3415        models = sorted(snapshot_table_infos)
3416        list_length = len(models)
3417        if (
3418            self.verbosity < Verbosity.VERY_VERBOSE
3419            and list_length > self.INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD
3420        ):
3421            self._print(
3422                f"- `{models[0].display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}`"
3423            )
3424            self._print(f"- `.... {list_length - 2} more ....`\n")
3425            self._print(
3426                f"- `{models[-1].display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}`"
3427            )
3428        else:
3429            for snapshot_table_info in models:
3430                category_str = SNAPSHOT_CHANGE_CATEGORY_STR[snapshot_table_info.change_category]
3431                self._print(
3432                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}` ({category_str})"
3433                )
3434
3435    def _print_modified_models(
3436        self,
3437        context_diff: ContextDiff,
3438        modified_snapshots: t.Set[Snapshot],
3439        environment_naming_info: EnvironmentNamingInfo,
3440        default_catalog: t.Optional[str] = None,
3441        no_diff: bool = True,
3442    ) -> None:
3443        directly_modified = []
3444        indirectly_modified: t.List[Snapshot] = []
3445        metadata_modified = []
3446        for snapshot in modified_snapshots:
3447            if context_diff.directly_modified(snapshot.name):
3448                directly_modified.append(snapshot)
3449            elif context_diff.indirectly_modified(snapshot.name):
3450                indirectly_modified.append(snapshot)
3451            elif context_diff.metadata_updated(snapshot.name):
3452                metadata_modified.append(snapshot)
3453        if directly_modified:
3454            self._print("\n**Directly Modified:**")
3455            for snapshot in sorted(directly_modified):
3456                category_str = SNAPSHOT_CHANGE_CATEGORY_STR[snapshot.change_category]
3457                self._print(
3458                    f"* `{snapshot.display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}` ({category_str})"
3459                )
3460
3461                indirectly_modified_children = sorted(
3462                    [s for s in indirectly_modified if snapshot.snapshot_id in s.parents]
3463                )
3464
3465                if not no_diff:
3466                    diff_text = context_diff.text_diff(snapshot.name)
3467                    # sometimes there is no text_diff, like on a seed model where the data has been updated
3468                    if diff_text:
3469                        diff_text = f"\n```diff\n{diff_text}\n```"
3470                        # these are part of a Markdown list, so indent them by 2 spaces to relate them to the current list item
3471                        diff_text_indented = "\n".join(
3472                            [f"  {line}" for line in diff_text.splitlines()]
3473                        )
3474                        self._print(diff_text_indented)
3475                    else:
3476                        if indirectly_modified_children:
3477                            self._print("\n")
3478
3479                if indirectly_modified_children:
3480                    self._print("  Indirectly Modified Children:")
3481                    for child_snapshot in indirectly_modified_children:
3482                        child_category_str = SNAPSHOT_CHANGE_CATEGORY_STR[
3483                            child_snapshot.change_category
3484                        ]
3485                        self._print(
3486                            f"    - `{child_snapshot.display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}` ({child_category_str})"
3487                        )
3488                    self._print("\n")
3489
3490        if indirectly_modified:
3491            self._print("\n**Indirectly Modified:**")
3492            self._print_models_with_threshold(
3493                environment_naming_info, set(indirectly_modified), default_catalog
3494            )
3495        if metadata_modified:
3496            self._print("\n**Metadata Updated:**")
3497            for snapshot in sorted(metadata_modified):
3498                self._print(
3499                    f"- `{snapshot.display_name(environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}`"
3500                )
3501
3502    def _show_missing_dates(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
3503        """Displays the models with missing dates."""
3504        missing_intervals = plan.missing_intervals
3505        if not missing_intervals:
3506            return
3507        self._print("\n**Models needing backfill:**")
3508        snapshots = []
3509        for missing in missing_intervals:
3510            snapshot = plan.context_diff.snapshots[missing.snapshot_id]
3511            if not snapshot.is_model:
3512                continue
3513
3514            preview_modifier = ""
3515            if not plan.deployability_index.is_deployable(snapshot):
3516                preview_modifier = " (**preview**)"
3517
3518            display_name = snapshot.display_name(
3519                plan.environment_naming_info,
3520                default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
3521                dialect=self.dialect,
3522            )
3523            snapshots.append(
3524                f"* `{display_name}`: \\[{_format_missing_intervals(snapshot, missing)}]{preview_modifier}"
3525            )
3526
3527        length = len(snapshots)
3528        if (
3529            self.verbosity < Verbosity.VERY_VERBOSE
3530            and length > self.INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD
3531        ):
3532            self._print(snapshots[0])
3533            self._print(f"- `.... {length - 2} more ....`\n")
3534            self._print(snapshots[-1])
3535        else:
3536            for snap in snapshots:
3537                self._print(snap)
3538
3539    def _show_categorized_snapshots(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
3540        context_diff = plan.context_diff
3541        for snapshot in plan.categorized:
3542            if context_diff.directly_modified(snapshot.name):
3543                category_str = SNAPSHOT_CHANGE_CATEGORY_STR[snapshot.change_category]
3544                tree = Tree(
3545                    f"[bold][direct]Directly Modified: {snapshot.display_name(plan.environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)} ({category_str})"
3546                )
3547                indirect_tree = None
3548                for child_sid in sorted(plan.indirectly_modified.get(snapshot.snapshot_id, set())):
3549                    child_snapshot = context_diff.snapshots[child_sid]
3550                    if not indirect_tree:
3551                        indirect_tree = Tree("[indirect]Indirectly Modified Children:")
3552                        tree.add(indirect_tree)
3553                    child_category_str = SNAPSHOT_CHANGE_CATEGORY_STR[
3554                        child_snapshot.change_category
3555                    ]
3556                    indirect_tree.add(
3557                        f"[indirect]{child_snapshot.display_name(plan.environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)} ({child_category_str})"
3558                    )
3559                if indirect_tree:
3560                    indirect_tree = self._limit_model_names(indirect_tree, self.verbosity)
3561            elif context_diff.metadata_updated(snapshot.name):
3562                tree = Tree(
3563                    f"[bold][metadata]Metadata Updated: {snapshot.display_name(plan.environment_naming_info, default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect)}"
3564                )
3565            else:
3566                continue
3567
3568            self._print(f"```diff\n{context_diff.text_diff(snapshot.name)}\n```\n")
3569            self._print("```\n")
3570            self._print(tree)
3571            self._print("\n```")
3572
3573    def stop_evaluation_progress(self, success: bool = True) -> None:
3574        super().stop_evaluation_progress(success)
3575        self._print("\n")
3576
3577    def stop_creation_progress(self, success: bool = True) -> None:
3578        super().stop_creation_progress(success)
3579        self._print("\n")
3580
3581    def stop_promotion_progress(self, success: bool = True) -> None:
3582        super().stop_promotion_progress(success)
3583        self._print("\n")
3584
3585    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
3586        super().log_warning(short_message, long_message, print=not self.warning_capture_only)
3587
3588    def log_error(self, message: str) -> None:
3589        super().log_error(message, print=not self.error_capture_only)
3590
3591    def log_success(self, message: str) -> None:
3592        self._print(message)
3593
3594    def log_test_results(self, result: ModelTextTestResult, target_dialect: str) -> None:
3595        # We don't log the test results if no tests were ran
3596        if not result.testsRun:
3597            return
3598
3599        message = f"Ran `{result.testsRun}` Tests Against `{target_dialect}`"
3600
3601        if result.wasSuccessful():
3602            self._print(f"**Successfully {message}**\n\n")
3603        else:
3604            self._print("```")
3605            self._log_test_details(result, unittest_char_separator=False)
3606            self._print("```\n\n")
3607
3608            fail_and_error_tests = result.get_fail_and_error_tests()
3609            self._print(f"**{message}**\n")
3610            self._print(f"**Failed tests ({len(fail_and_error_tests)}):**")
3611            for test in fail_and_error_tests:
3612                if isinstance(test, ModelTest):
3613                    self._print(f" • `{test.model.name}`::`{test.test_name}`\n\n")
3614
3615    def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
3616        if snapshot_names:
3617            self._print(f"**Skipped models**")
3618            for snapshot_name in snapshot_names:
3619                self._print(f"* `{snapshot_name}`")
3620            self._print("")
3621
3622    def log_failed_models(self, errors: t.List[NodeExecutionFailedError]) -> None:
3623        if errors:
3624            self._print("**Failed models**")
3625
3626            error_messages = _format_node_errors(errors)
3627
3628            for node_name, msg in error_messages.items():
3629                self._print(f"* `{node_name}`\n")
3630                self._print("  ```")
3631                self._print(msg)
3632                self._print("  ```")
3633
3634            self._print("")
3635
3636    def show_linter_violations(
3637        self, violations: t.List[RuleViolation], model: Model, is_error: bool = False
3638    ) -> None:
3639        severity = "**errors**" if is_error else "warnings"
3640        violations_msg = "\n".join(f" - {violation}" for violation in violations)
3641        msg = f"\nLinter {severity} for `{model._path}`:\n{violations_msg}\n"
3642
3643        self._print(msg)
3644        if is_error:
3645            self._errors.append(msg)
3646        else:
3647            self._warnings.append(msg)
3648
3649    @property
3650    def captured_warnings(self) -> str:
3651        return self._render_alert_block("WARNING", self._warnings)
3652
3653    @property
3654    def captured_errors(self) -> str:
3655        return self._render_alert_block("CAUTION", self._errors)
3656
3657    def _render_alert_block(self, block_type: str, items: t.List[str]) -> str:
3658        # GitHub Markdown alert syntax, https://docs.github.com/en/get-started/writing-on-github/getting-started-with-writing-and-formatting-on-github/basic-writing-and-formatting-syntax#alerts
3659        if items:
3660            item_contents = ""
3661            list_indicator = "- " if len(items) > 1 else ""
3662
3663            for item in items:
3664                item = item.replace("\n", "\n> ")
3665                item_contents += f">\n> {list_indicator}{item}\n"
3666
3667                if len(item_contents) > self.alert_block_max_content_length:
3668                    truncation_msg = (
3669                        "...\n>\n> Truncated. Please check the console for full information.\n"
3670                    )
3671                    item_contents = item_contents[
3672                        0 : self.alert_block_max_content_length - len(truncation_msg)
3673                    ]
3674                    item_contents += truncation_msg
3675                    break
3676
3677            if len(item_contents) > self.alert_block_collapsible_threshold:
3678                item_contents = f"> <details>\n{item_contents}> </details>"
3679
3680            return f"> [!{block_type}]\n{item_contents}\n"
3681
3682        return ""
3683
3684    def _print(self, value: t.Any, **kwargs: t.Any) -> None:
3685        self.console.print(value, **kwargs)
3686        with self.console.capture() as capture:
3687            self.console.print(value, **kwargs)
3688        self._captured_outputs.append(capture.get())
3689
3690
3691class DatabricksMagicConsole(CaptureTerminalConsole):
3692    """
3693    Note: Databricks Magic Console currently does not support progress bars while a plan is being applied. The
3694    NotebookMagicConsole does support progress bars, but they will time out after 5 minutes of execution
3695    and it makes it difficult to see the progress of the plan.
3696    """
3697
3698    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
3699        super().__init__(*args, **kwargs)
3700        self.evaluation_batch_progress: t.Dict[SnapshotId, t.Tuple[str, int]] = {}
3701        self.promotion_status: t.Tuple[int, int] = (0, 0)
3702        self.model_creation_status: t.Tuple[int, int] = (0, 0)
3703        self.migration_status: t.Tuple[int, int] = (0, 0)
3704
3705    def _print(self, value: t.Any, **kwargs: t.Any) -> None:
3706        super()._print(value, **kwargs)
3707        for captured_output in self._captured_outputs:
3708            print(captured_output)
3709        self.consume_captured_output()
3710
3711    def _prompt(self, message: str, **kwargs: t.Any) -> t.Any:
3712        self._print(message)
3713        return super()._prompt("", **kwargs)
3714
3715    def _confirm(self, message: str, **kwargs: t.Any) -> bool:
3716        message = f"{message} [y/n]"
3717        self._print(message)
3718        return super()._confirm("", **kwargs)
3719
3720    def start_evaluation_progress(
3721        self,
3722        batched_intervals: t.Dict[Snapshot, Intervals],
3723        environment_naming_info: EnvironmentNamingInfo,
3724        default_catalog: t.Optional[str],
3725        audit_only: bool = False,
3726    ) -> None:
3727        self.evaluation_model_batch_sizes = {
3728            snapshot: len(intervals) for snapshot, intervals in batched_intervals.items()
3729        }
3730        self.evaluation_environment_naming_info = environment_naming_info
3731        self.default_catalog = default_catalog
3732
3733    def start_snapshot_evaluation_progress(
3734        self, snapshot: Snapshot, audit_only: bool = False
3735    ) -> None:
3736        if not self.evaluation_batch_progress.get(snapshot.snapshot_id):
3737            display_name = snapshot.display_name(
3738                self.evaluation_environment_naming_info,
3739                self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None,
3740                dialect=self.dialect,
3741            )
3742            self.evaluation_batch_progress[snapshot.snapshot_id] = (display_name, 0)
3743            print(
3744                f"Starting '{display_name}', Total batches: {self.evaluation_model_batch_sizes[snapshot]}"
3745            )
3746
3747    def update_snapshot_evaluation_progress(
3748        self,
3749        snapshot: Snapshot,
3750        interval: Interval,
3751        batch_idx: int,
3752        duration_ms: t.Optional[int],
3753        num_audits_passed: int,
3754        num_audits_failed: int,
3755        audit_only: bool = False,
3756        execution_stats: t.Optional[QueryExecutionStats] = None,
3757        auto_restatement_triggers: t.Optional[t.List[SnapshotId]] = None,
3758    ) -> None:
3759        view_name, loaded_batches = self.evaluation_batch_progress[snapshot.snapshot_id]
3760
3761        if audit_only:
3762            print(f"Completed Auditing {view_name}")
3763            return
3764
3765        total_batches = self.evaluation_model_batch_sizes[snapshot]
3766
3767        loaded_batches += 1
3768        self.evaluation_batch_progress[snapshot.snapshot_id] = (view_name, loaded_batches)
3769
3770        finished_loading = loaded_batches == total_batches
3771        status = "Loaded" if finished_loading else "Loading"
3772        print(f"{status} '{view_name}', Completed Batches: {loaded_batches}/{total_batches}")
3773        if finished_loading:
3774            total_finished_loading = len(
3775                [
3776                    s
3777                    for s, total in self.evaluation_model_batch_sizes.items()
3778                    if self.evaluation_batch_progress.get(s.snapshot_id, (None, -1))[1] == total
3779                ]
3780            )
3781            total = len(self.evaluation_batch_progress)
3782            print(f"Completed Loading {total_finished_loading}/{total} Models")
3783
3784    def stop_evaluation_progress(self, success: bool = True) -> None:
3785        self.evaluation_batch_progress = {}
3786        super().stop_evaluation_progress(success)
3787        print(f"Loading {'succeeded' if success else 'failed'}")
3788
3789    def start_creation_progress(
3790        self,
3791        snapshots: t.List[Snapshot],
3792        environment_naming_info: EnvironmentNamingInfo,
3793        default_catalog: t.Optional[str],
3794    ) -> None:
3795        """Indicates that a new creation progress has begun."""
3796        self.model_creation_status = (0, len(snapshots))
3797        print("Starting Creating New Model Versions")
3798
3799    def update_creation_progress(self, snapshot: SnapshotInfoLike) -> None:
3800        """Update the snapshot creation progress."""
3801        num_creations, total_creations = self.model_creation_status
3802        num_creations += 1
3803        self.model_creation_status = (num_creations, total_creations)
3804        if num_creations % 5 == 0:
3805            print(f"Created New Model Versions: {num_creations}/{total_creations}")
3806
3807    def stop_creation_progress(self, success: bool = True) -> None:
3808        """Stop the snapshot creation progress."""
3809        self.model_creation_status = (0, 0)
3810        print(f"New Model Creation {'succeeded' if success else 'failed'}")
3811
3812    def start_promotion_progress(
3813        self,
3814        snapshots: t.List[SnapshotTableInfo],
3815        environment_naming_info: EnvironmentNamingInfo,
3816        default_catalog: t.Optional[str],
3817    ) -> None:
3818        """Indicates that a new snapshot promotion progress has begun."""
3819        self.promotion_status = (0, len(snapshots))
3820        print(f"Virtually Updating '{environment_naming_info.name}'")
3821
3822    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
3823        """Update the snapshot promotion progress."""
3824        num_promotions, total_promotions = self.promotion_status
3825        num_promotions += 1
3826        self.promotion_status = (num_promotions, total_promotions)
3827        if num_promotions % 5 == 0:
3828            print(f"Virtually Updated {num_promotions}/{total_promotions}")
3829
3830    def stop_promotion_progress(self, success: bool = True) -> None:
3831        """Stop the snapshot promotion progress."""
3832        self.promotion_status = (0, 0)
3833        print(f"Virtual Update {'succeeded' if success else 'failed'}")
3834
3835    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
3836        """Indicates that a new migration progress has begun."""
3837        self.migration_status = (0, total_tasks)
3838        print("Starting Migration")
3839
3840    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
3841        """Update the migration progress."""
3842        num_migrations, total_migrations = self.migration_status
3843        num_migrations += num_tasks
3844        self.migration_status = (num_migrations, total_migrations)
3845        if num_migrations % 5 == 0:
3846            print(f"Migration Updated {num_migrations}/{total_migrations}")
3847
3848    def log_migration_status(self, success: bool = True) -> None:
3849        """Log the migration status."""
3850        print(f"Migration {'succeeded' if success else 'failed'}")
3851
3852    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
3853        """Stop the migration progress."""
3854        self.migration_status = (0, 0)
3855        print(f"Snapshot migration {'succeeded' if success else 'failed'}")
3856
3857    def start_env_migration_progress(self, total_tasks: int) -> None:
3858        """Indicates that a new migration progress has begun."""
3859        self.env_migration_status = (0, total_tasks)
3860        print("Starting Environment migration")
3861
3862    def update_env_migration_progress(self, num_tasks: int) -> None:
3863        """Update the migration progress."""
3864        num_migrations, total_migrations = self.env_migration_status
3865        num_migrations += num_tasks
3866        self.env_migration_status = (num_migrations, total_migrations)
3867        if num_migrations % 5 == 0:
3868            print(f"Environment migration Updated {num_migrations}/{total_migrations}")
3869
3870    def stop_env_migration_progress(self, success: bool = True) -> None:
3871        """Stop the migration progress."""
3872        self.env_migration_status = (0, 0)
3873        print(f"Environment migration {'succeeded' if success else 'failed'}")
3874
3875
3876class DebuggerTerminalConsole(TerminalConsole):
3877    """A terminal console to use while debugging with no fluff, progress bars, etc."""
3878
3879    def __init__(
3880        self,
3881        console: t.Optional[RichConsole],
3882        *args: t.Any,
3883        dialect: DialectType = None,
3884        ignore_warnings: bool = False,
3885        **kwargs: t.Any,
3886    ) -> None:
3887        self.console: RichConsole = console or srich.console
3888        self.dialect = dialect
3889        self.verbosity = Verbosity.DEFAULT
3890        self.ignore_warnings = ignore_warnings
3891
3892    def _write(self, msg: t.Any, *args: t.Any, **kwargs: t.Any) -> None:
3893        self.console.log(msg, *args, **kwargs)
3894
3895    def start_plan_evaluation(self, plan: EvaluatablePlan) -> None:
3896        self._write("Starting plan", plan.plan_id)
3897
3898    def stop_plan_evaluation(self) -> None:
3899        self._write("Stopping plan")
3900
3901    def start_evaluation_progress(
3902        self,
3903        batched_intervals: t.Dict[Snapshot, Intervals],
3904        environment_naming_info: EnvironmentNamingInfo,
3905        default_catalog: t.Optional[str],
3906        audit_only: bool = False,
3907    ) -> None:
3908        message = "evaluation" if not audit_only else "auditing"
3909        self._write(
3910            f"Starting {message} for {sum(len(intervals) for intervals in batched_intervals.values())} snapshots"
3911        )
3912
3913    def start_snapshot_evaluation_progress(
3914        self, snapshot: Snapshot, audit_only: bool = False
3915    ) -> None:
3916        self._write(f"{'Evaluating' if not audit_only else 'Auditing'} {snapshot.name}")
3917
3918    def update_snapshot_evaluation_progress(
3919        self,
3920        snapshot: Snapshot,
3921        interval: Interval,
3922        batch_idx: int,
3923        duration_ms: t.Optional[int],
3924        num_audits_passed: int,
3925        num_audits_failed: int,
3926        audit_only: bool = False,
3927        execution_stats: t.Optional[QueryExecutionStats] = None,
3928        auto_restatement_triggers: t.Optional[t.List[SnapshotId]] = None,
3929    ) -> None:
3930        message = f"Evaluated {snapshot.name} | batch={batch_idx} | duration={duration_ms}ms | num_audits_passed={num_audits_passed} | num_audits_failed={num_audits_failed}"
3931
3932        if auto_restatement_triggers:
3933            message += f" | auto_restatement_triggers=[{', '.join(trigger.name for trigger in auto_restatement_triggers)}]"
3934
3935        if audit_only:
3936            message = f"Audited {snapshot.name} | duration={duration_ms}ms | num_audits_passed={num_audits_passed} | num_audits_failed={num_audits_failed}"
3937
3938        self._write(message)
3939
3940    def stop_evaluation_progress(self, success: bool = True) -> None:
3941        self._write(f"Stopping evaluation with success={success}")
3942
3943    def start_creation_progress(
3944        self,
3945        snapshots: t.List[Snapshot],
3946        environment_naming_info: EnvironmentNamingInfo,
3947        default_catalog: t.Optional[str],
3948    ) -> None:
3949        self._write(f"Starting creation for {len(snapshots)} snapshots")
3950
3951    def update_creation_progress(self, snapshot: SnapshotInfoLike) -> None:
3952        self._write(f"Creating {snapshot.name}")
3953
3954    def stop_creation_progress(self, success: bool = True) -> None:
3955        self._write(f"Stopping creation with success={success}")
3956
3957    def update_cleanup_progress(self, object_name: str) -> None:
3958        self._write(f"Cleaning up {object_name}")
3959
3960    def start_promotion_progress(
3961        self,
3962        snapshots: t.List[SnapshotTableInfo],
3963        environment_naming_info: EnvironmentNamingInfo,
3964        default_catalog: t.Optional[str],
3965    ) -> None:
3966        if snapshots:
3967            self._write(f"Starting promotion for {len(snapshots)} snapshots")
3968
3969    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
3970        self._write(f"Promoting {snapshot.name}")
3971
3972    def stop_promotion_progress(self, success: bool = True) -> None:
3973        self._write(f"Stopping promotion with success={success}")
3974
3975    def start_snapshot_migration_progress(self, total_tasks: int) -> None:
3976        self._write(f"Starting migration for {total_tasks} snapshots")
3977
3978    def update_snapshot_migration_progress(self, num_tasks: int) -> None:
3979        self._write(f"Migration {num_tasks}")
3980
3981    def log_migration_status(self, success: bool = True) -> None:
3982        self._write(f"Migration finished with success={success}")
3983
3984    def stop_snapshot_migration_progress(self, success: bool = True) -> None:
3985        self._write(f"Stopping snapshot migration with success={success}")
3986
3987    def start_env_migration_progress(self, total_tasks: int) -> None:
3988        self._write(f"Starting migration for {total_tasks} environments")
3989
3990    def update_env_migration_progress(self, num_tasks: int) -> None:
3991        self._write(f"Environment migration {num_tasks}")
3992
3993    def stop_env_migration_progress(self, success: bool = True) -> None:
3994        self._write(f"Stopping environment migration with success={success}")
3995
3996    def show_environment_difference_summary(
3997        self,
3998        context_diff: ContextDiff,
3999        no_diff: bool = True,
4000    ) -> None:
4001        self._write("Environment Difference Summary:")
4002
4003        if context_diff.has_requirement_changes:
4004            self._write(f"Requirements:\n{context_diff.requirements_diff()}")
4005
4006        if context_diff.has_environment_statements_changes and not no_diff:
4007            self._write("Environment statements:\n")
4008            for _, diff in context_diff.environment_statements_diff(
4009                include_python_env=not context_diff.is_new_environment
4010            ):
4011                self._write(diff)
4012
4013    def show_model_difference_summary(
4014        self,
4015        context_diff: ContextDiff,
4016        environment_naming_info: EnvironmentNamingInfo,
4017        default_catalog: t.Optional[str],
4018        no_diff: bool = True,
4019    ) -> None:
4020        self._write("Model Difference Summary:")
4021
4022        for added in context_diff.new_snapshots:
4023            self._write(f"  Added: {added}")
4024        for removed in context_diff.removed_snapshots:
4025            self._write(f"  Removed: {removed}")
4026        for modified in context_diff.modified_snapshots:
4027            self._write(f"  Modified: {modified}")
4028
4029    def log_test_results(self, result: ModelTextTestResult, target_dialect: str) -> None:
4030        self._write("Test Results:", result)
4031
4032    def show_sql(self, sql: str) -> None:
4033        self._write(sql)
4034
4035    def log_status_update(self, message: str) -> None:
4036        self._write(message, style="bold blue")
4037
4038    def log_error(self, message: str) -> None:
4039        self._write(message, style="bold red")
4040
4041    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
4042        logger.warning(long_message or short_message)
4043        if not self.ignore_warnings:
4044            self._write(short_message, style="bold yellow")
4045
4046    def log_success(self, message: str) -> None:
4047        self._write(message, style="bold green")
4048
4049    def loading_start(self, message: t.Optional[str] = None) -> uuid.UUID:
4050        self._write(message)
4051        return uuid.uuid4()
4052
4053    def loading_stop(self, id: uuid.UUID) -> None:
4054        self._write("Done")
4055
4056    def show_schema_diff(self, schema_diff: SchemaDiff) -> None:
4057        self._write(schema_diff)
4058
4059    def show_row_diff(
4060        self, row_diff: RowDiff, show_sample: bool = True, skip_grain_check: bool = False
4061    ) -> None:
4062        self._write(row_diff)
4063
4064    def show_table_diff(
4065        self,
4066        table_diffs: t.List[TableDiff],
4067        show_sample: bool = True,
4068        skip_grain_check: bool = False,
4069        temp_schema: t.Optional[str] = None,
4070    ) -> None:
4071        for table_diff in table_diffs:
4072            self.show_table_diff_summary(table_diff)
4073            self.show_schema_diff(table_diff.schema_diff())
4074            self.show_row_diff(
4075                table_diff.row_diff(temp_schema=temp_schema, skip_grain_check=skip_grain_check),
4076                show_sample=show_sample,
4077                skip_grain_check=skip_grain_check,
4078            )
4079
4080    def update_table_diff_progress(self, model: str) -> None:
4081        self._write(f"Finished table diff for: {model}")
4082
4083    def start_table_diff_progress(self, models_to_diff: int) -> None:
4084        self._write("Table diff started")
4085
4086    def start_table_diff_model_progress(self, model: str) -> None:
4087        self._write(f"Calculating differences for: {model}")
4088
4089    def stop_table_diff_progress(self, success: bool) -> None:
4090        self._write(f"Table diff finished with success={success}")
4091
4092    def show_table_diff_details(
4093        self,
4094        models_to_diff: t.List[str],
4095    ) -> None:
4096        if models_to_diff:
4097            models = "\n".join(models_to_diff)
4098            self._write(f"Models to compare: {models}")
4099
4100    def show_table_diff_summary(self, table_diff: TableDiff) -> None:
4101        if table_diff.model_name:
4102            self._write(f"Model: {table_diff.model_name}")
4103            self._write(f"Source env: {table_diff.source_alias}")
4104            self._write(f"Target env: {table_diff.target_alias}")
4105        self._write(f"Source table: {table_diff.source}")
4106        self._write(f"Target table: {table_diff.target}")
4107        _, _, key_column_names = table_diff.key_columns
4108        keys = ", ".join(key_column_names)
4109        self._write(f"Join On: {keys}")
4110
4111
4112_CONSOLE: Console = NoopConsole()
4113
4114
4115def set_console(console: Console) -> None:
4116    """Sets the console instance."""
4117    global _CONSOLE
4118    _CONSOLE = console
4119
4120
4121def configure_console(**kwargs: t.Any) -> None:
4122    """Configures the console instance."""
4123    global _CONSOLE
4124    _CONSOLE = create_console(**kwargs)
4125
4126
4127def get_console() -> Console:
4128    """Returns the console instance or creates a new one if it hasn't been created yet."""
4129    return _CONSOLE
4130
4131
4132def create_console(
4133    **kwargs: t.Any,
4134) -> TerminalConsole | DatabricksMagicConsole | NotebookMagicConsole:
4135    """
4136    Creates a new console instance that is appropriate for the current runtime environment.
4137
4138    Note: Google Colab environment is untested and currently assumes is compatible with the base
4139    NotebookMagicConsole.
4140    """
4141    from sqlmesh import RuntimeEnv
4142
4143    runtime_env = RuntimeEnv.get()
4144
4145    runtime_env_mapping = {
4146        RuntimeEnv.DATABRICKS: DatabricksMagicConsole,
4147        RuntimeEnv.JUPYTER: NotebookMagicConsole,
4148        RuntimeEnv.TERMINAL: TerminalConsole,
4149        RuntimeEnv.GOOGLE_COLAB: NotebookMagicConsole,
4150        RuntimeEnv.DEBUGGER: DebuggerTerminalConsole,
4151        RuntimeEnv.CI: MarkdownConsole,
4152    }
4153    rich_console_kwargs: t.Dict[str, t.Any] = {"theme": srich.theme}
4154    if runtime_env.is_jupyter or runtime_env.is_google_colab:
4155        rich_console_kwargs["force_jupyter"] = True
4156    return runtime_env_mapping[runtime_env](
4157        **{**{"console": RichConsole(**rich_console_kwargs)}, **kwargs}
4158    )
4159
4160
4161def _format_missing_intervals(snapshot: Snapshot, missing: SnapshotIntervals) -> str:
4162    return (
4163        missing.format_intervals(snapshot.node.interval_unit)
4164        if snapshot.is_incremental
4165        else "recreate view"
4166        if snapshot.is_view
4167        else "full refresh"
4168    )
4169
4170
4171def _format_node_errors(errors: t.List[NodeExecutionFailedError]) -> t.Dict[str, str]:
4172    """Formats a list of node execution errors for display."""
4173
4174    def _format_node_error(ex: NodeExecutionFailedError) -> str:
4175        cause = ex.__cause__ if ex.__cause__ else ex
4176
4177        error_msg = str(cause)
4178
4179        if isinstance(cause, NodeAuditsErrors):
4180            error_msg = _format_audits_errors(cause)
4181        elif not isinstance(cause, (NodeExecutionFailedError, PythonModelEvalError)):
4182            error_msg = "  " + error_msg.replace("\n", "\n  ")
4183            error_msg = (
4184                f"  {cause.__class__.__name__}:\n{error_msg}"  # include error class name in msg
4185            )
4186        error_msg = error_msg.replace("\n", "\n  ")
4187        error_msg = error_msg + "\n" if not error_msg.rstrip(" ").endswith("\n") else error_msg
4188
4189        return error_msg
4190
4191    error_messages = {}
4192
4193    num_fails = len(errors)
4194    for i, error in enumerate(errors):
4195        node_name = ""
4196        if isinstance(error.node, SnapshotId):
4197            node_name = error.node.name
4198        elif hasattr(error.node, "snapshot_name"):
4199            node_name = error.node.snapshot_name
4200
4201        msg = _format_node_error(error)
4202        msg = "  " + msg.replace("\n", "\n  ")
4203        if i == (num_fails - 1):
4204            msg = msg if msg.rstrip(" ").endswith("\n") else msg + "\n"
4205
4206        error_messages[node_name] = msg
4207
4208    return error_messages
4209
4210
4211def _format_audits_errors(error: NodeAuditsErrors) -> str:
4212    error_messages = []
4213    for err in error.errors:
4214        audit_args_sql = []
4215        for arg_name, arg_value in err.audit_args.items():
4216            audit_args_sql.append(f"{arg_name} := {arg_value.sql(dialect=err.adapter_dialect)}")
4217        audit_args_sql_msg = ("\n".join(audit_args_sql) + "\n\n") if audit_args_sql else ""
4218
4219        err_msg = f"'{err.audit_name}' audit error: {err.count} {'row' if err.count == 1 else 'rows'} failed"
4220
4221        query = "\n  ".join(textwrap.wrap(err.sql(err.adapter_dialect), width=LINE_WRAP_WIDTH))
4222        msg = f"{err_msg}\n\nAudit arguments\n  {audit_args_sql_msg}Audit query\n  {query}\n\n"
4223        msg = msg.replace("\n", "\n  ")
4224        error_messages.append(msg)
4225    return "  " + "\n".join(error_messages)
4226
4227
4228def _format_interval(snapshot: Snapshot, interval: Interval) -> str:
4229    """Format an interval with an optional prefix."""
4230    inclusive_interval = make_inclusive(interval[0], interval[1])
4231    if snapshot.model.interval_unit.is_date_granularity:
4232        return f"{to_ds(inclusive_interval[0])} - {to_ds(inclusive_interval[1])}"
4233
4234    if inclusive_interval[0].date() == inclusive_interval[1].date():
4235        # omit end date if interval start/end on same day
4236        return f"{to_ds(inclusive_interval[0])} {inclusive_interval[0].strftime('%H:%M:%S')}-{inclusive_interval[1].strftime('%H:%M:%S')}"
4237
4238    return f"{inclusive_interval[0].strftime('%Y-%m-%d %H:%M:%S')} - {inclusive_interval[1].strftime('%Y-%m-%d %H:%M:%S')}"
4239
4240
4241def _format_signal_interval(snapshot: Snapshot, interval: Interval) -> str:
4242    """Format an interval for signal output (without 'insert' prefix)."""
4243    return _format_interval(snapshot, interval)
4244
4245
4246def _format_evaluation_model_interval(snapshot: Snapshot, interval: Interval) -> str:
4247    """Format an interval for evaluation output (with 'insert' prefix)."""
4248    if snapshot.is_model and (
4249        snapshot.model.kind.is_incremental
4250        or snapshot.model.kind.is_managed
4251        or snapshot.model.kind.is_custom
4252    ):
4253        formatted_interval = _format_interval(snapshot, interval)
4254        return f"insert {formatted_interval}"
4255
4256    return ""
4257
4258
4259def _create_evaluation_model_annotation(
4260    snapshot: Snapshot,
4261    interval_info: t.Optional[str],
4262    execution_stats: t.Optional[QueryExecutionStats],
4263) -> str:
4264    annotation = None
4265    execution_stats_str = ""
4266    if execution_stats:
4267        rows_processed = execution_stats.total_rows_processed
4268        if rows_processed:
4269            # 1.00 and 1.0 to 1
4270            rows_processed_str = metric(rows_processed).replace(".00", "").replace(".0", "")
4271            execution_stats_str += f"{rows_processed_str} row{'s' if rows_processed > 1 else ''}"
4272
4273        bytes_processed = execution_stats.total_bytes_processed
4274        execution_stats_str += (
4275            f"{', ' if execution_stats_str else ''}{naturalsize(bytes_processed, binary=True)}"
4276            if bytes_processed
4277            else ""
4278        )
4279    execution_stats_str = f" ({execution_stats_str})" if execution_stats_str else ""
4280
4281    if snapshot.is_audit:
4282        annotation = "run standalone audit"
4283    if snapshot.is_model:
4284        if snapshot.model.kind.is_external:
4285            annotation = "run external audits"
4286        if snapshot.model.kind.is_view:
4287            annotation = "recreate view"
4288        if snapshot.model.kind.is_seed:
4289            annotation = f"insert seed file{execution_stats_str}"
4290        if snapshot.model.kind.is_full:
4291            annotation = f"full refresh{execution_stats_str}"
4292        if snapshot.model.kind.is_incremental_by_unique_key:
4293            annotation = f"insert/update rows{execution_stats_str}"
4294        if snapshot.model.kind.is_incremental_by_partition:
4295            annotation = f"insert partitions{execution_stats_str}"
4296
4297    if annotation:
4298        return annotation
4299
4300    return f"{interval_info}{execution_stats_str}" if interval_info else ""
4301
4302
4303def _calculate_interval_str_len(
4304    snapshot: Snapshot,
4305    intervals: t.List[Interval],
4306    execution_stats: t.Optional[QueryExecutionStats] = None,
4307) -> int:
4308    interval_str_len = 0
4309    for interval in intervals:
4310        interval_str_len = max(
4311            interval_str_len,
4312            len(
4313                _create_evaluation_model_annotation(
4314                    snapshot, _format_evaluation_model_interval(snapshot, interval), execution_stats
4315                )
4316            ),
4317        )
4318    return interval_str_len
4319
4320
4321def _calculate_audit_str_len(snapshot: Snapshot, audit_padding: int = 0) -> int:
4322    # The annotation includes audit results. We cannot build the audits result string
4323    # until after evaluation occurs, but we must determine the annotation column width here.
4324    # Therefore, we add enough padding for the longest possible audits result string.
4325    audit_str_len = 0
4326    audit_base_str_len = len(f", audits ") + 1  # +1 for check/X
4327    if snapshot.is_audit:
4328        # +1 for "1" audit count, +1 for red X
4329        audit_str_len = max(
4330            audit_str_len, audit_base_str_len + (2 if not snapshot.audit.blocking else 1)
4331        )
4332    if snapshot.is_model and snapshot.model.audits:
4333        num_audits = len(snapshot.model.audits_with_args)
4334        num_nonblocking_audits = sum(
4335            1
4336            for audit in snapshot.model.audits_with_args
4337            if not audit[0].blocking
4338            or ("blocking" in audit[1] and audit[1]["blocking"] == exp.false())
4339        )
4340        if num_audits == 1:
4341            # +1 for "1" audit count, +1 for red X
4342            # if audit_padding is > 0 we're using "failed" instead of red X
4343            audit_len = (
4344                audit_base_str_len
4345                + (2 if num_nonblocking_audits else 1)
4346                + (
4347                    audit_padding - 1
4348                    if num_nonblocking_audits and audit_padding > 0
4349                    else audit_padding
4350                )
4351            )
4352        else:
4353            audit_len = audit_base_str_len + len(str(num_audits)) + audit_padding
4354            if num_nonblocking_audits:
4355                # +1 for space, +1 for red X
4356                # if audit_padding is > 0 we're using "failed" instead of red X
4357                audit_len += (
4358                    len(str(num_nonblocking_audits))
4359                    + 2
4360                    + (audit_padding - 1 if audit_padding > 0 else audit_padding)
4361                )
4362        audit_str_len = max(audit_str_len, audit_len)
4363    return audit_str_len
4364
4365
4366def _calculate_annotation_str_len(
4367    batched_intervals: t.Dict[Snapshot, t.List[Interval]],
4368    audit_padding: int = 0,
4369    execution_stats_len: int = 0,
4370) -> int:
4371    annotation_str_len = 0
4372    for snapshot, intervals in batched_intervals.items():
4373        annotation_str_len = max(
4374            annotation_str_len,
4375            _calculate_interval_str_len(snapshot, intervals)
4376            + _calculate_audit_str_len(snapshot, audit_padding)
4377            + execution_stats_len,
4378        )
4379    return annotation_str_len