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