|
41 | 41 | from sqlmesh.core.loader import Loader, SqlMeshLoader |
42 | 42 | from sqlmesh.core.notification_target import NotificationTarget |
43 | 43 | from sqlmesh.core.user import User |
| 44 | +from sqlmesh.utils.date import to_timestamp, now, now_timestamp |
44 | 45 | from sqlmesh.utils.errors import ConfigError |
45 | 46 | from sqlmesh.utils.pydantic import field_validator, model_validator |
46 | 47 |
|
@@ -97,8 +98,8 @@ class Config(BaseConfig): |
97 | 98 | default_gateway: str = "" |
98 | 99 | notification_targets: t.List[NotificationTarget] = [] |
99 | 100 | project: str = "" |
100 | | - snapshot_ttl: str = c.DEFAULT_SNAPSHOT_TTL |
101 | | - environment_ttl: t.Optional[str] = c.DEFAULT_ENVIRONMENT_TTL |
| 101 | + snapshot_ttl: NoPastTTLString = c.DEFAULT_SNAPSHOT_TTL |
| 102 | + environment_ttl: t.Optional[NoPastTTLString] = c.DEFAULT_ENVIRONMENT_TTL |
102 | 103 | ignore_patterns: t.List[str] = c.IGNORE_PATTERNS |
103 | 104 | time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT |
104 | 105 | users: t.List[User] = [] |
@@ -302,3 +303,20 @@ def dialect(self) -> t.Optional[str]: |
302 | 303 | @property |
303 | 304 | def fingerprint(self) -> str: |
304 | 305 | return str(zlib.crc32(pickle.dumps(self.dict(exclude={"loader", "notification_targets"})))) |
| 306 | + |
| 307 | + |
| 308 | +def validate_no_past_ttl(v: str) -> str: |
| 309 | + current_time = now() |
| 310 | + if to_timestamp(v, relative_base=current_time) < to_timestamp(current_time): |
| 311 | + raise ValueError( |
| 312 | + f"TTL '{v}' is in the past. Please specify a relative time in the future. Ex: `in 1 week` instead of `1 week`." |
| 313 | + ) |
| 314 | + return v |
| 315 | + |
| 316 | + |
| 317 | +if t.TYPE_CHECKING: |
| 318 | + NoPastTTLString = str |
| 319 | +else: |
| 320 | + from pydantic.functional_validators import BeforeValidator |
| 321 | + |
| 322 | + NoPastTTLString = t.Annotated[str, BeforeValidator(validate_no_past_ttl)] |
0 commit comments