Feat: Introduce pinned environments which are omitted by the janitor process (#1043)

Author: izeigerman

docs/reference/configuration.md

@@ -144,13 +144,14 @@ See [Airflow Integration Guide](../integrations/airflow.md) for information on h
 | `default_gateway`         | The name of a gateway to use if one is not provided explicitly (Default: the gateway defined first in the `gateways` option)                           | string     | N        |
 
 ## SQLMesh-specific configurations
-| Option                    | Description                                                                                                                                                                                                                                                                                        |         Type         | Required |
+| Option                    | Description                                                                                                                                                                                                                                                                                        | Type                 | Required |
 |---------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:--------------------:|:--------:|
-| `snapshot_ttl`            | The period of time that a model snapshot not a part of any environment should exist before being deleted. This is defined as a string with the default `in 1 week`. Other [relative dates](https://dateparser.readthedocs.io/en/latest/) can be used, such as `in 30 days`. (Default: `in 1 week`) |        string        |    N     |
-| `environment_ttl`         | The period of time that a development environment should exist before being deleted. This is defined as a string with the default `in 1 week`. Other [relative dates](https://dateparser.readthedocs.io/en/latest/) can be used, such as `in 30 days`. (Default: `in 1 week`)                      |        string        |    N     |
-| `ignore_patterns`         | Files that match glob patterns specified in this list are ignored when scanning the project folder (Default: `[]`)                                                                                                                                                                                |     list[string]     |    N     |
-| `time_column_format`      | The default format to use for all model time columns. This time format uses [python format codes](https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes) (Default: `%Y-%m-%d`)                                                                                       |        string        |    N     |
-| `auto_categorize_changes` | Indicates whether SQLMesh should attempt to automatically [categorize](../concepts/plans.md#change-categories) model changes during plan creation per each model source type ([Additional Details](#auto-categorize-changes))                                                                       | dict[string, string] |    N     |
+| `snapshot_ttl`            | The period of time that a model snapshot not a part of any environment should exist before being deleted. This is defined as a string with the default `in 1 week`. Other [relative dates](https://dateparser.readthedocs.io/en/latest/) can be used, such as `in 30 days`. (Default: `in 1 week`) | string               | N        |
+| `environment_ttl`         | The period of time that a development environment should exist before being deleted. This is defined as a string with the default `in 1 week`. Other [relative dates](https://dateparser.readthedocs.io/en/latest/) can be used, such as `in 30 days`. (Default: `in 1 week`)                      | string               | N        |
+| `pinned_environments`     | The list of development environments that are exempt from deletion due to expiration                                                                                                                                                                                                               | list[string]         | N        |
+| `ignore_patterns`         | Files that match glob patterns specified in this list are ignored when scanning the project folder (Default: `[]`)                                                                                                                                                                                 | list[string]         | N        |
+| `time_column_format`      | The default format to use for all model time columns. This time format uses [python format codes](https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes) (Default: `%Y-%m-%d`)                                                                                        | string               | N        |
+| `auto_categorize_changes` | Indicates whether SQLMesh should attempt to automatically [categorize](../concepts/plans.md#change-categories) model changes during plan creation per each model source type ([Additional Details](#auto-categorize-changes))                                                                      | dict[string, string] | N        |
 
 ## Model configuration
 This section contains options that are specific to models, which are set automatically unless explicitly overridden in the model definition.

sqlmesh/core/config/root.py

@@ -37,6 +37,7 @@ class Config(BaseConfig):
         auto_categorize_changes: Indicates whether SQLMesh should attempt to automatically categorize model changes (breaking / non-breaking)
             during plan creation.
         users: A list of users that can be used for approvals/notifications.
+        pinned_environments: A list of development environment names that should not be deleted by the janitor task.
         model_defaults: Default values for model definitions.
     """
 
@@ -54,6 +55,7 @@ class Config(BaseConfig):
     auto_categorize_changes: CategorizerConfig = CategorizerConfig()
     users: t.List[User] = []
     model_defaults: ModelDefaultsConfig = ModelDefaultsConfig()
+    pinned_environments: t.Set[str] = set()
     loader: t.Type[Loader] = SqlMeshLoader
     env_vars: t.Dict[str, str] = {}
 
@@ -64,6 +66,7 @@ class Config(BaseConfig):
         "users": UpdateStrategy.EXTEND,
         "model_defaults": UpdateStrategy.NESTED_UPDATE,
         "auto_categorize_changes": UpdateStrategy.NESTED_UPDATE,
+        "pinned_environments": UpdateStrategy.EXTEND,
     }
 
     @validator("gateways", always=True)

sqlmesh/core/context.py

@@ -233,6 +233,7 @@ def __init__(
         self.gateway = gateway
         self._scheduler = self.config.get_scheduler(self.gateway)
         self.environment_ttl = self.config.environment_ttl
+        self.pinned_environments = Environment.normalize_names(self.config.pinned_environments)
         self.auto_categorize_changes = self.config.auto_categorize_changes
 
         connection_config = self.config.get_connection(self.gateway)
@@ -695,6 +696,10 @@ def plan(
 
         self._run_plan_tests(skip_tests=skip_tests)
 
+        environment_ttl = (
+            self.environment_ttl if environment not in self.pinned_environments else None
+        )
+
         plan = Plan(
             context_diff=self._context_diff(environment or c.PROD, create_from=create_from),
             state_reader=self.state_reader,
@@ -707,7 +712,7 @@ def plan(
             skip_backfill=skip_backfill,
             is_dev=environment != c.PROD,
             forward_only=forward_only,
-            environment_ttl=self.environment_ttl,
+            environment_ttl=environment_ttl,
             categorizer_config=self.auto_categorize_changes,
             auto_categorization_enabled=not no_auto_categorization,
             effective_from=effective_from,

sqlmesh/core/environment.py

@@ -59,3 +59,7 @@ def normalize_name(cls, v: str | Environment) -> str | Environment:
         if not isinstance(v, str):
             raise TypeError(f"Expected str or Environment, got {type(v).__name__}")
         return cls._normalize_name(v)
+
+    @classmethod
+    def normalize_names(cls, values: t.Iterable[str]) -> t.Set[str]:
+        return {cls.normalize_name(value) for value in values}