fix: prevent past ttl values for environment and snapshot (#4158)

Author: eakmanrq

sqlmesh/core/config/root.py

@@ -41,6 +41,7 @@
 from sqlmesh.core.loader import Loader, SqlMeshLoader
 from sqlmesh.core.notification_target import NotificationTarget
 from sqlmesh.core.user import User
+from sqlmesh.utils.date import to_timestamp, now, now_timestamp
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import field_validator, model_validator
 
@@ -88,7 +89,7 @@ class Config(BaseConfig):
         after_all: SQL statements or macros to be executed at the end of the `sqlmesh plan` and `sqlmesh run` commands.
     """
 
-    gateways: t.Dict[str, GatewayConfig] = {"": GatewayConfig()}
+    gateways: GatewayDict = {"": GatewayConfig()}
     default_connection: SerializableConnectionConfig = DuckDBConnectionConfig()
     default_test_connection_: t.Optional[SerializableConnectionConfig] = Field(
         default=None, alias="default_test_connection"
@@ -97,8 +98,8 @@ class Config(BaseConfig):
     default_gateway: str = ""
     notification_targets: t.List[NotificationTarget] = []
     project: str = ""
-    snapshot_ttl: str = c.DEFAULT_SNAPSHOT_TTL
-    environment_ttl: t.Optional[str] = c.DEFAULT_ENVIRONMENT_TTL
+    snapshot_ttl: NoPastTTLString = c.DEFAULT_SNAPSHOT_TTL
+    environment_ttl: t.Optional[NoPastTTLString] = c.DEFAULT_ENVIRONMENT_TTL
     ignore_patterns: t.List[str] = c.IGNORE_PATTERNS
     time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT
     users: t.List[User] = []
@@ -108,12 +109,12 @@ class Config(BaseConfig):
     loader_kwargs: t.Dict[str, t.Any] = {}
     env_vars: t.Dict[str, str] = {}
     username: str = ""
-    physical_schema_mapping: t.Dict[re.Pattern, str] = {}
+    physical_schema_mapping: RegexKeyDict = {}
     environment_suffix_target: EnvironmentSuffixTarget = Field(
         default=EnvironmentSuffixTarget.default
     )
     gateway_managed_virtual_layer: bool = False
-    environment_catalog_mapping: t.Dict[re.Pattern, str] = {}
+    environment_catalog_mapping: RegexKeyDict = {}
     default_target_environment: str = c.PROD
     log_limit: int = c.DEFAULT_LOG_LIMIT
     cicd_bot: t.Optional[CICDBotConfig] = None
@@ -154,23 +155,6 @@ class Config(BaseConfig):
     _scheduler_config_validator = scheduler_config_validator  # type: ignore
     _variables_validator = variables_validator
 
-    @field_validator("gateways", mode="before")
-    @classmethod
-    def _gateways_ensure_dict(cls, value: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
-        try:
-            if not isinstance(value, GatewayConfig):
-                GatewayConfig.parse_obj(value)
-            return {"": value}
-        except Exception:
-            return value
-
-    @field_validator("environment_catalog_mapping", "physical_schema_mapping", mode="before")
-    @classmethod
-    def _validate_regex_keys(
-        cls, value: t.Dict[str | re.Pattern, t.Any]
-    ) -> t.Dict[re.Pattern, t.Any]:
-        return compile_regex_mapping(value)
-
     @model_validator(mode="before")
     def _normalize_and_validate_fields(cls, data: t.Any) -> t.Any:
         if not isinstance(data, dict):
@@ -302,3 +286,37 @@ def dialect(self) -> t.Optional[str]:
     @property
     def fingerprint(self) -> str:
         return str(zlib.crc32(pickle.dumps(self.dict(exclude={"loader", "notification_targets"}))))
+
+
+def validate_no_past_ttl(v: str) -> str:
+    current_time = now()
+    if to_timestamp(v, relative_base=current_time) < to_timestamp(current_time):
+        raise ValueError(
+            f"TTL '{v}' is in the past. Please specify a relative time in the future. Ex: `in 1 week` instead of `1 week`."
+        )
+    return v
+
+
+def gateways_ensure_dict(value: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
+    try:
+        if not isinstance(value, GatewayConfig):
+            GatewayConfig.parse_obj(value)
+        return {"": value}
+    except Exception:
+        return value
+
+
+def validate_regex_key_dict(value: t.Dict[str | re.Pattern, t.Any]) -> t.Dict[re.Pattern, t.Any]:
+    return compile_regex_mapping(value)
+
+
+if t.TYPE_CHECKING:
+    NoPastTTLString = str
+    GatewayDict = t.Dict[str, GatewayConfig]
+    RegexKeyDict = t.Dict[re.Pattern, str]
+else:
+    from pydantic.functional_validators import BeforeValidator
+
+    NoPastTTLString = t.Annotated[str, BeforeValidator(validate_no_past_ttl)]
+    GatewayDict = t.Annotated[t.Dict[str, GatewayConfig], BeforeValidator(gateways_ensure_dict)]
+    RegexKeyDict = t.Annotated[t.Dict[re.Pattern, str], BeforeValidator(validate_regex_key_dict)]

tests/integrations/github/cicd/test_config.py

@@ -200,3 +200,48 @@ def test_validation(tmp_path):
         ValueError, match="merge_method must be set if enable_deploy_command is True"
     ):
         load_config_from_paths(Config, project_paths=[tmp_path / "config.yaml"])
+
+
+def test_ttl_in_past(tmp_path):
+    create_temp_file(
+        tmp_path,
+        pathlib.Path("config.yaml"),
+        """
+environment_ttl: in 1 week
+model_defaults:
+    dialect: duckdb
+""",
+    )
+
+    config = load_config_from_paths(Config, project_paths=[tmp_path / "config.yaml"])
+    assert config.environment_ttl == "in 1 week"
+
+    create_temp_file(
+        tmp_path,
+        pathlib.Path("config.yaml"),
+        """
+environment_ttl: 1 week
+model_defaults:
+    dialect: duckdb
+""",
+    )
+    with pytest.raises(
+        ValueError,
+        match="TTL '1 week' is in the past. Please specify a relative time in the future. Ex: `in 1 week` instead of `1 week`.",
+    ):
+        load_config_from_paths(Config, project_paths=[tmp_path / "config.yaml"])
+
+        create_temp_file(
+            tmp_path,
+            pathlib.Path("config.yaml"),
+            """
+snapshot_ttl: 1 week
+model_defaults:
+    dialect: duckdb
+    """,
+        )
+        with pytest.raises(
+            ValueError,
+            match="TTL '1 week' is in the past. Please specify a relative time in the future. Ex: `in 1 week` instead of `1 week`.",
+        ):
+            load_config_from_paths(Config, project_paths=[tmp_path / "config.yaml"])