fix: allow subclassing of config again

Author: eakmanrq

sqlmesh/core/config/root.py

@@ -6,7 +6,6 @@
 import zlib
 
 from pydantic import Field
-from pydantic.functional_validators import BeforeValidator
 from sqlglot import exp
 from sqlglot.helper import first
 from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
@@ -44,42 +43,11 @@
 from sqlmesh.core.user import User
 from sqlmesh.utils.date import to_timestamp, now
 from sqlmesh.utils.errors import ConfigError
-from sqlmesh.utils.pydantic import model_validator
-
-
-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)
-
+from sqlmesh.utils.pydantic import model_validator, field_validator
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import Self
 
-    NoPastTTLString = str
-    GatewayDict = t.Dict[str, GatewayConfig]
-    RegexKeyDict = t.Dict[re.Pattern, str]
-else:
-    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)]
-
 
 class Config(BaseConfig):
     """An object used by a Context to configure your SQLMesh project.
@@ -121,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: GatewayDict = {"": GatewayConfig()}
+    gateways: t.Dict[str, GatewayConfig] = {"": GatewayConfig()}
     default_connection: SerializableConnectionConfig = DuckDBConnectionConfig()
     default_test_connection_: t.Optional[SerializableConnectionConfig] = Field(
         default=None, alias="default_test_connection"
@@ -130,8 +98,8 @@ class Config(BaseConfig):
     default_gateway: str = ""
     notification_targets: t.List[NotificationTarget] = []
     project: str = ""
-    snapshot_ttl: NoPastTTLString = c.DEFAULT_SNAPSHOT_TTL
-    environment_ttl: t.Optional[NoPastTTLString] = c.DEFAULT_ENVIRONMENT_TTL
+    snapshot_ttl: str = c.DEFAULT_SNAPSHOT_TTL
+    environment_ttl: t.Optional[str] = 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] = []
@@ -141,12 +109,12 @@ class Config(BaseConfig):
     loader_kwargs: t.Dict[str, t.Any] = {}
     env_vars: t.Dict[str, str] = {}
     username: str = ""
-    physical_schema_mapping: RegexKeyDict = {}
+    physical_schema_mapping: t.Dict[re.Pattern, str] = {}
     environment_suffix_target: EnvironmentSuffixTarget = Field(
         default=EnvironmentSuffixTarget.default
     )
     gateway_managed_virtual_layer: bool = False
-    environment_catalog_mapping: RegexKeyDict = {}
+    environment_catalog_mapping: t.Dict[re.Pattern, str] = {}
     default_target_environment: str = c.PROD
     log_limit: int = c.DEFAULT_LOG_LIMIT
     cicd_bot: t.Optional[CICDBotConfig] = None
@@ -187,6 +155,33 @@ 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)
+
+    @field_validator("snapshot_ttl", "environment_ttl", mode="before")
+    @classmethod
+    def validate_no_past_ttl(cls, 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
+
     @model_validator(mode="before")
     def _normalize_and_validate_fields(cls, data: t.Any) -> t.Any:
         if not isinstance(data, dict):