sqlmesh.integrations.github.cicd.config

 1import typing as t
 2from enum import Enum
 3
 4from pydantic import Field
 5
 6from sqlmesh.core.config import CategorizerConfig
 7from sqlmesh.core.config.base import BaseConfig
 8from sqlmesh.utils.date import TimeLike
 9from sqlmesh.utils.pydantic import model_validator
10from sqlmesh.core.console import get_console
11
12
13class MergeMethod(str, Enum):
14    MERGE = "merge"
15    SQUASH = "squash"
16    REBASE = "rebase"
17
18
19class GithubCICDBotConfig(BaseConfig):
20    type_: t.Literal["github"] = Field(alias="type", default="github")
21
22    invalidate_environment_after_deploy: bool = True
23    enable_deploy_command: bool = False
24    merge_method: t.Optional[MergeMethod] = None
25    command_namespace: t.Optional[str] = None
26    auto_categorize_changes_: t.Optional[CategorizerConfig] = Field(
27        default=None, alias="auto_categorize_changes"
28    )
29    default_pr_start: t.Optional[TimeLike] = None
30    skip_pr_backfill_: t.Optional[bool] = Field(default=None, alias="skip_pr_backfill")
31    pr_include_unmodified_: t.Optional[bool] = Field(default=None, alias="pr_include_unmodified")
32    run_on_deploy_to_prod: bool = False
33    pr_environment_name: t.Optional[str] = None
34    pr_min_intervals: t.Optional[int] = None
35    prod_branch_names_: t.Optional[str] = Field(default=None, alias="prod_branch_name")
36    forward_only_branch_suffix_: t.Optional[str] = Field(
37        default=None, alias="forward_only_branch_suffix"
38    )
39    check_if_blocked_on_deploy_to_prod: bool = True
40
41    @model_validator(mode="before")
42    @classmethod
43    def _validate(cls, data: t.Any) -> t.Any:
44        if not isinstance(data, dict):
45            return data
46
47        if data.get("enable_deploy_command") and not data.get("merge_method"):
48            raise ValueError("merge_method must be set if enable_deploy_command is True")
49        if data.get("command_namespace") and not data.get("enable_deploy_command"):
50            raise ValueError("enable_deploy_command must be set if command_namespace is set")
51
52        return data
53
54    @property
55    def prod_branch_names(self) -> t.List[str]:
56        if self.prod_branch_names_:
57            return [self.prod_branch_names_]
58        return ["main", "master"]
59
60    @property
61    def auto_categorize_changes(self) -> CategorizerConfig:
62        return self.auto_categorize_changes_ or CategorizerConfig.all_off()
63
64    @property
65    def pr_include_unmodified(self) -> bool:
66        return self.pr_include_unmodified_ or False
67
68    @property
69    def skip_pr_backfill(self) -> bool:
70        if self.skip_pr_backfill_ is None:
71            get_console().log_warning(
72                "`skip_pr_backfill` is unset, defaulting it to `true` (no data will be backfilled).\n"
73                "Future versions of SQLMesh will default to `skip_pr_backfill: false` to align with the CLI default behaviour.\n"
74                "If you would like to preserve the current behaviour and remove this warning, please explicitly set `skip_pr_backfill: true` in the bot config.\n\n"
75                "For more information on configuring the bot, see: https://sqlmesh.readthedocs.io/en/stable/integrations/github/"
76            )
77            return True
78        return self.skip_pr_backfill_
79
80    @property
81    def forward_only_branch_suffix(self) -> str:
82        return self.forward_only_branch_suffix_ or "-forward-only"
83
84    FIELDS_FOR_ANALYTICS: t.ClassVar[t.Set[str]] = {
85        "invalidate_environment_after_deploy",
86        "enable_deploy_command",
87        "merge_method",
88        "command_namespace",
89        "auto_categorize_changes",
90        "default_pr_start",
91        "skip_pr_backfill",
92        "pr_include_unmodified",
93        "run_on_deploy_to_prod",
94        "pr_min_intervals",
95        "forward_only_branch_suffix",
96    }