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                preview_start=self.bot_config.default_pr_preview_start,
 408                preview_min_intervals=self.bot_config.pr_preview_min_intervals,
 409                skip_backfill=self.bot_config.skip_pr_backfill,
 410                include_unmodified=self.bot_config.pr_include_unmodified,
 411                forward_only=self.forward_only_plan,
 412            )
 413        assert self._pr_plan_builder
 414        return self._pr_plan_builder.build()
 415
 416    @property
 417    def pr_plan_or_none(self) -> t.Optional[Plan]:
 418        try:
 419            return self.pr_plan
 420        except:
 421            return None
 422
 423    @property
 424    def pr_plan_flags(self) -> t.Optional[t.Dict[str, UserProvidedFlags]]:
 425        if pr_plan := self.pr_plan_or_none:
 426            return pr_plan.user_provided_flags
 427        if pr_plan_builder := self._pr_plan_builder:
 428            return pr_plan_builder._user_provided_flags
 429        return None
 430
 431    @property
 432    def prod_plan(self) -> Plan:
 433        if not self._prod_plan_builder:
 434            self._prod_plan_builder = self._context.plan_builder(
 435                c.PROD,
 436                no_gaps=True,
 437                skip_tests=True,
 438                skip_linter=True,
 439                categorizer_config=self.bot_config.auto_categorize_changes,
 440                run=self.bot_config.run_on_deploy_to_prod,
 441                forward_only=self.forward_only_plan,
 442            )
 443        assert self._prod_plan_builder
 444        return self._prod_plan_builder.build()
 445
 446    @property
 447    def prod_plan_with_gaps(self) -> Plan:
 448        if not self._prod_plan_with_gaps_builder:
 449            self._prod_plan_with_gaps_builder = self._context.plan_builder(
 450                c.PROD,
 451                # this is required to highlight any data gaps between this PR environment and prod (since PR environments may only contain a subset of data)
 452                no_gaps=False,
 453                skip_tests=True,
 454                skip_linter=True,
 455                categorizer_config=self.bot_config.auto_categorize_changes,
 456                run=self.bot_config.run_on_deploy_to_prod,
 457                forward_only=self.forward_only_plan,
 458            )
 459        assert self._prod_plan_with_gaps_builder
 460        return self._prod_plan_with_gaps_builder.build()
 461
 462    @property
 463    def bot_config(self) -> GithubCICDBotConfig:
 464        bot_config = self._context.config.cicd_bot or GithubCICDBotConfig(
 465            auto_categorize_changes=self._context.auto_categorize_changes
 466        )
 467        return bot_config
 468
 469    @property
 470    def modified_snapshots(self) -> t.Dict[SnapshotId, t.Union[Snapshot, SnapshotTableInfo]]:
 471        return self.prod_plan_with_gaps.modified_snapshots
 472
 473    @property
 474    def removed_snapshots(self) -> t.Set[SnapshotId]:
 475        return set(self.prod_plan_with_gaps.context_diff.removed_snapshots)
 476
 477    @property
 478    def pr_targets_prod_branch(self) -> bool:
 479        return self._pull_request.base.ref in self.bot_config.prod_branch_names
 480
 481    @property
 482    def forward_only_plan(self) -> bool:
 483        default = self._context.config.plan.forward_only
 484        head_ref = self._pull_request.head.ref
 485        if isinstance(head_ref, str):
 486            return head_ref.endswith(self.bot_config.forward_only_branch_suffix) or default
 487        return default
 488
 489    @classmethod
 490    def _append_output(cls, key: str, value: str) -> None:
 491        """
 492        Appends the given key/value to output so they can be read by following steps
 493        """
 494        logger.debug(f"Setting output. Key: {key}, Value: {value}")
 495
 496        # GitHub Actions sets this environment variable
 497        if output_file := os.environ.get("GITHUB_OUTPUT"):
 498            with open(output_file, "a", encoding="utf-8") as fh:
 499                print(f"{key}={value}", file=fh)
 500
 501    def get_forward_only_plan_post_deployment_tip(self, plan: Plan) -> str:
 502        if not plan.forward_only:
 503            return ""
 504
 505        example_model_name = "<model name>"
 506        for snapshot_id in sorted(plan.snapshots):
 507            snapshot = plan.snapshots[snapshot_id]
 508            if snapshot.is_incremental:
 509                example_model_name = snapshot.node.name
 510                break
 511
 512        return (
 513            "> [!TIP]\n"
 514            "> 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"
 515            "> \n"
 516            f"> `$ sqlmesh plan --restate-model {example_model_name} --start YYYY-MM-DD --end YYYY-MM-DD`\n"
 517            ">\n"
 518            "> Learn more: https://sqlmesh.readthedocs.io/en/stable/concepts/plans/?h=restate#restatement-plans"
 519        )
 520
 521    def get_plan_summary(self, plan: Plan) -> str:
 522        # use Verbosity.VERY_VERBOSE to prevent the list of models from being truncated
 523        # this is particularly important for the "Models needing backfill" list because
 524        # there is no easy way to tell this otherwise
 525        orig_verbosity = self._console.verbosity
 526        self._console.verbosity = Verbosity.VERY_VERBOSE
 527
 528        try:
 529            # Clear out any output that might exist from prior steps
 530            self._console.consume_captured_output()
 531            if plan.restatements:
 532                self._console._print("\n**Restating models**\n")
 533            else:
 534                self._console.show_environment_difference_summary(
 535                    context_diff=plan.context_diff,
 536                    no_diff=False,
 537                )
 538            if plan.context_diff.has_changes:
 539                self._console.show_model_difference_summary(
 540                    context_diff=plan.context_diff,
 541                    environment_naming_info=plan.environment_naming_info,
 542                    default_catalog=self._context.default_catalog,
 543                    no_diff=False,
 544                )
 545            difference_summary = self._console.consume_captured_output()
 546            self._console._show_missing_dates(plan, self._context.default_catalog)
 547            missing_dates = self._console.consume_captured_output()
 548
 549            plan_flags_section = (
 550                f"\n\n{self._generate_plan_flags_section(plan.user_provided_flags)}"
 551                if plan.user_provided_flags
 552                else ""
 553            )
 554
 555            if not difference_summary and not missing_dates:
 556                return f"No changes to apply.{plan_flags_section}"
 557
 558            warnings_block = self._console.consume_captured_warnings()
 559            errors_block = self._console.consume_captured_errors()
 560
 561            return f"{warnings_block}{errors_block}{difference_summary}\n{missing_dates}{plan_flags_section}"
 562        except PlanError as e:
 563            logger.exception("Plan failed to generate")
 564            return f"Plan failed to generate. Check for pending or unresolved changes. Error: {e}"
 565        finally:
 566            self._console.verbosity = orig_verbosity
 567
 568    def get_pr_environment_summary(
 569        self, conclusion: GithubCheckConclusion, exception: t.Optional[Exception] = None
 570    ) -> str:
 571        heading = ""
 572        summary = ""
 573
 574        if conclusion.is_success:
 575            summary = self._get_pr_environment_summary_success()
 576        elif conclusion.is_action_required:
 577            heading = f":warning: Action Required to create or update PR Environment `{self.pr_environment_name}` :warning:"
 578            summary = self._get_pr_environment_summary_action_required(exception)
 579        elif conclusion.is_failure:
 580            heading = (
 581                f":x: Failed to create or update PR Environment `{self.pr_environment_name}` :x:"
 582            )
 583            summary = self._get_pr_environment_summary_failure(exception)
 584        elif conclusion.is_skipped:
 585            heading = f":next_track_button: Skipped creating or updating PR Environment `{self.pr_environment_name}` :next_track_button:"
 586            summary = self._get_pr_environment_summary_skipped(exception)
 587        else:
 588            heading = f":interrobang: Got an unexpected conclusion: {conclusion.value}"
 589
 590        # note: we just add warnings here, errors will be covered by the "failure" conclusion
 591        if warnings := self._console.consume_captured_warnings():
 592            summary = f"{warnings}\n{summary}"
 593
 594        return f"{heading}\n\n{summary}".strip()
 595
 596    def _get_pr_environment_summary_success(self) -> str:
 597        prod_plan = self.prod_plan_with_gaps
 598
 599        if not prod_plan.has_changes:
 600            summary = "No models were modified in this PR.\n"
 601        else:
 602            intro = self._generate_pr_environment_summary_intro()
 603            summary = intro + self._generate_pr_environment_summary_list(prod_plan)
 604
 605        if prod_plan.user_provided_flags:
 606            summary += self._generate_plan_flags_section(prod_plan.user_provided_flags)
 607
 608        return summary
 609
 610    def _get_pr_environment_summary_skipped(self, exception: t.Optional[Exception] = None) -> str:
 611        if isinstance(exception, NoChangesPlanError):
 612            skip_reason = "No changes were detected compared to the prod environment."
 613        elif isinstance(exception, TestFailure):
 614            skip_reason = "Unit Test(s) Failed so skipping PR creation"
 615        else:
 616            skip_reason = "A prior stage failed resulting in skipping PR creation."
 617
 618        return skip_reason
 619
 620    def _get_pr_environment_summary_action_required(
 621        self, exception: t.Optional[Exception] = None
 622    ) -> str:
 623        plan = self.pr_plan_or_none
 624        if isinstance(exception, UncategorizedPlanError) and plan:
 625            failure_msg = f"The following models could not be categorized automatically:\n"
 626            for snapshot in plan.uncategorized:
 627                failure_msg += f"- {snapshot.name}\n"
 628            failure_msg += (
 629                f"\nRun `sqlmesh plan {self.pr_environment_name}` locally to apply these changes.\n\n"
 630                "If you would like the bot to automatically categorize changes, check the [documentation](https://sqlmesh.readthedocs.io/en/stable/integrations/github/) for more information."
 631            )
 632        else:
 633            failure_msg = "Please check the Actions Workflow logs for more information."
 634
 635        return failure_msg
 636
 637    def _get_pr_environment_summary_failure(self, exception: t.Optional[Exception] = None) -> str:
 638        console_output = self._console.consume_captured_output()
 639        failure_msg = ""
 640
 641        if isinstance(exception, PlanError):
 642            if exception.args and (msg := exception.args[0]) and isinstance(msg, str):
 643                failure_msg += f"*{msg}*\n"
 644            if console_output:
 645                failure_msg += f"\n{console_output}"
 646        elif isinstance(exception, (SQLMeshError, SqlglotError, ValueError)):
 647            # this logic is taken from the global error handler attached to the CLI, which uses `click.echo()` to output the message
 648            # so cant be re-used here because it bypasses the Console
 649            failure_msg = f"**Error:** {str(exception)}"
 650        elif exception:
 651            logger.debug(
 652                "Got unexpected error. Error Type: "
 653                + str(type(exception))
 654                + " Stack trace: "
 655                + traceback.format_exc()
 656            )
 657            failure_msg = f"This is an unexpected error.\n\n**Exception:**\n```\n{traceback.format_exc()}\n```"
 658
 659        if captured_errors := self._console.consume_captured_errors():
 660            failure_msg = f"{captured_errors}\n{failure_msg}"
 661
 662        if plan_flags := self.pr_plan_flags:
 663            failure_msg += f"\n\n{self._generate_plan_flags_section(plan_flags)}"
 664
 665        return failure_msg
 666
 667    def run_tests(self) -> t.Tuple[ModelTextTestResult, str]:
 668        """
 669        Run tests for the PR
 670        """
 671        return self._context._run_tests(verbosity=Verbosity.VERBOSE)
 672
 673    def run_linter(self) -> None:
 674        """
 675        Run linter for the PR
 676        """
 677        self._console.consume_captured_output()
 678        self._context.lint_models()
 679
 680    def _get_or_create_comment(self, header: str = BOT_HEADER_MSG) -> IssueComment:
 681        comment = seq_get(
 682            [comment for comment in self._issue.get_comments() if header in comment.body],
 683            0,
 684        )
 685        if not comment:
 686            logger.debug(f"Did not find comment so creating one with header: {header}")
 687            return self._issue.create_comment(header)
 688        logger.debug(f"Found comment with header: {header}")
 689        return comment
 690
 691    def _get_merge_state_status(self) -> MergeStateStatus:
 692        """
 693        This feature is currently in preview and therefore not available in the python module.
 694        So we query GraphQL directly instead.
 695        """
 696        headers = {
 697            "Authorization": f"Bearer {self._token}",
 698            "Accept": "application/vnd.github.merge-info-preview+json",
 699        }
 700        query = f"""{{
 701            repository(owner: "{self._event.pull_request_info.owner}", name: "{self._event.pull_request_info.repo}") {{
 702                pullRequest(number: {self._event.pull_request_info.pr_number}) {{
 703                    title
 704                    state
 705                    mergeStateStatus
 706                }}
 707            }}
 708        }}"""
 709        request = requests.post(
 710            os.environ["GITHUB_GRAPHQL_URL"],
 711            json={"query": query},
 712            headers=headers,
 713        )
 714        if request.status_code == 200:
 715            merge_status = MergeStateStatus(
 716                request.json()["data"]["repository"]["pullRequest"]["mergeStateStatus"].lower()
 717            )
 718            logger.debug(f"Merge state status: {merge_status.value}")
 719            return merge_status
 720        raise CICDBotError(f"Unable to get merge state status. Error: {request.text}")
 721
 722    def update_sqlmesh_comment_info(
 723        self, value: str, *, dedup_regex: t.Optional[str]
 724    ) -> t.Tuple[bool, IssueComment]:
 725        """
 726        Update the SQLMesh PR Comment for the given lookup key with the given value. If a comment does not exist then
 727        it creates one. It determines the comment to update by looking for a comment with the header. If a dedup
 728        regex is provided then it will check if the value already exists in the comment and if so it will not update
 729        """
 730        comment = self._get_or_create_comment()
 731        if dedup_regex:
 732            # If we find a match against the regex then we just return since the comment has already been posted
 733            if seq_get(re.findall(dedup_regex, comment.body), 0):
 734                return False, comment
 735        full_comment = f"{comment.body}\n{value}"
 736        body, *truncated = self._chunk_up_api_message(f"{full_comment}")
 737        if truncated:
 738            logger.warning(
 739                f"Comment body was too long so we truncated it. Full text: {full_comment}"
 740            )
 741        comment.edit(body=body)
 742        return True, comment
 743
 744    def update_pr_environment(self) -> None:
 745        """
 746        Creates a PR environment from the logic present in the PR. If the PR contains changes that are
 747        uncategorized, then an error will be raised.
 748        """
 749        self._console.consume_captured_output()  # clear output buffer
 750        self._context.apply(self.pr_plan)  # will raise if PR environment creation fails
 751
 752        # update PR info comment
 753        vde_title = "- :eyes: To **review** this PR's changes, use virtual data environment:"
 754        comment_value = f"{vde_title}\n  - `{self.pr_environment_name}`"
 755        if self.bot_config.enable_deploy_command:
 756            full_command = f"{self.bot_config.command_namespace or ''}/deploy"
 757            comment_value += f"\n- :arrow_forward: To **apply** this PR's plan to prod, comment:\n  - `{full_command}`"
 758        dedup_regex = vde_title.replace("*", r"\*") + r".*"
 759        updated_comment, _ = self.update_sqlmesh_comment_info(
 760            value=comment_value,
 761            dedup_regex=dedup_regex,
 762        )
 763        if updated_comment:
 764            self._append_output("created_pr_environment", "true")
 765
 766    def deploy_to_prod(self) -> None:
 767        """
 768        Attempts to deploy a plan to prod. If the plan is not up-to-date or has gaps then it will raise.
 769        """
 770        # If the PR is already merged then we will not deploy to prod if this event was triggered prior to the merge.
 771        # The deploy can still happen if the workflow is configured to listen for `closed` events.
 772        if self._pull_request.merged and not self._event.is_pull_request_closed:
 773            raise CICDBotError(
 774                "PR is already merged and this event was triggered prior to the merge."
 775            )
 776        merge_status = self._get_merge_state_status()
 777        if self.bot_config.check_if_blocked_on_deploy_to_prod and merge_status.is_blocked:
 778            raise CICDBotError(
 779                "Branch protection or ruleset requirement is likely not satisfied, e.g. missing CODEOWNERS approval. "
 780                "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."
 781            )
 782        if merge_status.is_dirty:
 783            raise CICDBotError(
 784                "Merge commit cannot be cleanly created. Likely from a merge conflict. "
 785                "Please check PR and resolve any issues."
 786            )
 787        plan_summary = f"""<details>
 788  <summary>:ship: Prod Plan Being Applied</summary>
 789
 790{self.get_plan_summary(self.prod_plan)}
 791</details>
 792
 793"""
 794        if self.forward_only_plan:
 795            plan_summary = (
 796                f"{self.get_forward_only_plan_post_deployment_tip(self.prod_plan)}\n{plan_summary}"
 797            )
 798
 799        self.update_sqlmesh_comment_info(
 800            value=plan_summary,
 801            dedup_regex=None,
 802        )
 803        self._context.apply(self.prod_plan)
 804
 805    def try_invalidate_pr_environment(self) -> None:
 806        """
 807        Marks the PR environment for garbage collection.
 808        """
 809        if self.bot_config.invalidate_environment_after_deploy:
 810            self._context.invalidate_environment(self.pr_environment_name)
 811
 812    def _update_check(
 813        self,
 814        name: str,
 815        status: GithubCheckStatus,
 816        title: str,
 817        conclusion: t.Optional[GithubCheckConclusion] = None,
 818        full_summary: t.Optional[str] = None,
 819    ) -> None:
 820        """
 821        Updates the status of the merge commit.
 822        """
 823        current_time = now()
 824        kwargs: t.Dict[str, t.Any] = {
 825            "name": name,
 826            # Note: The environment variable `GITHUB_SHA` would be the merge commit so that is why instead we
 827            # get the last commit on the PR.
 828            "head_sha": self._pull_request.head.sha,
 829            "status": status.value,
 830        }
 831        if status.is_in_progress:
 832            kwargs["started_at"] = current_time
 833        if status.is_completed:
 834            kwargs["completed_at"] = current_time
 835        if conclusion:
 836            kwargs["conclusion"] = conclusion.value
 837        full_summary = full_summary or title
 838        summary, text, *truncated = self._chunk_up_api_message(full_summary) + [None]
 839        if truncated and truncated[0] is not None:
 840            logger.warning(f"Summary was too long so we truncated it. Full text: {full_summary}")
 841        kwargs["output"] = {"title": title, "summary": summary}
 842        if text:
 843            kwargs["output"]["text"] = text
 844        logger.debug(f"Updating check with kwargs: {kwargs}")
 845
 846        if self.running_in_github_actions:
 847            # Only make the API call to update the checks if we are running within GitHub Actions
 848            # One very annoying limitation of the Pull Request Checks API is that its only available to GitHub Apps
 849            # and not personal access tokens, which makes it unable to be utilized during local development
 850            if name in self._check_run_mapping:
 851                logger.debug(f"Found check run in mapping so updating it. Name: {name}")
 852                check_run = self._check_run_mapping[name]
 853                check_run.edit(
 854                    **{
 855                        k: v
 856                        for k, v in kwargs.items()
 857                        if k not in ("name", "head_sha", "started_at")
 858                    }
 859                )
 860            else:
 861                logger.debug(f"Did not find check run in mapping so creating it. Name: {name}")
 862                self._check_run_mapping[name] = self._repo.create_check_run(**kwargs)
 863        else:
 864            # Output the summary using print() so the newlines are resolved and the result can easily
 865            # be disambiguated from the rest of the console output and copy+pasted into a Markdown renderer
 866            print(
 867                f"---CHECK OUTPUT START: {kwargs['output']['title']} ---\n{kwargs['output']['summary']}\n---CHECK OUTPUT END---\n"
 868            )
 869
 870        if conclusion:
 871            self._append_output(
 872                word_characters_only(name.replace("SQLMesh - ", "").lower()), conclusion.value
 873            )
 874
 875    def _update_check_handler(
 876        self,
 877        check_name: str,
 878        status: GithubCheckStatus,
 879        conclusion: t.Optional[GithubCheckConclusion],
 880        status_handler: t.Callable[[GithubCheckStatus], t.Tuple[str, t.Optional[str]]],
 881        conclusion_handler: t.Callable[
 882            [GithubCheckConclusion], t.Tuple[GithubCheckConclusion, str, t.Optional[str]]
 883        ],
 884    ) -> None:
 885        if conclusion:
 886            conclusion, title, summary = conclusion_handler(conclusion)
 887        else:
 888            title, summary = status_handler(status)
 889        self._update_check(
 890            name=check_name,
 891            status=status,
 892            title=title,
 893            conclusion=conclusion,
 894            full_summary=summary,
 895        )
 896
 897    def update_linter_check(
 898        self,
 899        status: GithubCheckStatus,
 900        conclusion: t.Optional[GithubCheckConclusion] = None,
 901    ) -> None:
 902        if not self._context.config.linter.enabled:
 903            return
 904
 905        def conclusion_handler(
 906            conclusion: GithubCheckConclusion,
 907        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
 908            linter_summary = self._console.consume_captured_output() or "Linter Success"
 909
 910            title = "Linter results"
 911
 912            return conclusion, title, linter_summary
 913
 914        self._update_check_handler(
 915            check_name="SQLMesh - Linter",
 916            status=status,
 917            conclusion=conclusion,
 918            status_handler=lambda status: (
 919                {
 920                    GithubCheckStatus.IN_PROGRESS: "Running linter",
 921                    GithubCheckStatus.QUEUED: "Waiting to Run linter",
 922                }[status],
 923                None,
 924            ),
 925            conclusion_handler=conclusion_handler,
 926        )
 927
 928    def update_test_check(
 929        self,
 930        status: GithubCheckStatus,
 931        conclusion: t.Optional[GithubCheckConclusion] = None,
 932        result: t.Optional[ModelTextTestResult] = None,
 933        traceback: t.Optional[str] = None,
 934    ) -> None:
 935        """
 936        Updates the status of tests for code in the PR
 937        """
 938
 939        def conclusion_handler(
 940            conclusion: GithubCheckConclusion,
 941            result: t.Optional[ModelTextTestResult],
 942        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
 943            if result:
 944                # Clear out console
 945                self._console.consume_captured_output()
 946                self._console.log_test_results(
 947                    result,
 948                    self._context.test_connection_config._engine_adapter.DIALECT,
 949                )
 950                test_summary = self._console.consume_captured_output()
 951                test_title = "Tests Passed" if result.wasSuccessful() else "Tests Failed"
 952                test_conclusion = (
 953                    GithubCheckConclusion.SUCCESS
 954                    if result.wasSuccessful()
 955                    else GithubCheckConclusion.FAILURE
 956                )
 957                return test_conclusion, test_title, test_summary
 958            if traceback:
 959                self._console._print(traceback)
 960
 961            test_title = "Skipped Tests" if conclusion.is_skipped else "Tests Failed"
 962            return conclusion, test_title, traceback
 963
 964        self._update_check_handler(
 965            check_name="SQLMesh - Run Unit Tests",
 966            status=status,
 967            conclusion=conclusion,
 968            status_handler=lambda status: (
 969                {
 970                    GithubCheckStatus.IN_PROGRESS: "Running Tests",
 971                    GithubCheckStatus.QUEUED: "Waiting to Run Tests",
 972                }[status],
 973                None,
 974            ),
 975            conclusion_handler=functools.partial(conclusion_handler, result=result),
 976        )
 977
 978    def update_required_approval_check(
 979        self, status: GithubCheckStatus, conclusion: t.Optional[GithubCheckConclusion] = None
 980    ) -> None:
 981        """
 982        Updates the status of the merge commit for the required approval.
 983        """
 984
 985        def conclusion_handler(
 986            conclusion: GithubCheckConclusion,
 987        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
 988            test_summary = "**List of possible required approvers:**\n"
 989            for user in self._required_approvers:
 990                test_summary += f"- `{user.github_username or user.username}`\n"
 991
 992            title = (
 993                f"Obtained approval from required approvers: {', '.join([user.github_username or user.username for user in self._required_approvers_with_approval])}"
 994                if conclusion.is_success
 995                else "Need a Required Approval"
 996            )
 997            return conclusion, title, test_summary
 998
 999        # If we get a skip that means required approvers is not configured therefore it does not need to be displayed
1000        if conclusion and conclusion.is_skipped:
1001            return
1002
1003        self._update_check_handler(
1004            check_name="SQLMesh - Has Required Approval",
1005            status=status,
1006            conclusion=conclusion,
1007            status_handler=lambda status: (
1008                {
1009                    GithubCheckStatus.IN_PROGRESS: "Checking if we have required Approvers",
1010                    GithubCheckStatus.QUEUED: "Waiting to Check if we have required Approvers",
1011                }[status],
1012                None,
1013            ),
1014            conclusion_handler=conclusion_handler,
1015        )
1016
1017    def update_pr_environment_check(
1018        self, status: GithubCheckStatus, exception: t.Optional[Exception] = None
1019    ) -> t.Optional[GithubCheckConclusion]:
1020        """
1021        Updates the status of the merge commit for the PR environment.
1022        """
1023        conclusion: t.Optional[GithubCheckConclusion] = None
1024        if isinstance(exception, (NoChangesPlanError, TestFailure, LinterError)):
1025            conclusion = GithubCheckConclusion.SKIPPED
1026        elif isinstance(exception, UncategorizedPlanError):
1027            conclusion = GithubCheckConclusion.ACTION_REQUIRED
1028        elif exception:
1029            conclusion = GithubCheckConclusion.FAILURE
1030        elif status.is_completed:
1031            conclusion = GithubCheckConclusion.SUCCESS
1032
1033        check_title_static = "PR Virtual Data Environment: "
1034        check_title = check_title_static + self.pr_environment_name
1035
1036        def conclusion_handler(
1037            conclusion: GithubCheckConclusion, exception: t.Optional[Exception]
1038        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
1039            summary = self.get_pr_environment_summary(conclusion, exception)
1040            self._append_output("pr_environment_name", self.pr_environment_name)
1041            return conclusion, check_title, summary
1042
1043        self._update_check_handler(
1044            check_name="SQLMesh - PR Environment Synced",
1045            status=status,
1046            conclusion=conclusion,
1047            status_handler=lambda status: (
1048                check_title,
1049                {
1050                    GithubCheckStatus.QUEUED: f":pause_button: Waiting to create or update PR Environment `{self.pr_environment_name}`",
1051                    GithubCheckStatus.IN_PROGRESS: f":rocket: Creating or Updating PR Environment `{self.pr_environment_name}`",
1052                }[status],
1053            ),
1054            conclusion_handler=functools.partial(conclusion_handler, exception=exception),
1055        )
1056        return conclusion
1057
1058    def update_prod_plan_preview_check(
1059        self,
1060        status: GithubCheckStatus,
1061        conclusion: t.Optional[GithubCheckConclusion] = None,
1062        summary: t.Optional[str] = None,
1063    ) -> None:
1064        """
1065        Updates the status of the merge commit for the prod plan preview.
1066        """
1067
1068        def conclusion_handler(
1069            conclusion: GithubCheckConclusion, summary: t.Optional[str] = None
1070        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
1071            conclusion_to_title = {
1072                GithubCheckConclusion.SUCCESS: "Prod Plan Preview",
1073                GithubCheckConclusion.CANCELLED: "Cancelled generating prod plan preview",
1074                GithubCheckConclusion.SKIPPED: "Skipped generating prod plan preview since PR was not synchronized",
1075                GithubCheckConclusion.FAILURE: "Failed to generate prod plan preview",
1076            }
1077            title = conclusion_to_title.get(
1078                conclusion, f"Got an unexpected conclusion: {conclusion.value}"
1079            )
1080            if conclusion == GithubCheckConclusion.SUCCESS and summary:
1081                summary = (
1082                    f"This is a preview that shows the differences between this PR environment `{self.pr_environment_name}` and `prod`.\n\n"
1083                    "These are the changes that would be deployed.\n\n"
1084                ) + summary
1085
1086            return conclusion, title, summary
1087
1088        self._update_check_handler(
1089            check_name="SQLMesh - Prod Plan Preview",
1090            status=status,
1091            conclusion=conclusion,
1092            status_handler=lambda status: (
1093                {
1094                    GithubCheckStatus.IN_PROGRESS: "Generating Prod Plan",
1095                    GithubCheckStatus.QUEUED: "Waiting to Generate Prod Plan",
1096                }[status],
1097                None,
1098            ),
1099            conclusion_handler=functools.partial(conclusion_handler, summary=summary),
1100        )
1101
1102    def update_prod_environment_check(
1103        self,
1104        status: GithubCheckStatus,
1105        conclusion: t.Optional[GithubCheckConclusion] = None,
1106        skip_reason: t.Optional[str] = None,
1107        plan_error: t.Optional[PlanError] = None,
1108    ) -> None:
1109        """
1110        Updates the status of the merge commit for the prod environment.
1111        """
1112
1113        def conclusion_handler(
1114            conclusion: GithubCheckConclusion, skip_reason: t.Optional[str] = None
1115        ) -> t.Tuple[GithubCheckConclusion, str, t.Optional[str]]:
1116            conclusion_to_title = {
1117                GithubCheckConclusion.SUCCESS: "Deployed to Prod",
1118                GithubCheckConclusion.CANCELLED: "Cancelled deploying to prod",
1119                GithubCheckConclusion.SKIPPED: "Skipped deployment",
1120                GithubCheckConclusion.FAILURE: "Failed to deploy to prod",
1121                GithubCheckConclusion.ACTION_REQUIRED: "Failed due to error applying plan",
1122            }
1123            title = (
1124                conclusion_to_title.get(conclusion)
1125                or f"Got an unexpected conclusion: {conclusion.value}"
1126            )
1127            if conclusion.is_skipped:
1128                summary = skip_reason
1129            elif conclusion.is_failure:
1130                captured_errors = self._console.consume_captured_errors()
1131                summary = (
1132                    captured_errors or f"{title}\n\n**Error:**\n```\n{traceback.format_exc()}\n```"
1133                )
1134            elif conclusion.is_action_required:
1135                if plan_error:
1136                    summary = f"**Plan error:**\n```\n{plan_error}\n```"
1137                else:
1138                    summary = "Got an action required conclusion but no plan error was provided. This is unexpected."
1139            else:
1140                summary = "**Generated Prod Plan**\n" + self.get_plan_summary(self.prod_plan)
1141
1142            return conclusion, title, summary
1143
1144        self._update_check_handler(
1145            check_name="SQLMesh - Prod Environment Synced",
1146            status=status,
1147            conclusion=conclusion,
1148            status_handler=lambda status: (
1149                {
1150                    GithubCheckStatus.IN_PROGRESS: "Deploying to Prod",
1151                    GithubCheckStatus.QUEUED: "Waiting to see if we can deploy to prod",
1152                }[status],
1153                None,
1154            ),
1155            conclusion_handler=functools.partial(conclusion_handler, skip_reason=skip_reason),
1156        )
1157
1158    def try_merge_pr(self) -> None:
1159        """
1160        Merges the PR using the merge method defined in the bot config. If one is not defined then a merge is not
1161        performed
1162        """
1163        if self.bot_config.merge_method:
1164            logger.debug(f"Merging PR with merge method: {self.bot_config.merge_method.value}")
1165            self._pull_request.merge(merge_method=self.bot_config.merge_method.value)
1166        else:
1167            logger.debug("No merge method defined so skipping merge")
1168
1169    def get_command_from_comment(self) -> BotCommand:
1170        """
1171        Gets the command from the comment
1172        """
1173        if not self._event.is_comment_added:
1174            logger.debug("Event is not a comment so returning invalid")
1175            return BotCommand.INVALID
1176        if self._event.pull_request_comment_body is None:
1177            raise CICDBotError("Unable to get comment body")
1178        logger.debug(f"Getting command from comment body: {self._event.pull_request_comment_body}")
1179        return BotCommand.from_comment_body(
1180            self._event.pull_request_comment_body, self.bot_config.command_namespace
1181        )
1182
1183    def _chunk_up_api_message(self, message: str) -> t.List[str]:
1184        """
1185        Chunks up the message into `MAX_BYTE_LENGTH` byte chunks
1186        """
1187        message_encoded = message.encode("utf-8")
1188        return [
1189            message_encoded[i : i + self.MAX_BYTE_LENGTH].decode("utf-8", "ignore")
1190            for i in range(0, len(message_encoded), self.MAX_BYTE_LENGTH)
1191        ]
1192
1193    @property
1194    def running_in_github_actions(self) -> bool:
1195        return os.environ.get("GITHUB_ACTIONS", None) == "true"
1196
1197    @property
1198    def version_info(self) -> str:
1199        from sqlmesh.cli.main import _sqlmesh_version
1200
1201        return _sqlmesh_version()
1202
1203    def _generate_plan_flags_section(
1204        self, user_provided_flags: t.Dict[str, UserProvidedFlags]
1205    ) -> str:
1206        # collapsed section syntax:
1207        # https://docs.github.com/en/get-started/writing-on-github/working-with-advanced-formatting/organizing-information-with-collapsed-sections#creating-a-collapsed-section
1208        section = "<details>\n\n<summary>Plan flags</summary>\n\n"
1209        for flag_name, flag_value in user_provided_flags.items():
1210            section += f"- `{flag_name}` = `{flag_value}`\n"
1211        section += "\n</details>"
1212
1213        return section
1214
1215    def _generate_pr_environment_summary_intro(self) -> str:
1216        note = ""
1217        subset_reasons = []
1218
1219        if self.bot_config.skip_pr_backfill:
1220            subset_reasons.append("`skip_pr_backfill` is enabled")
1221
1222        if default_pr_start := self.bot_config.default_pr_start:
1223            subset_reasons.append(f"`default_pr_start` is set to `{default_pr_start}`")
1224
1225        if subset_reasons:
1226            note = (
1227                "> [!IMPORTANT]\n"
1228                f"> This PR environment may only contain a subset of data because:\n"
1229                + "\n".join(f"> - {r}" for r in subset_reasons)
1230                + "\n"
1231                "> \n"
1232                "> This means that deploying to `prod` may not be a simple virtual update if there is still some data to load.\n"
1233                "> See `Dates not loaded in PR` below or the `Prod Plan Preview` check for more information.\n\n"
1234            )
1235
1236        return (
1237            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"
1238            + note
1239        )
1240
1241    def _generate_pr_environment_summary_list(self, plan: Plan) -> str:
1242        added_snapshot_ids = set(plan.context_diff.added)
1243        modified_snapshot_ids = set(
1244            s.snapshot_id for s, _ in plan.context_diff.modified_snapshots.values()
1245        )
1246        removed_snapshot_ids = set(plan.context_diff.removed_snapshots.keys())
1247
1248        # note: we sort these to get a deterministic order for the output tests
1249        table_records = sorted(
1250            [
1251                SnapshotSummaryRecord(snapshot_id=snapshot_id, plan=plan)
1252                for snapshot_id in (
1253                    added_snapshot_ids | modified_snapshot_ids | removed_snapshot_ids
1254                )
1255            ],
1256            key=lambda r: r.display_name,
1257        )
1258
1259        sections = [
1260            ("### Added", [r for r in table_records if r.is_added]),
1261            ("### Removed", [r for r in table_records if r.is_removed]),
1262            ("### Directly Modified", [r for r in table_records if r.is_directly_modified]),
1263            ("### Indirectly Modified", [r for r in table_records if r.is_indirectly_modified]),
1264            (
1265                "### Metadata Updated",
1266                [r for r in table_records if r.is_metadata_updated and not r.is_modified],
1267            ),
1268        ]
1269
1270        summary = ""
1271        for title, records in sections:
1272            if records:
1273                summary += f"\n{title}\n"
1274
1275            for record in records:
1276                summary += f"{record.as_markdown_list_item}\n"
1277
1278        return summary
1279
1280
1281@dataclass
1282class SnapshotSummaryRecord:
1283    snapshot_id: SnapshotId
1284    plan: Plan
1285
1286    @property
1287    def snapshot(self) -> Snapshot:
1288        if self.is_removed:
1289            raise ValueError("Removed snapshots only have SnapshotTableInfo available")
1290        return self.plan.snapshots[self.snapshot_id]
1291
1292    @cached_property
1293    def snapshot_table_info(self) -> SnapshotTableInfo:
1294        if self.is_removed:
1295            return self.plan.modified_snapshots[self.snapshot_id].table_info
1296        return self.plan.snapshots[self.snapshot_id].table_info
1297
1298    @property
1299    def display_name(self) -> str:
1300        dialect = None if self.is_removed else self.snapshot.node.dialect
1301        return self.snapshot_table_info.display_name(
1302            self.plan.environment_naming_info, default_catalog=None, dialect=dialect
1303        )
1304
1305    @property
1306    def change_category(self) -> str:
1307        if self.is_removed:
1308            return SNAPSHOT_CHANGE_CATEGORY_STR[SnapshotChangeCategory.BREAKING]
1309
1310        if change_category := self.snapshot.change_category:
1311            return SNAPSHOT_CHANGE_CATEGORY_STR[change_category]
1312
1313        return "Uncategorized"
1314
1315    @property
1316    def is_added(self) -> bool:
1317        return self.snapshot_id in self.plan.context_diff.added
1318
1319    @property
1320    def is_removed(self) -> bool:
1321        return self.snapshot_id in self.plan.context_diff.removed_snapshots
1322
1323    @property
1324    def is_dev_preview(self) -> bool:
1325        return not self.plan.deployability_index.is_deployable(self.snapshot_id)
1326
1327    @property
1328    def is_directly_modified(self) -> bool:
1329        return self.plan.context_diff.directly_modified(self.snapshot_table_info.name)
1330
1331    @property
1332    def is_indirectly_modified(self) -> bool:
1333        return self.plan.context_diff.indirectly_modified(self.snapshot_table_info.name)
1334
1335    @property
1336    def is_modified(self) -> bool:
1337        return self.is_directly_modified or self.is_indirectly_modified
1338
1339    @property
1340    def is_metadata_updated(self) -> bool:
1341        return self.plan.context_diff.metadata_updated(self.snapshot_table_info.name)
1342
1343    @property
1344    def is_incremental(self) -> bool:
1345        return self.snapshot_table_info.is_incremental
1346
1347    @property
1348    def modification_type(self) -> str:
1349        if self.is_directly_modified:
1350            return "Directly modified"
1351        if self.is_indirectly_modified:
1352            return "Indirectly modified"
1353        if self.is_metadata_updated:
1354            return "Metadata updated"
1355
1356        return "Unknown"
1357
1358    @property
1359    def loaded_intervals(self) -> SnapshotIntervals:
1360        if self.is_removed:
1361            raise ValueError("Removed snapshots dont have loaded intervals available")
1362
1363        return SnapshotIntervals(
1364            snapshot_id=self.snapshot_id,
1365            intervals=(
1366                self.snapshot.dev_intervals
1367                if self.snapshot.is_forward_only
1368                else self.snapshot.intervals
1369            ),
1370        )
1371
1372    @property
1373    def loaded_intervals_rendered(self) -> str:
1374        if self.is_removed:
1375            return "REMOVED"
1376
1377        return self._format_intervals(self.loaded_intervals)
1378
1379    @property
1380    def missing_intervals(self) -> t.Optional[SnapshotIntervals]:
1381        return next(
1382            (si for si in self.plan.missing_intervals if si.snapshot_id == self.snapshot_id),
1383            None,
1384        )
1385
1386    @property
1387    def missing_intervals_formatted(self) -> str:
1388        if not self.is_removed and (intervals := self.missing_intervals):
1389            return self._format_intervals(intervals)
1390
1391        return "N/A"
1392
1393    @property
1394    def as_markdown_list_item(self) -> str:
1395        if self.is_removed:
1396            return f"- `{self.display_name}` ({self.change_category})"
1397
1398        how_applied = ""
1399
1400        if not self.is_incremental:
1401            from sqlmesh.core.console import _format_missing_intervals
1402
1403            # note: this is to re-use the '[recreate view]' and '[full refresh]' text and keep it in sync with updates to the CLI
1404            # it doesnt actually use the passed intervals, those are handled differently
1405            how_applied = _format_missing_intervals(self.snapshot, self.loaded_intervals)
1406
1407        how_applied_str = f" [{how_applied}]" if how_applied else ""
1408
1409        item = f"- `{self.display_name}` ({self.change_category})\n"
1410
1411        if self.snapshot_table_info.model_kind_name:
1412            item += f"  **Kind:** {self.snapshot_table_info.model_kind_name}{how_applied_str}\n"
1413
1414        if self.is_incremental:
1415            # in-depth interval info is only relevant for incremental models
1416            item += f"  **Dates loaded in PR:** [{self.loaded_intervals_rendered}]\n"
1417            if self.missing_intervals:
1418                item += f"  **Dates *not* loaded in PR:** [{self.missing_intervals_formatted}]\n"
1419
1420        return item
1421
1422    def _format_intervals(self, intervals: SnapshotIntervals) -> str:
1423        preview_modifier = " (**preview**)" if self.is_dev_preview else ""
1424        return f"{intervals.format_intervals(self.snapshot.node.interval_unit)}{preview_modifier}"