sqlmesh.integrations.github.cicd.controller

   1from __future__ import annotations
   2
   3import functools
   4import json
   5import logging
   6import os
   7import pathlib
   8import re
   9import traceback
  10import typing as t
  11from enum import Enum
  12from pathlib import Path
  13from dataclasses import dataclass
  14from functools import cached_property
  15
  16import requests
  17from sqlglot.helper import seq_get
  18
  19from sqlmesh.core import constants as c
  20from sqlmesh.core.console import SNAPSHOT_CHANGE_CATEGORY_STR, get_console, MarkdownConsole
  21from sqlmesh.core.context import Context
  22from sqlmesh.core.test.result import ModelTextTestResult
  23from sqlmesh.core.environment import Environment
  24from sqlmesh.core.plan import Plan, PlanBuilder, SnapshotIntervals
  25from sqlmesh.core.plan.definition import UserProvidedFlags
  26from sqlmesh.core.snapshot.definition import (
  27    Snapshot,
  28    SnapshotChangeCategory,
  29    SnapshotId,
  30    SnapshotTableInfo,
  31)
  32from sqlglot.errors import SqlglotError
  33from sqlmesh.core.user import User
  34from sqlmesh.core.config import Config
  35from sqlmesh.integrations.github.cicd.config import GithubCICDBotConfig
  36from sqlmesh.utils import word_characters_only, Verbosity
  37from sqlmesh.utils.date import now
  38from sqlmesh.utils.errors import (
  39    CICDBotError,
  40    NoChangesPlanError,
  41    PlanError,
  42    UncategorizedPlanError,
  43    LinterError,
  44    SQLMeshError,
  45)
  46from sqlmesh.utils.pydantic import PydanticModel
  47
  48if t.TYPE_CHECKING:
  49    from github import Github
  50    from github.CheckRun import CheckRun
  51    from github.Issue import Issue
  52    from github.IssueComment import IssueComment
  53    from github.PullRequest import PullRequest
  54    from github.PullRequestReview import PullRequestReview
  55    from github.Repository import Repository
  56
  57logger = logging.getLogger(__name__)
  58
  59
  60class TestFailure(Exception):
  61    pass
  62
  63
  64class PullRequestInfo(PydanticModel):
  65    """Contains information related to a pull request that can be used to construct other objects/URLs"""
  66
  67    owner: str
  68    repo: str
  69    pr_number: int
  70
  71    @property
  72    def full_repo_path(self) -> str:
  73        return "/".join([self.owner, self.repo])
  74
  75    @classmethod
  76    def create_from_pull_request_url(cls, pull_request_url: str) -> PullRequestInfo:
  77        owner, repo, _, pr_number = pull_request_url.split("/")[-4:]
  78        return cls(
  79            owner=owner,
  80            repo=repo,
  81            pr_number=int(pr_number),
  82        )
  83
  84
  85class GithubCheckStatus(str, Enum):
  86    QUEUED = "queued"
  87    IN_PROGRESS = "in_progress"
  88    COMPLETED = "completed"
  89
  90    @property
  91    def is_queued(self) -> bool:
  92        return self == GithubCheckStatus.QUEUED
  93
  94    @property
  95    def is_in_progress(self) -> bool:
  96        return self == GithubCheckStatus.IN_PROGRESS
  97
  98    @property
  99    def is_completed(self) -> bool:
 100        return self == GithubCheckStatus.COMPLETED
 101
 102
 103class GithubCheckConclusion(str, Enum):
 104    SUCCESS = "success"
 105    FAILURE = "failure"
 106    NEUTRAL = "neutral"
 107    CANCELLED = "cancelled"
 108    TIMED_OUT = "timed_out"
 109    ACTION_REQUIRED = "action_required"
 110    SKIPPED = "skipped"
 111
 112    @property
 113    def is_success(self) -> bool:
 114        return self == GithubCheckConclusion.SUCCESS
 115
 116    @property
 117    def is_failure(self) -> bool:
 118        return self == GithubCheckConclusion.FAILURE
 119
 120    @property
 121    def is_neutral(self) -> bool:
 122        return self == GithubCheckConclusion.NEUTRAL
 123
 124    @property
 125    def is_cancelled(self) -> bool:
 126        return self == GithubCheckConclusion.CANCELLED
 127
 128    @property
 129    def is_timed_out(self) -> bool:
 130        return self == GithubCheckConclusion.TIMED_OUT
 131
 132    @property
 133    def is_action_required(self) -> bool:
 134        return self == GithubCheckConclusion.ACTION_REQUIRED
 135
 136    @property
 137    def is_skipped(self) -> bool:
 138        return self == GithubCheckConclusion.SKIPPED
 139
 140
 141class MergeStateStatus(str, Enum):
 142    """
 143    https://docs.github.com/en/graphql/reference/enums#mergestatestatus
 144    """
 145
 146    BEHIND = "behind"
 147    BLOCKED = "blocked"
 148    CLEAN = "clean"
 149    DIRTY = "dirty"
 150    DRAFT = "draft"
 151    HAS_HOOKS = "has_hooks"
 152    UNKNOWN = "unknown"
 153    UNSTABLE = "unstable"
 154
 155    @property
 156    def is_behind(self) -> bool:
 157        return self == MergeStateStatus.BEHIND
 158
 159    @property
 160    def is_blocked(self) -> bool:
 161        return self == MergeStateStatus.BLOCKED
 162
 163    @property
 164    def is_clean(self) -> bool:
 165        return self == MergeStateStatus.CLEAN
 166
 167    @property
 168    def is_dirty(self) -> bool:
 169        return self == MergeStateStatus.DIRTY
 170
 171    @property
 172    def is_draft(self) -> bool:
 173        return self == MergeStateStatus.DRAFT
 174
 175    @property
 176    def is_has_hooks(self) -> bool:
 177        return self == MergeStateStatus.HAS_HOOKS
 178
 179    @property
 180    def is_unknown(self) -> bool:
 181        return self == MergeStateStatus.UNKNOWN
 182
 183    @property
 184    def is_unstable(self) -> bool:
 185        return self == MergeStateStatus.UNSTABLE
 186
 187
 188class BotCommand(Enum):
 189    INVALID = 1
 190    DEPLOY_PROD = 2
 191
 192    @classmethod
 193    def from_comment_body(cls, body: str, namespace: t.Optional[str] = None) -> BotCommand:
 194        body = body.strip()
 195        namespace = namespace.strip() if namespace else ""
 196        input_to_command = {
 197            namespace + "/deploy": cls.DEPLOY_PROD,
 198        }
 199        return input_to_command.get(body, cls.INVALID)
 200
 201    @property
 202    def is_invalid(self) -> bool:
 203        return self == self.INVALID
 204
 205    @property
 206    def is_deploy_prod(self) -> bool:
 207        return self == self.DEPLOY_PROD
 208
 209
 210class GithubEvent:
 211    """
 212    Takes a Github Actions event payload and provides a simple interface to access
 213    """
 214
 215    def __init__(self, payload: t.Dict[str, t.Any]) -> None:
 216        self.payload = payload
 217        self._pull_request_info: t.Optional[PullRequestInfo] = None
 218
 219    @classmethod
 220    def from_obj(cls, obj: t.Dict[str, t.Any]) -> GithubEvent:
 221        return cls(payload=obj)
 222
 223    @classmethod
 224    def from_path(cls, path: t.Union[str, pathlib.Path]) -> GithubEvent:
 225        with open(pathlib.Path(path), "r", encoding="utf-8") as f:
 226            return cls.from_obj(json.load(f))
 227
 228    @classmethod
 229    def from_env(cls) -> GithubEvent:
 230        return cls.from_path(os.environ["GITHUB_EVENT_PATH"])
 231
 232    @property
 233    def is_review(self) -> bool:
 234        return bool(self.payload.get("review"))
 235
 236    @property
 237    def is_comment(self) -> bool:
 238        comment = self.payload.get("comment")
 239        if not comment:
 240            return False
 241        if not comment.get("body"):
 242            return False
 243        return True
 244
 245    @property
 246    def is_comment_added(self) -> bool:
 247        return self.is_comment and self.payload.get("action") != "deleted"
 248
 249    @property
 250    def is_pull_request(self) -> bool:
 251        return bool(self.payload.get("pull_request"))
 252
 253    @property
 254    def is_pull_request_closed(self) -> bool:
 255        return self.is_pull_request and self.payload.get("action") == "closed"
 256
 257    @property
 258    def pull_request_url(self) -> str:
 259        if self.is_review:
 260            return self.payload["review"]["pull_request_url"]
 261        if self.is_comment:
 262            return self.payload["issue"]["pull_request"]["url"]
 263        if self.is_pull_request:
 264            return self.payload["pull_request"]["_links"]["self"]["href"]
 265        raise CICDBotError("Unable to determine pull request url")
 266
 267    @property
 268    def pull_request_info(self) -> PullRequestInfo:
 269        if not self._pull_request_info:
 270            self._pull_request_info = PullRequestInfo.create_from_pull_request_url(
 271                self.pull_request_url
 272            )
 273        return self._pull_request_info
 274
 275    @property
 276    def pull_request_comment_body(self) -> t.Optional[str]:
 277        if self.is_comment_added:
 278            return self.payload["comment"]["body"]
 279        return None
 280
 281
 282class GithubController:
 283    BOT_HEADER_MSG = ":robot: **SQLMesh Bot Info** :robot:"
 284    MAX_BYTE_LENGTH = 65535
 285
 286    def __init__(
 287        self,
 288        paths: t.Union[Path, t.Iterable[Path]],
 289        token: str,
 290        config: t.Optional[t.Union[Config, str]] = None,
 291        event: t.Optional[GithubEvent] = None,
 292        client: t.Optional[Github] = None,
 293        context: t.Optional[Context] = None,
 294    ) -> None:
 295        from github import Github
 296
 297        logger.debug(f"Initializing GithubController with paths: {paths} and config: {config}")
 298
 299        self.config = config
 300        self._paths = paths
 301        self._token = token
 302        self._event = event or GithubEvent.from_env()
 303        logger.debug(f"Github event: {json.dumps(self._event.payload)}")
 304        self._pr_plan_builder: t.Optional[PlanBuilder] = None
 305        self._prod_plan_builder: t.Optional[PlanBuilder] = None
 306        self._prod_plan_with_gaps_builder: t.Optional[PlanBuilder] = None
 307        self._check_run_mapping: t.Dict[str, CheckRun] = {}
 308
 309        if not isinstance(get_console(), MarkdownConsole):
 310            raise CICDBotError("Console must be a markdown console.")
 311        self._console = t.cast(MarkdownConsole, get_console())
 312
 313        from github.Consts import DEFAULT_BASE_URL
 314        from github.Auth import Token
 315
 316        self._client: Github = client or Github(
 317            base_url=os.environ.get("GITHUB_API_URL", DEFAULT_BASE_URL), auth=Token(self._token)
 318        )
 319
 320        self._repo: Repository = self._client.get_repo(
 321            self._event.pull_request_info.full_repo_path, lazy=True
 322        )
 323        self._pull_request: PullRequest = self._repo.get_pull(
 324            self._event.pull_request_info.pr_number
 325        )
 326        self._issue: Issue = self._repo.get_issue(self._event.pull_request_info.pr_number)
 327        self._reviews: t.Iterable[PullRequestReview] = self._pull_request.get_reviews()
 328        # TODO: The python module says that user names can be None and this is not currently handled
 329        self._approvers: t.Set[str] = {
 330            review.user.login or "UNKNOWN"
 331            for review in self._reviews
 332            if review.state.lower() == "approved"
 333        }
 334        logger.debug(f"Approvers: {', '.join(self._approvers)}")
 335        self._context: Context = context or Context(paths=self._paths, config=self.config)
 336
 337        # Bot config needs the context to be initialized
 338        logger.debug(f"Bot config: {self.bot_config.json(indent=2)}")
 339
 340    @property
 341    def deploy_command_enabled(self) -> bool:
 342        return self.bot_config.enable_deploy_command
 343
 344    @property
 345    def is_comment_added(self) -> bool:
 346        return self._event.is_comment_added
 347
 348    @property
 349    def _required_approvers(self) -> t.List[User]:
 350        required_approvers = [
 351            user
 352            for user in self._context.users
 353            if user.is_required_approver and user.github_username
 354        ]
 355        logger.debug(
 356            f"Required approvers: {', '.join(user.github_username for user in required_approvers if user.github_username)}"
 357        )
 358        return required_approvers
 359
 360    @property
 361    def _required_approvers_with_approval(self) -> t.List[User]:
 362        return [
 363            user for user in self._required_approvers if user.github_username in self._approvers
 364        ]
 365
 366    @property
 367    def pr_environment_name(self) -> str:
 368        return Environment.sanitize_name(
 369            "_".join(
 370                [
 371                    self.bot_config.pr_environment_name or self._event.pull_request_info.repo,
 372                    str(self._event.pull_request_info.pr_number),
 373                ]
 374            )
 375        )
 376
 377    @property
 378    def do_required_approval_check(self) -> bool:
 379        """We want to skip required approval check if no users have this role"""
 380        do_required_approval_check = bool(self._required_approvers)
 381        logger.debug(f"Do required approval check: {do_required_approval_check}")
 382        return do_required_approval_check
 383
 384    @property
 385    def has_required_approval(self) -> bool:
 386        """
 387        Check if the PR has a required approver.
 388
 389        TODO: Allow defining requiring some number, or all, required approvers.
 390        """
 391        if not self._required_approvers or self._required_approvers_with_approval:
 392            logger.debug("Has required Approval")
 393            return True
 394        logger.debug("Does not have required approval")
 395        return False
 396
 397    @property
 398    def pr_plan(self) -> Plan:
 399        if not self._pr_plan_builder:
 400            self._pr_plan_builder = self._context.plan_builder(
 401                environment=self.pr_environment_name,
 402                skip_tests=True,
 403                skip_linter=True,
 404                categorizer_config=self.bot_config.auto_categorize_changes,
 405                start=self.bot_config.default_pr_start,
 406                min_intervals=self.bot_config.pr_min_intervals,
 407                skip_backfill=self.bot_config.skip_pr_backfill,
 408                include_unmodified=self.bot_config.pr_include_unmodified,
 409                forward_only=self.forward_only_plan,
 410            )
 411        assert self._pr_plan_builder
 412        return self._pr_plan_builder.build()
 413
 414    @property
 415    def pr_plan_or_none(self) -> t.Optional[Plan]:
 416        try:
 417            return self.pr_plan
 418        except:
 419            return None
 420
 421    @property
 422    def pr_plan_flags(self) -> t.Optional[t.Dict[str, UserProvidedFlags]]:
 423        if pr_plan := self.pr_plan_or_none:
 424            return pr_plan.user_provided_flags
 425        if pr_plan_builder := self._pr_plan_builder:
 426            return pr_plan_builder._user_provided_flags
 427        return None
 428
 429    @property
 430    def prod_plan(self) -> Plan:
 431        if not self._prod_plan_builder:
 432            self._prod_plan_builder = self._context.plan_builder(
 433                c.PROD,
 434                no_gaps=True,
 435                skip_tests=True,
 436                skip_linter=True,
 437                categorizer_config=self.bot_config.auto_categorize_changes,
 438                run=self.bot_config.run_on_deploy_to_prod,
 439                forward_only=self.forward_only_plan,
 440            )
 441        assert self._prod_plan_builder
 442        return self._prod_plan_builder.build()
 443
 444    @property
 445    def prod_plan_with_gaps(self) -> Plan:
 446        if not self._prod_plan_with_gaps_builder:
 447            self._prod_plan_with_gaps_builder = self._context.plan_builder(
 448                c.PROD,
 449                # this is required to highlight any data gaps between this PR environment and prod (since PR environments may only contain a subset of data)
 450                no_gaps=False,
 451                skip_tests=True,
 452                skip_linter=True,
 453                categorizer_config=self.bot_config.auto_categorize_changes,
 454                run=self.bot_config.run_on_deploy_to_prod,
 455                forward_only=self.forward_only_plan,
 456            )
 457        assert self._prod_plan_with_gaps_builder
 458        return self._prod_plan_with_gaps_builder.build()
 459
 460    @property
 461    def bot_config(self) -> GithubCICDBotConfig:
 462        bot_config = self._context.config.cicd_bot or GithubCICDBotConfig(
 463            auto_categorize_changes=self._context.auto_categorize_changes
 464        )
 465        return bot_config
 466
 467    @property
 468    def modified_snapshots(self) -> t.Dict[SnapshotId, t.Union[Snapshot, SnapshotTableInfo]]:
 469        return self.prod_plan_with_gaps.modified_snapshots
 470
 471    @property
 472    def removed_snapshots(self) -> t.Set[SnapshotId]:
 473        return set(self.prod_plan_with_gaps.context_diff.removed_snapshots)
 474
 475    @property
 476    def pr_targets_prod_branch(self) -> bool:
 477        return self._pull_request.base.ref in self.bot_config.prod_branch_names
 478
 479    @property
 480    def forward_only_plan(self) -> bool:
 481        default = self._context.config.plan.forward_only
 482        head_ref = self._pull_request.head.ref
 483        if isinstance(head_ref, str):
 484            return head_ref.endswith(self.bot_config.forward_only_branch_suffix) or default
 485        return default
 486
 487    @classmethod
 488    def _append_output(cls, key: str, value: str) -> None:
 489        """
 490        Appends the given key/value to output so they can be read by following steps
 491        """
 492        logger.debug(f"Setting output. Key: {key}, Value: {value}")
 493
 494        # GitHub Actions sets this environment variable
 495        if output_file := os.environ.get("GITHUB_OUTPUT"):
 496            with open(output_file, "a", encoding="utf-8") as fh:
 497                print(f"{key}={value}", file=fh)
 498
 499    def get_forward_only_plan_post_deployment_tip(self, plan: Plan) -> str:
 500        if not plan.forward_only:
 501            return ""
 502
 503        example_model_name = "<model name>"
 504        for snapshot_id in sorted(plan.snapshots):
 505            snapshot = plan.snapshots[snapshot_id]
 506            if snapshot.is_incremental:
 507                example_model_name = snapshot.node.name
 508                break
 509
 510        return (
 511            "> [!TIP]\n"
 512            "> In order to see this forward-only plan retroactively apply to historical intervals on the production model, run the below for date ranges in scope:\n"
 513            "> \n"
 514            f"> `$ sqlmesh plan --restate-model {example_model_name} --start YYYY-MM-DD --end YYYY-MM-DD`\n"
 515            ">\n"
 516            "> Learn more: https://sqlmesh.readthedocs.io/en/stable/concepts/plans/?h=restate#restatement-plans"
 517        )
 518
 519    def get_plan_summary(self, plan: Plan) -> str:
 520        # use Verbosity.VERY_VERBOSE to prevent the list of models from being truncated
 521        # this is particularly important for the "Models needing backfill" list because
 522        # there is no easy way to tell this otherwise
 523        orig_verbosity = self._console.verbosity
 524        self._console.verbosity = Verbosity.VERY_VERBOSE
 525
 526        try:
 527            # Clear out any output that might exist from prior steps
 528            self._console.consume_captured_output()
 529            if plan.restatements:
 530                self._console._print("\n**Restating models**\n")
 531            else:
 532                self._console.show_environment_difference_summary(
 533                    context_diff=plan.context_diff,
 534                    no_diff=False,
 535                )
 536            if plan.context_diff.has_changes:
 537                self._console.show_model_difference_summary(
 538                    context_diff=plan.context_diff,
 539                    environment_naming_info=plan.environment_naming_info,
 540                    default_catalog=self._context.default_catalog,
 541                    no_diff=False,
 542                )
 543            difference_summary = self._console.consume_captured_output()
 544            self._console._show_missing_dates(plan, self._context.default_catalog)
 545            missing_dates = self._console.consume_captured_output()
 546
 547            plan_flags_section = (
 548                f"\n\n{self._generate_plan_flags_section(plan.user_provided_flags)}"
 549                if plan.user_provided_flags
 550                else ""
 551            )
 552
 553            if not difference_summary and not missing_dates:
 554                return f"No changes to apply.{plan_flags_section}"
 555
 556            warnings_block = self._console.consume_captured_warnings()
 557            errors_block = self._console.consume_captured_errors()
 558
 559            return f"{warnings_block}{errors_block}{difference_summary}\n{missing_dates}{plan_flags_section}"
 560        except PlanError as e:
 561            logger.exception("Plan failed to generate")
 562            return f"Plan failed to generate. Check for pending or unresolved changes. Error: {e}"
 563        finally:
 564            self._console.verbosity = orig_verbosity
 565
 566    def get_pr_environment_summary(
 567        self, conclusion: GithubCheckConclusion, exception: t.Optional[Exception] = None
 568    ) -> str:
 569        heading = ""
 570        summary = ""
 571
 572        if conclusion.is_success:
 573            summary = self._get_pr_environment_summary_success()
 574        elif conclusion.is_action_required:
 575            heading = f":warning: Action Required to create or update PR Environment `{self.pr_environment_name}` :warning:"
 576            summary = self._get_pr_environment_summary_action_required(exception)
 577        elif conclusion.is_failure:
 578            heading = (
 579                f":x: Failed to create or update PR Environment `{self.pr_environment_name}` :x:"
 580            )
 581            summary = self._get_pr_environment_summary_failure(exception)
 582        elif conclusion.is_skipped:
 583            heading = f":next_track_button: Skipped creating or updating PR Environment `{self.pr_environment_name}` :next_track_button:"
 584            summary = self._get_pr_environment_summary_skipped(exception)
 585        else:
 586            heading = f":interrobang: Got an unexpected conclusion: {conclusion.value}"
 587
 588        # note: we just add warnings here, errors will be covered by the "failure" conclusion
 589        if warnings := self._console.consume_captured_warnings():
 590            summary = f"{warnings}\n{summary}"
 591
 592        return f"{heading}\n\n{summary}".strip()
 593
 594    def _get_pr_environment_summary_success(self) -> str:
 595        prod_plan = self.prod_plan_with_gaps
 596
 597        if not prod_plan.has_changes:
 598            summary = "No models were modified in this PR.\n"
 599        else:
 600            intro = self._generate_pr_environment_summary_intro()
 601            summary = intro + self._generate_pr_environment_summary_list(prod_plan)
 602
 603        if prod_plan.user_provided_flags:
 604            summary += self._generate_plan_flags_section(prod_plan.user_provided_flags)
 605
 606        return summary
 607
 608    def _get_pr_environment_summary_skipped(self, exception: t.Optional[Exception] = None) -> str:
 609        if isinstance(exception, NoChangesPlanError):
 610            skip_reason = "No changes were detected compared to the prod environment."
 611        elif isinstance(exception, TestFailure):
 612            skip_reason = "Unit Test(s) Failed so skipping PR creation"
 613        else:
 614            skip_reason = "A prior stage failed resulting in skipping PR creation."
 615
 616        return skip_reason
 617
 618    def _get_pr_environment_summary_action_required(
 619        self, exception: t.Optional[Exception] = None
 620    ) -> str:
 621        plan = self.pr_plan_or_none
 622        if isinstance(exception, UncategorizedPlanError) and plan:
 623            failure_msg = f"The following models could not be categorized automatically:\n"
 624            for snapshot in plan.uncategorized:
 625                failure_msg += f"- {snapshot.name}\n"
 626            failure_msg += (
 627                f"\nRun `sqlmesh plan {self.pr_environment_name}` locally to apply these changes.\n\n"
 628                "If you would like the bot to automatically categorize changes, check the [documentation](https://sqlmesh.readthedocs.io/en/stable/integrations/github/) for more information."
 629            )
 630        else:
 631            failure_msg = "Please check the Actions Workflow logs for more information."
 632
 633        return failure_msg
 634
 635    def _get_pr_environment_summary_failure(self, exception: t.Optional[Exception] = None) -> str:
 636        console_output = self._console.consume_captured_output()
 637        failure_msg = ""
 638
 639        if isinstance(exception, PlanError):
 640            if exception.args and (msg := exception.args[0]) and isinstance(msg, str):
 641                failure_msg += f"*{msg}*\n"
 642            if console_output:
 643                failure_msg += f"\n{console_output}"
 644        elif isinstance(exception, (SQLMeshError, SqlglotError, ValueError)):
 645            # this logic is taken from the global error handler attached to the CLI, which uses `click.echo()` to output the message
 646            # so cant be re-used here because it bypasses the Console
 647            failure_msg = f"**Error:** {str(exception)}"
 648        elif exception:
 649            logger.debug(
 650                "Got unexpected error. Error Type: "
 651                + str(type(exception))
 652                + " Stack trace: "
 653                + traceback.format_exc()
 654            )
 655            failure_msg = f"This is an unexpected error.\n\n**Exception:**\n```\n{traceback.format_exc()}\n```"
 656
 657        if captured_errors := self._console.consume_captured_errors():
 658            failure_msg = f"{captured_errors}\n{failure_msg}"
 659
 660        if plan_flags := self.pr_plan_flags:
 661            failure_msg += f"\n\n{self._generate_plan_flags_section(plan_flags)}"
 662
 663        return failure_msg
 664
 665    def run_tests(self) -> t.Tuple[ModelTextTestResult, str]:
 666        """
 667        Run tests for the PR
 668        """
 669        return self._context._run_tests(verbosity=Verbosity.VERBOSE)
 670
 671    def run_linter(self) -> None:
 672        """
 673        Run linter for the PR
 674        """
 675        self._console.consume_captured_output()
 676        self._context.lint_models()
 677
 678    def _get_or_create_comment(self, header: str = BOT_HEADER_MSG) -> IssueComment:
 679        comment = seq_get(
 680            [comment for comment in self._issue.get_comments() if header in comment.body],
 681            0,
 682        )
 683        if not comment:
 684            logger.debug(f"Did not find comment so creating one with header: {header}")
 685            return self._issue.create_comment(header)
 686        logger.debug(f"Found comment with header: {header}")
 687        return comment
 688
 689    def _get_merge_state_status(self) -> MergeStateStatus:
 690        """
 691        This feature is currently in preview and therefore not available in the python module.
 692        So we query GraphQL directly instead.
 693        """
 694        headers = {
 695            "Authorization": f"Bearer {self._token}",
 696            "Accept": "application/vnd.github.merge-info-preview+json",
 697        }
 698        query = f"""{{
 699            repository(owner: "{self._event.pull_request_info.owner}", name: "{self._event.pull_request_info.repo}") {{
 700                pullRequest(number: {self._event.pull_request_info.pr_number}) {{
 701                    title
 702                    state
 703                    mergeStateStatus
 704                }}
 705            }}
 706        }}"""
 707        request = requests.post(
 708            os.environ["GITHUB_GRAPHQL_URL"],
 709            json={"query": query},
 710            headers=headers,
 711        )
 712        if request.status_code == 200:
 713            merge_status = MergeStateStatus(
 714                request.json()["data"]["repository"]["pullRequest"]["mergeStateStatus"].lower()
 715            )
 716            logger.debug(f"Merge state status: {merge_status.value}")
 717            return merge_status
 718        raise CICDBotError(f"Unable to get merge state status. Error: {request.text}")
 719
 720    def update_sqlmesh_comment_info(
 721        self, value: str, *, dedup_regex: t.Optional[str]
 722    ) -> t.Tuple[bool, IssueComment]:
 723        """
 724        Update the SQLMesh PR Comment for the given lookup key with the given value. If a comment does not exist then
 725        it creates one. It determines the comment to update by looking for a comment with the header. If a dedup
 726        regex is provided then it will check if the value already exists in the comment and if so it will not update
 727        """
 728        comment = self._get_or_create_comment()
 729        if dedup_regex:
 730            # If we find a match against the regex then we just return since the comment has already been posted
 731            if seq_get(re.findall(dedup_regex, comment.body), 0):
 732                return False, comment
 733        full_comment = f"{comment.body}\n{value}"
 734        body, *truncated = self._chunk_up_api_message(f"{full_comment}")
 735        if truncated:
 736            logger.warning(
 737                f"Comment body was too long so we truncated it. Full text: {full_comment}"
 738            )
 739        comment.edit(body=body)
 740        return True, comment
 741
 742    def update_pr_environment(self) -> None:
 743        """
 744        Creates a PR environment from the logic present in the PR. If the PR contains changes that are
 745        uncategorized, then an error will be raised.
 746        """
 747        self._console.consume_captured_output()  # clear output buffer
 748        self._context.apply(self.pr_plan)  # will raise if PR environment creation fails
 749
 750        # update PR info comment
 751        vde_title = "- :eyes: To **review** this PR's changes, use virtual data environment:"
 752        comment_value = f"{vde_title}\n  - `{self.pr_environment_name}`"
 753        if self.bot_config.enable_deploy_command:
 754            full_command = f"{self.bot_config.command_namespace or ''}/deploy"
 755            comment_value += f"\n- :arrow_forward: To **apply** this PR's plan to prod, comment:\n  - `{full_command}`"
 756        dedup_regex = vde_title.replace("*", r"\*") + r".*"
 757        updated_comment, _ = self.update_sqlmesh_comment_info(
 758            value=comment_value,
 759            dedup_regex=dedup_regex,
 760        )
 761        if updated_comment:
 762            self._append_output("created_pr_environment", "true")
 763
 764    def deploy_to_prod(self) -> None:
 765        """
 766        Attempts to deploy a plan to prod. If the plan is not up-to-date or has gaps then it will raise.
 767        """
 768        # If the PR is already merged then we will not deploy to prod if this event was triggered prior to the merge.
 769        # The deploy can still happen if the workflow is configured to listen for `closed` events.
 770        if self._pull_request.merged and not self._event.is_pull_request_closed:
 771            raise CICDBotError(
 772                "PR is already merged and this event was triggered prior to the merge."
 773            )
 774        merge_status = self._get_merge_state_status()
 775        if self.bot_config.check_if_blocked_on_deploy_to_prod and merge_status.is_blocked:
 776            raise CICDBotError(
 777                "Branch protection or ruleset requirement is likely not satisfied, e.g. missing CODEOWNERS approval. "
 778                "Please check PR and resolve any issues. To disable this check, set `check_if_blocked_on_deploy_to_prod` to false in the bot configuration."
 779            )
 780        if merge_status.is_dirty:
 781            raise CICDBotError(
 782                "Merge commit cannot be cleanly created. Likely from a merge conflict. "
 783                "Please check PR and resolve any issues."
 784            )
 785        plan_summary = f"""<details>
 786  <summary>:ship: Prod Plan Being Applied</summary>
 787
 788{self.get_plan_summary(self.prod_plan)}
 789</details>
 790
 791"""
 792        if self.forward_only_plan:
 793            plan_summary = (
 794                f"{self.get_forward_only_plan_post_deployment_tip(self.prod_plan)}\n{plan_summary}"
 795            )
 796
 797        self.update_sqlmesh_comment_info(
 798            value=plan_summary,
 799            dedup_regex=None,
 800        )
 801        self._context.apply(self.prod_plan)
 802
 803    def try_invalidate_pr_environment(self) -> None:
 804        """
 805        Marks the PR environment for garbage collection.
 806        """
 807        if self.bot_config.invalidate_environment_after_deploy:
 808            self._context.invalidate_environment(self.pr_environment_name)
 809
 810    def _update_check(
 811        self,
 812        name: str,
 813        status: GithubCheckStatus,
 814        title: str,
 815        conclusion: t.Optional[GithubCheckConclusion] = None,
 816        full_summary: t.Optional[str] = None,
 817    ) -> None:
 818        """
 819        Updates the status of the merge commit.
 820        """
 821        current_time = now()
 822        kwargs: t.Dict[str, t.Any] = {
 823            "name": name,
 824            # Note: The environment variable `GITHUB_SHA` would be the merge commit so that is why instead we
 825            # get the last commit on the PR.
 826            "head_sha": self._pull_request.head.sha,
 827            "status": status.value,
 828        }
 829        if status.is_in_progress:
 830            kwargs["started_at"] = current_time
 831        if status.is_completed:
 832            kwargs["completed_at"] = current_time
 833        if conclusion:
 834            kwargs["conclusion"] = conclusion.value
 835        full_summary = full_summary or title
 836        summary, text, *truncated = self._chunk_up_api_message(full_summary) + [None]
 837        if truncated and truncated[0] is not None:
 838            logger.warning(f"Summary was too long so we truncated it. Full text: {full_summary}")
 839        kwargs["output"] = {"title": title, "summary": summary}
 840        if text:
 841            kwargs["output"]["text"] = text
 842        logger.debug(f"Updating check with kwargs: {kwargs}")
 843
 844        if self.running_in_github_actions:
 845            # Only make the API call to update the checks if we are running within GitHub Actions
 846            # One very annoying limitation of the Pull Request Checks API is that its only available to GitHub Apps
 847            # and not personal access tokens, which makes it unable to be utilized during local development
 848            if name in self._check_run_mapping:
 849                logger.debug(f"Found check run in mapping so updating it. Name: {name}")
 850                check_run = self._check_run_mapping[name]
 851                check_run.edit(
 852                    **{
 853                        k: v
 854                        for k, v in kwargs.items()
 855                        if k not in ("name", "head_sha", "started_at")
 856                    }
 857                )
 858            else:
 859                logger.debug(f"Did not find check run in mapping so creating it. Name: {name}")
 860                self._check_run_mapping[name] = self._repo.create_check_run(**kwargs)
 861        else:
 862            # Output the summary using print() so the newlines are resolved and the result can easily
 863            # be disambiguated from the rest of the console output and copy+pasted into a Markdown renderer
 864            print(
 865                f"---CHECK OUTPUT START: {kwargs['output']['title']} ---\n{kwargs['output']['summary']}\n---CHECK OUTPUT END---\n"
 866            )
 867
 868        if conclusion:
 869            self._append_output(
 870                word_characters_only(name.replace("SQLMesh - ", "").lower()), conclusion.value
 871            )
 872
 873    def _update_check_handler(
 874        self,
 875        check_name: str,
 876        status: GithubCheckStatus,
 877        conclusion: t.Optional[GithubCheckConclusion],
 878        status_handler: t.Callable[[GithubCheckStatus], t.Tuple[str, t.Optional[str]]],
 879        conclusion_handler: t.Callable[
 880            [GithubCheckConclusion], t.Tuple[GithubCheckConclusion, str, t.Optional[str]]
 881        ],
 882    ) -> None:
 883        if conclusion:
 884            conclusion, title, summary = conclusion_handler(conclusion)
 885        else:
 886            title, summary = status_handler(status)
 887        self._update_check(
 888            name=check_name,
 889            status=status,
 890            title=title,
 891            conclusion=conclusion,
 892            full_summary=summary,
 893        )
 894
 895    def update_linter_check(
 896        self,
 897        status: GithubCheckStatus,
 898        conclusion: t.Optional[GithubCheckConclusion] = None,
 899    ) -> None:
 900        if not self._context.config.linter.enabled:
 901            return
 902
 903        def conclusion_handler(
 904            conclusion: GithubCheckConclusion,
 905        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
 906            linter_summary = self._console.consume_captured_output() or "Linter Success"
 907
 908            title = "Linter results"
 909
 910            return conclusion, title, linter_summary
 911
 912        self._update_check_handler(
 913            check_name="SQLMesh - Linter",
 914            status=status,
 915            conclusion=conclusion,
 916            status_handler=lambda status: (
 917                {
 918                    GithubCheckStatus.IN_PROGRESS: "Running linter",
 919                    GithubCheckStatus.QUEUED: "Waiting to Run linter",
 920                }[status],
 921                None,
 922            ),
 923            conclusion_handler=conclusion_handler,
 924        )
 925
 926    def update_test_check(
 927        self,
 928        status: GithubCheckStatus,
 929        conclusion: t.Optional[GithubCheckConclusion] = None,
 930        result: t.Optional[ModelTextTestResult] = None,
 931        traceback: t.Optional[str] = None,
 932    ) -> None:
 933        """
 934        Updates the status of tests for code in the PR
 935        """
 936
 937        def conclusion_handler(
 938            conclusion: GithubCheckConclusion,
 939            result: t.Optional[ModelTextTestResult],
 940        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
 941            if result:
 942                # Clear out console
 943                self._console.consume_captured_output()
 944                self._console.log_test_results(
 945                    result,
 946                    self._context.test_connection_config._engine_adapter.DIALECT,
 947                )
 948                test_summary = self._console.consume_captured_output()
 949                test_title = "Tests Passed" if result.wasSuccessful() else "Tests Failed"
 950                test_conclusion = (
 951                    GithubCheckConclusion.SUCCESS
 952                    if result.wasSuccessful()
 953                    else GithubCheckConclusion.FAILURE
 954                )
 955                return test_conclusion, test_title, test_summary
 956            if traceback:
 957                self._console._print(traceback)
 958
 959            test_title = "Skipped Tests" if conclusion.is_skipped else "Tests Failed"
 960            return conclusion, test_title, traceback
 961
 962        self._update_check_handler(
 963            check_name="SQLMesh - Run Unit Tests",
 964            status=status,
 965            conclusion=conclusion,
 966            status_handler=lambda status: (
 967                {
 968                    GithubCheckStatus.IN_PROGRESS: "Running Tests",
 969                    GithubCheckStatus.QUEUED: "Waiting to Run Tests",
 970                }[status],
 971                None,
 972            ),
 973            conclusion_handler=functools.partial(conclusion_handler, result=result),
 974        )
 975
 976    def update_required_approval_check(
 977        self, status: GithubCheckStatus, conclusion: t.Optional[GithubCheckConclusion] = None
 978    ) -> None:
 979        """
 980        Updates the status of the merge commit for the required approval.
 981        """
 982
 983        def conclusion_handler(
 984            conclusion: GithubCheckConclusion,
 985        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
 986            test_summary = "**List of possible required approvers:**\n"
 987            for user in self._required_approvers:
 988                test_summary += f"- `{user.github_username or user.username}`\n"
 989
 990            title = (
 991                f"Obtained approval from required approvers: {', '.join([user.github_username or user.username for user in self._required_approvers_with_approval])}"
 992                if conclusion.is_success
 993                else "Need a Required Approval"
 994            )
 995            return conclusion, title, test_summary
 996
 997        # If we get a skip that means required approvers is not configured therefore it does not need to be displayed
 998        if conclusion and conclusion.is_skipped:
 999            return
1000
1001        self._update_check_handler(
1002            check_name="SQLMesh - Has Required Approval",
1003            status=status,
1004            conclusion=conclusion,
1005            status_handler=lambda status: (
1006                {
1007                    GithubCheckStatus.IN_PROGRESS: "Checking if we have required Approvers",
1008                    GithubCheckStatus.QUEUED: "Waiting to Check if we have required Approvers",
1009                }[status],
1010                None,
1011            ),
1012            conclusion_handler=conclusion_handler,
1013        )
1014
1015    def update_pr_environment_check(
1016        self, status: GithubCheckStatus, exception: t.Optional[Exception] = None
1017    ) -> t.Optional[GithubCheckConclusion]:
1018        """
1019        Updates the status of the merge commit for the PR environment.
1020        """
1021        conclusion: t.Optional[GithubCheckConclusion] = None
1022        if isinstance(exception, (NoChangesPlanError, TestFailure, LinterError)):
1023            conclusion = GithubCheckConclusion.SKIPPED
1024        elif isinstance(exception, UncategorizedPlanError):
1025            conclusion = GithubCheckConclusion.ACTION_REQUIRED
1026        elif exception:
1027            conclusion = GithubCheckConclusion.FAILURE
1028        elif status.is_completed:
1029            conclusion = GithubCheckConclusion.SUCCESS
1030
1031        check_title_static = "PR Virtual Data Environment: "
1032        check_title = check_title_static + self.pr_environment_name
1033
1034        def conclusion_handler(
1035            conclusion: GithubCheckConclusion, exception: t.Optional[Exception]
1036        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
1037            summary = self.get_pr_environment_summary(conclusion, exception)
1038            self._append_output("pr_environment_name", self.pr_environment_name)
1039            return conclusion, check_title, summary
1040
1041        self._update_check_handler(
1042            check_name="SQLMesh - PR Environment Synced",
1043            status=status,
1044            conclusion=conclusion,
1045            status_handler=lambda status: (
1046                check_title,
1047                {
1048                    GithubCheckStatus.QUEUED: f":pause_button: Waiting to create or update PR Environment `{self.pr_environment_name}`",
1049                    GithubCheckStatus.IN_PROGRESS: f":rocket: Creating or Updating PR Environment `{self.pr_environment_name}`",
1050                }[status],
1051            ),
1052            conclusion_handler=functools.partial(conclusion_handler, exception=exception),
1053        )
1054        return conclusion
1055
1056    def update_prod_plan_preview_check(
1057        self,
1058        status: GithubCheckStatus,
1059        conclusion: t.Optional[GithubCheckConclusion] = None,
1060        summary: t.Optional[str] = None,
1061    ) -> None:
1062        """
1063        Updates the status of the merge commit for the prod plan preview.
1064        """
1065
1066        def conclusion_handler(
1067            conclusion: GithubCheckConclusion, summary: t.Optional[str] = None
1068        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
1069            conclusion_to_title = {
1070                GithubCheckConclusion.SUCCESS: "Prod Plan Preview",
1071                GithubCheckConclusion.CANCELLED: "Cancelled generating prod plan preview",
1072                GithubCheckConclusion.SKIPPED: "Skipped generating prod plan preview since PR was not synchronized",
1073                GithubCheckConclusion.FAILURE: "Failed to generate prod plan preview",
1074            }
1075            title = conclusion_to_title.get(
1076                conclusion, f"Got an unexpected conclusion: {conclusion.value}"
1077            )
1078            if conclusion == GithubCheckConclusion.SUCCESS and summary:
1079                summary = (
1080                    f"This is a preview that shows the differences between this PR environment `{self.pr_environment_name}` and `prod`.\n\n"
1081                    "These are the changes that would be deployed.\n\n"
1082                ) + summary
1083
1084            return conclusion, title, summary
1085
1086        self._update_check_handler(
1087            check_name="SQLMesh - Prod Plan Preview",
1088            status=status,
1089            conclusion=conclusion,
1090            status_handler=lambda status: (
1091                {
1092                    GithubCheckStatus.IN_PROGRESS: "Generating Prod Plan",
1093                    GithubCheckStatus.QUEUED: "Waiting to Generate Prod Plan",
1094                }[status],
1095                None,
1096            ),
1097            conclusion_handler=functools.partial(conclusion_handler, summary=summary),
1098        )
1099
1100    def update_prod_environment_check(
1101        self,
1102        status: GithubCheckStatus,
1103        conclusion: t.Optional[GithubCheckConclusion] = None,
1104        skip_reason: t.Optional[str] = None,
1105        plan_error: t.Optional[PlanError] = None,
1106    ) -> None:
1107        """
1108        Updates the status of the merge commit for the prod environment.
1109        """
1110
1111        def conclusion_handler(
1112            conclusion: GithubCheckConclusion, skip_reason: t.Optional[str] = None
1113        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
1114            conclusion_to_title = {
1115                GithubCheckConclusion.SUCCESS: "Deployed to Prod",
1116                GithubCheckConclusion.CANCELLED: "Cancelled deploying to prod",
1117                GithubCheckConclusion.SKIPPED: "Skipped deployment",
1118                GithubCheckConclusion.FAILURE: "Failed to deploy to prod",
1119                GithubCheckConclusion.ACTION_REQUIRED: "Failed due to error applying plan",
1120            }
1121            title = (
1122                conclusion_to_title.get(conclusion)
1123                or f"Got an unexpected conclusion: {conclusion.value}"
1124            )
1125            if conclusion.is_skipped:
1126                summary = skip_reason
1127            elif conclusion.is_failure:
1128                captured_errors = self._console.consume_captured_errors()
1129                summary = (
1130                    captured_errors or f"{title}\n\n**Error:**\n```\n{traceback.format_exc()}\n```"
1131                )
1132            elif conclusion.is_action_required:
1133                if plan_error:
1134                    summary = f"**Plan error:**\n```\n{plan_error}\n```"
1135                else:
1136                    summary = "Got an action required conclusion but no plan error was provided. This is unexpected."
1137            else:
1138                summary = "**Generated Prod Plan**\n" + self.get_plan_summary(self.prod_plan)
1139
1140            return conclusion, title, summary
1141
1142        self._update_check_handler(
1143            check_name="SQLMesh - Prod Environment Synced",
1144            status=status,
1145            conclusion=conclusion,
1146            status_handler=lambda status: (
1147                {
1148                    GithubCheckStatus.IN_PROGRESS: "Deploying to Prod",
1149                    GithubCheckStatus.QUEUED: "Waiting to see if we can deploy to prod",
1150                }[status],
1151                None,
1152            ),
1153            conclusion_handler=functools.partial(conclusion_handler, skip_reason=skip_reason),
1154        )
1155
1156    def try_merge_pr(self) -> None:
1157        """
1158        Merges the PR using the merge method defined in the bot config. If one is not defined then a merge is not
1159        performed
1160        """
1161        if self.bot_config.merge_method:
1162            logger.debug(f"Merging PR with merge method: {self.bot_config.merge_method.value}")
1163            self._pull_request.merge(merge_method=self.bot_config.merge_method.value)
1164        else:
1165            logger.debug("No merge method defined so skipping merge")
1166
1167    def get_command_from_comment(self) -> BotCommand:
1168        """
1169        Gets the command from the comment
1170        """
1171        if not self._event.is_comment_added:
1172            logger.debug("Event is not a comment so returning invalid")
1173            return BotCommand.INVALID
1174        if self._event.pull_request_comment_body is None:
1175            raise CICDBotError("Unable to get comment body")
1176        logger.debug(f"Getting command from comment body: {self._event.pull_request_comment_body}")
1177        return BotCommand.from_comment_body(
1178            self._event.pull_request_comment_body, self.bot_config.command_namespace
1179        )
1180
1181    def _chunk_up_api_message(self, message: str) -> t.List[str]:
1182        """
1183        Chunks up the message into `MAX_BYTE_LENGTH` byte chunks
1184        """
1185        message_encoded = message.encode("utf-8")
1186        return [
1187            message_encoded[i : i + self.MAX_BYTE_LENGTH].decode("utf-8", "ignore")
1188            for i in range(0, len(message_encoded), self.MAX_BYTE_LENGTH)
1189        ]
1190
1191    @property
1192    def running_in_github_actions(self) -> bool:
1193        return os.environ.get("GITHUB_ACTIONS", None) == "true"
1194
1195    @property
1196    def version_info(self) -> str:
1197        from sqlmesh.cli.main import _sqlmesh_version
1198
1199        return _sqlmesh_version()
1200
1201    def _generate_plan_flags_section(
1202        self, user_provided_flags: t.Dict[str, UserProvidedFlags]
1203    ) -> str:
1204        # collapsed section syntax:
1205        # https://docs.github.com/en/get-started/writing-on-github/working-with-advanced-formatting/organizing-information-with-collapsed-sections#creating-a-collapsed-section
1206        section = "<details>\n\n<summary>Plan flags</summary>\n\n"
1207        for flag_name, flag_value in user_provided_flags.items():
1208            section += f"- `{flag_name}` = `{flag_value}`\n"
1209        section += "\n</details>"
1210
1211        return section
1212
1213    def _generate_pr_environment_summary_intro(self) -> str:
1214        note = ""
1215        subset_reasons = []
1216
1217        if self.bot_config.skip_pr_backfill:
1218            subset_reasons.append("`skip_pr_backfill` is enabled")
1219
1220        if default_pr_start := self.bot_config.default_pr_start:
1221            subset_reasons.append(f"`default_pr_start` is set to `{default_pr_start}`")
1222
1223        if subset_reasons:
1224            note = (
1225                "> [!IMPORTANT]\n"
1226                f"> This PR environment may only contain a subset of data because:\n"
1227                + "\n".join(f"> - {r}" for r in subset_reasons)
1228                + "\n"
1229                "> \n"
1230                "> This means that deploying to `prod` may not be a simple virtual update if there is still some data to load.\n"
1231                "> See `Dates not loaded in PR` below or the `Prod Plan Preview` check for more information.\n\n"
1232            )
1233
1234        return (
1235            f"Here is a summary of data that has been loaded into the PR environment `{self.pr_environment_name}` and could be deployed to `prod`.\n\n"
1236            + note
1237        )
1238
1239    def _generate_pr_environment_summary_list(self, plan: Plan) -> str:
1240        added_snapshot_ids = set(plan.context_diff.added)
1241        modified_snapshot_ids = set(
1242            s.snapshot_id for s, _ in plan.context_diff.modified_snapshots.values()
1243        )
1244        removed_snapshot_ids = set(plan.context_diff.removed_snapshots.keys())
1245
1246        # note: we sort these to get a deterministic order for the output tests
1247        table_records = sorted(
1248            [
1249                SnapshotSummaryRecord(snapshot_id=snapshot_id, plan=plan)
1250                for snapshot_id in (
1251                    added_snapshot_ids | modified_snapshot_ids | removed_snapshot_ids
1252                )
1253            ],
1254            key=lambda r: r.display_name,
1255        )
1256
1257        sections = [
1258            ("### Added", [r for r in table_records if r.is_added]),
1259            ("### Removed", [r for r in table_records if r.is_removed]),
1260            ("### Directly Modified", [r for r in table_records if r.is_directly_modified]),
1261            ("### Indirectly Modified", [r for r in table_records if r.is_indirectly_modified]),
1262            (
1263                "### Metadata Updated",
1264                [r for r in table_records if r.is_metadata_updated and not r.is_modified],
1265            ),
1266        ]
1267
1268        summary = ""
1269        for title, records in sections:
1270            if records:
1271                summary += f"\n{title}\n"
1272
1273            for record in records:
1274                summary += f"{record.as_markdown_list_item}\n"
1275
1276        return summary
1277
1278
1279@dataclass
1280class SnapshotSummaryRecord:
1281    snapshot_id: SnapshotId
1282    plan: Plan
1283
1284    @property
1285    def snapshot(self) -> Snapshot:
1286        if self.is_removed:
1287            raise ValueError("Removed snapshots only have SnapshotTableInfo available")
1288        return self.plan.snapshots[self.snapshot_id]
1289
1290    @cached_property
1291    def snapshot_table_info(self) -> SnapshotTableInfo:
1292        if self.is_removed:
1293            return self.plan.modified_snapshots[self.snapshot_id].table_info
1294        return self.plan.snapshots[self.snapshot_id].table_info
1295
1296    @property
1297    def display_name(self) -> str:
1298        dialect = None if self.is_removed else self.snapshot.node.dialect
1299        return self.snapshot_table_info.display_name(
1300            self.plan.environment_naming_info, default_catalog=None, dialect=dialect
1301        )
1302
1303    @property
1304    def change_category(self) -> str:
1305        if self.is_removed:
1306            return SNAPSHOT_CHANGE_CATEGORY_STR[SnapshotChangeCategory.BREAKING]
1307
1308        if change_category := self.snapshot.change_category:
1309            return SNAPSHOT_CHANGE_CATEGORY_STR[change_category]
1310
1311        return "Uncategorized"
1312
1313    @property
1314    def is_added(self) -> bool:
1315        return self.snapshot_id in self.plan.context_diff.added
1316
1317    @property
1318    def is_removed(self) -> bool:
1319        return self.snapshot_id in self.plan.context_diff.removed_snapshots
1320
1321    @property
1322    def is_dev_preview(self) -> bool:
1323        return not self.plan.deployability_index.is_deployable(self.snapshot_id)
1324
1325    @property
1326    def is_directly_modified(self) -> bool:
1327        return self.plan.context_diff.directly_modified(self.snapshot_table_info.name)
1328
1329    @property
1330    def is_indirectly_modified(self) -> bool:
1331        return self.plan.context_diff.indirectly_modified(self.snapshot_table_info.name)
1332
1333    @property
1334    def is_modified(self) -> bool:
1335        return self.is_directly_modified or self.is_indirectly_modified
1336
1337    @property
1338    def is_metadata_updated(self) -> bool:
1339        return self.plan.context_diff.metadata_updated(self.snapshot_table_info.name)
1340
1341    @property
1342    def is_incremental(self) -> bool:
1343        return self.snapshot_table_info.is_incremental
1344
1345    @property
1346    def modification_type(self) -> str:
1347        if self.is_directly_modified:
1348            return "Directly modified"
1349        if self.is_indirectly_modified:
1350            return "Indirectly modified"
1351        if self.is_metadata_updated:
1352            return "Metadata updated"
1353
1354        return "Unknown"
1355
1356    @property
1357    def loaded_intervals(self) -> SnapshotIntervals:
1358        if self.is_removed:
1359            raise ValueError("Removed snapshots dont have loaded intervals available")
1360
1361        return SnapshotIntervals(
1362            snapshot_id=self.snapshot_id,
1363            intervals=(
1364                self.snapshot.dev_intervals
1365                if self.snapshot.is_forward_only
1366                else self.snapshot.intervals
1367            ),
1368        )
1369
1370    @property
1371    def loaded_intervals_rendered(self) -> str:
1372        if self.is_removed:
1373            return "REMOVED"
1374
1375        return self._format_intervals(self.loaded_intervals)
1376
1377    @property
1378    def missing_intervals(self) -> t.Optional[SnapshotIntervals]:
1379        return next(
1380            (si for si in self.plan.missing_intervals if si.snapshot_id == self.snapshot_id),
1381            None,
1382        )
1383
1384    @property
1385    def missing_intervals_formatted(self) -> str:
1386        if not self.is_removed and (intervals := self.missing_intervals):
1387            return self._format_intervals(intervals)
1388
1389        return "N/A"
1390
1391    @property
1392    def as_markdown_list_item(self) -> str:
1393        if self.is_removed:
1394            return f"- `{self.display_name}` ({self.change_category})"
1395
1396        how_applied = ""
1397
1398        if not self.is_incremental:
1399            from sqlmesh.core.console import _format_missing_intervals
1400
1401            # note: this is to re-use the '[recreate view]' and '[full refresh]' text and keep it in sync with updates to the CLI
1402            # it doesnt actually use the passed intervals, those are handled differently
1403            how_applied = _format_missing_intervals(self.snapshot, self.loaded_intervals)
1404
1405        how_applied_str = f" [{how_applied}]" if how_applied else ""
1406
1407        item = f"- `{self.display_name}` ({self.change_category})\n"
1408
1409        if self.snapshot_table_info.model_kind_name:
1410            item += f"  **Kind:** {self.snapshot_table_info.model_kind_name}{how_applied_str}\n"
1411
1412        if self.is_incremental:
1413            # in-depth interval info is only relevant for incremental models
1414            item += f"  **Dates loaded in PR:** [{self.loaded_intervals_rendered}]\n"
1415            if self.missing_intervals:
1416                item += f"  **Dates *not* loaded in PR:** [{self.missing_intervals_formatted}]\n"
1417
1418        return item
1419
1420    def _format_intervals(self, intervals: SnapshotIntervals) -> str:
1421        preview_modifier = " (**preview**)" if self.is_dev_preview else ""
1422        return f"{intervals.format_intervals(self.snapshot.node.interval_unit)}{preview_modifier}"