Chore: Improve error messages returned by pydantic (#4430)

Author: izeigerman

sqlmesh/core/config/connection.py

@@ -25,7 +25,12 @@
 from sqlmesh.core.engine_adapter.shared import CatalogSupport
 from sqlmesh.core.engine_adapter import EngineAdapter
 from sqlmesh.utils.errors import ConfigError
-from sqlmesh.utils.pydantic import ValidationInfo, field_validator, model_validator
+from sqlmesh.utils.pydantic import (
+    ValidationInfo,
+    field_validator,
+    model_validator,
+    validation_error_message,
+)
 from sqlmesh.utils.aws import validate_s3_uri
 
 if t.TYPE_CHECKING:
@@ -1846,7 +1851,13 @@ def _connection_config_validator(
 ) -> ConnectionConfig | None:
     if v is None or isinstance(v, ConnectionConfig):
         return v
-    return parse_connection_config(v)
+    try:
+        return parse_connection_config(v)
+    except pydantic.ValidationError as e:
+        raise ConfigError(
+            validation_error_message(e, f"Invalid '{v['type']}' connection config:")
+            + "\n\nVerify your config.yaml and environment variables."
+        )
 
 
 connection_config_validator: t.Callable = field_validator(

sqlmesh/core/config/loader.py

@@ -5,6 +5,7 @@
 import typing as t
 from pathlib import Path
 
+from pydantic import ValidationError
 from sqlglot.helper import ensure_list
 
 from sqlmesh.core import constants as c
@@ -13,6 +14,7 @@
 from sqlmesh.utils import env_vars, merge_dicts, sys_path
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.metaprogramming import import_python_file
+from sqlmesh.utils.pydantic import validation_error_message
 from sqlmesh.utils.yaml import load as yaml_load
 
 C = t.TypeVar("C", bound=Config)
@@ -99,9 +101,15 @@ def load_config_from_paths(
                 )
             non_python_configs.append(load_config_from_yaml(path))
         elif extension == "py":
-            python_config = load_config_from_python_module(
-                config_type, path, config_name=config_name
-            )
+            try:
+                python_config = load_config_from_python_module(
+                    config_type, path, config_name=config_name
+                )
+            except ValidationError as e:
+                raise ConfigError(
+                    validation_error_message(e, f"Invalid project config '{config_name}':")
+                    + "\n\nVerify your config.py."
+                )
         else:
             raise ConfigError(
                 f"Unsupported config file extension '{extension}' in config file '{path}'."
@@ -126,7 +134,13 @@ def load_config_from_paths(
                 f"'{default}' is not a valid model default configuration key. Please remove it from the `model_defaults` specification in your config file."
             )
 
-    non_python_config = config_type.parse_obj(non_python_config_dict)
+    try:
+        non_python_config = config_type.parse_obj(non_python_config_dict)
+    except ValidationError as e:
+        raise ConfigError(
+            validation_error_message(e, "Invalid project config:")
+            + "\n\nVerify your config.yaml and environment variables."
+        )
 
     no_dialect_err_msg = "Default model SQL dialect is a required configuration parameter. Set it in the `model_defaults` `dialect` key in your config file."
     if python_config:

sqlmesh/core/config/scheduler.py

@@ -3,7 +3,7 @@
 import abc
 import typing as t
 
-from pydantic import Field
+from pydantic import Field, ValidationError
 
 from sqlglot.helper import subclasses
 from sqlmesh.core.config.base import BaseConfig
@@ -16,7 +16,7 @@
 from sqlmesh.core.state_sync import EngineAdapterStateSync, StateSync
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.hashing import md5
-from sqlmesh.utils.pydantic import field_validator
+from sqlmesh.utils.pydantic import field_validator, validation_error_message
 
 if t.TYPE_CHECKING:
     from sqlmesh.core.context import GenericContext
@@ -159,7 +159,13 @@ def _scheduler_config_validator(
     if scheduler_type not in SCHEDULER_CONFIG_TO_TYPE:
         raise ConfigError(f"Unknown scheduler type '{scheduler_type}'.")
 
-    return SCHEDULER_CONFIG_TO_TYPE[scheduler_type](**v)
+    try:
+        return SCHEDULER_CONFIG_TO_TYPE[scheduler_type](**v)
+    except ValidationError as e:
+        raise ConfigError(
+            validation_error_message(e, f"Invalid '{scheduler_type}' scheduler config:")
+            + "\n\nVerify your config.yaml and environment variables."
+        )
 
 
 scheduler_config_validator = field_validator(

sqlmesh/core/loader.py

@@ -11,6 +11,7 @@
 from collections import Counter, defaultdict
 from dataclasses import dataclass
 from pathlib import Path
+from pydantic import ValidationError
 
 from sqlglot.errors import SqlglotError
 from sqlglot import exp
@@ -39,6 +40,7 @@
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.jinja import JinjaMacroRegistry, MacroExtractor
 from sqlmesh.utils.metaprogramming import import_python_file
+from sqlmesh.utils.pydantic import validation_error_message
 from sqlmesh.utils.yaml import YAML, load as yaml_load
 
 
@@ -241,7 +243,7 @@ def _load(path: Path) -> t.List[Model]:
                         for row in YAML().load(file.read())
                     ]
             except Exception as ex:
-                self._raise_failed_to_load_model_error(path, str(ex))
+                self._raise_failed_to_load_model_error(path, ex)
                 raise
 
         for path in paths_to_load:
@@ -372,8 +374,11 @@ def _get_variables(self, gateway_name: t.Optional[str] = None) -> t.Dict[str, t.
 
         return self._variables_by_gateway[gateway_name]
 
-    def _raise_failed_to_load_model_error(self, path: Path, message: str) -> None:
-        raise ConfigError(f"Failed to load model definition at '{path}'.\n{message}")
+    def _raise_failed_to_load_model_error(self, path: Path, error: t.Union[str, Exception]) -> None:
+        base_message = f"Failed to load model definition at '{path}':"
+        if isinstance(error, ValidationError):
+            raise ConfigError(validation_error_message(error, base_message))
+        raise ConfigError(f"{base_message}\n  {error}")
 
 
 class SqlMeshLoader(Loader):
@@ -496,7 +501,7 @@ def _load() -> t.List[Model]:
                         default_catalog_per_gateway=self.context.default_catalog_per_gateway,
                     )
                 except Exception as ex:
-                    self._raise_failed_to_load_model_error(path, str(ex))
+                    self._raise_failed_to_load_model_error(path, ex)
                     raise
 
             for model in cache.get_or_load_models(path, _load):
@@ -561,7 +566,7 @@ def _load_python_models(
                             if model.enabled:
                                 models[model.fqn] = model
                 except Exception as ex:
-                    self._raise_failed_to_load_model_error(path, str(ex))
+                    self._raise_failed_to_load_model_error(path, ex)
 
         finally:
             model_registry._dialect = None

sqlmesh/core/model/common.py

@@ -20,7 +20,7 @@
     prepare_env,
     serialize_env,
 )
-from sqlmesh.utils.pydantic import ValidationInfo, field_validator
+from sqlmesh.utils.pydantic import PydanticModel, ValidationInfo, field_validator
 
 if t.TYPE_CHECKING:
     from sqlglot.dialects.dialect import DialectType
@@ -260,6 +260,22 @@ def get_first_arg(keyword_arg_name: str) -> t.Any:
     return depends_on, used_variables
 
 
+def validate_extra_and_required_fields(
+    klass: t.Type[PydanticModel],
+    provided_fields: t.Set[str],
+    entity_name: str,
+) -> None:
+    missing_required_fields = klass.missing_required_fields(provided_fields)
+    if missing_required_fields:
+        raise_config_error(
+            f"Missing required fields {missing_required_fields} in the {entity_name}"
+        )
+
+    extra_fields = klass.extra_fields(provided_fields)
+    if extra_fields:
+        raise_config_error(f"Invalid extra fields {extra_fields} in the {entity_name}")
+
+
 def single_value_or_tuple(values: t.Sequence) -> exp.Identifier | exp.Tuple:
     return (
         exp.to_identifier(values[0])

sqlmesh/core/model/definition.py

@@ -31,6 +31,7 @@
     parse_dependencies,
     single_value_or_tuple,
     sorted_python_env_payloads,
+    validate_extra_and_required_fields,
 )
 from sqlmesh.core.model.meta import ModelMeta, FunctionCall
 from sqlmesh.core.model.kind import (
@@ -2190,15 +2191,11 @@ def load_sql_based_model(
     seed_properties = {
         p.name.lower(): p.args.get("value") for p in common_kwargs.pop("kind").expressions
     }
-    try:
-        return create_seed_model(
-            name,
-            SeedKind(**seed_properties),
-            **common_kwargs,
-        )
-    except Exception as ex:
-        raise_config_error(str(ex), path)
-        raise
+    return create_seed_model(
+        name,
+        SeedKind(**seed_properties),
+        **common_kwargs,
+    )
 
 
 def create_sql_model(
@@ -2398,7 +2395,9 @@ def _create_model(
     blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
     **kwargs: t.Any,
 ) -> Model:
-    _validate_model_fields(klass, {"name", *kwargs} - {"grain", "table_properties"}, path)
+    validate_extra_and_required_fields(
+        klass, {"name", *kwargs} - {"grain", "table_properties"}, "model definition"
+    )
 
     for prop in PROPERTIES:
         kwargs[prop] = _resolve_properties((defaults or {}).get(prop), kwargs.get(prop))
@@ -2457,21 +2456,17 @@ def _create_model(
     for jinja_macro in jinja_macros.root_macros.values():
         used_variables.update(extract_macro_references_and_variables(jinja_macro.definition)[1])
 
-    try:
-        model = klass(
-            name=name,
-            **{
-                **(defaults or {}),
-                "jinja_macros": jinja_macros or JinjaMacroRegistry(),
-                "dialect": dialect,
-                "depends_on": depends_on,
-                "physical_schema_override": physical_schema_override,
-                **kwargs,
-            },
-        )
-    except Exception as ex:
-        raise_config_error(str(ex), location=path)
-        raise
+    model = klass(
+        name=name,
+        **{
+            **(defaults or {}),
+            "jinja_macros": jinja_macros or JinjaMacroRegistry(),
+            "dialect": dialect,
+            "depends_on": depends_on,
+            "physical_schema_override": physical_schema_override,
+            **kwargs,
+        },
+    )
 
     audit_definitions = {
         **(audit_definitions or {}),
@@ -2636,19 +2631,6 @@ def _resolve_properties(
     return None
 
 
-def _validate_model_fields(klass: t.Type[_Model], provided_fields: t.Set[str], path: Path) -> None:
-    missing_required_fields = klass.missing_required_fields(provided_fields)
-    if missing_required_fields:
-        raise_config_error(
-            f"Missing required fields {missing_required_fields} in the model definition",
-            path,
-        )
-
-    extra_fields = klass.extra_fields(provided_fields)
-    if extra_fields:
-        raise_config_error(f"Invalid extra fields {extra_fields} in the model definition", path)
-
-
 def _list_of_calls_to_exp(value: t.List[t.Tuple[str, t.Dict[str, t.Any]]]) -> exp.Expression:
     return exp.Tuple(
         expressions=[

sqlmesh/core/model/kind.py

@@ -12,7 +12,11 @@
 from sqlglot.time import format_time
 
 from sqlmesh.core import dialect as d
-from sqlmesh.core.model.common import parse_properties, properties_validator
+from sqlmesh.core.model.common import (
+    parse_properties,
+    properties_validator,
+    validate_extra_and_required_fields,
+)
 from sqlmesh.core.model.seed import CsvSettings
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import (
@@ -1011,6 +1015,7 @@ def create_model_kind(v: t.Any, dialect: str, defaults: t.Dict[str, t.Any]) -> M
                 actual_kind_type, _ = custom_materialization
                 return actual_kind_type(**props)
 
+        validate_extra_and_required_fields(kind_type, set(props), f"model kind '{name}'")
         return kind_type(**props)
 
     name = (v.name if isinstance(v, exp.Expression) else str(v)).upper()

sqlmesh/core/model/meta.py

@@ -355,7 +355,7 @@ def _root_validator(self) -> Self:
                 and not (kind.is_view and kind.materialized)
             ):
                 name = field[:-1] if field.endswith("_") else field
-                raise ValueError(f"{name} field cannot be set for {kind} models")
+                raise ValueError(f"{name} field cannot be set for {kind.name} models")
         if kind.is_incremental_by_partition and not getattr(self, "partitioned_by_", None):
             raise ValueError(f"partitioned_by field is required for {kind.name} models")
 

sqlmesh/utils/pydantic.py

@@ -211,6 +211,20 @@ def positive_int_validator(v: t.Any) -> int:
     return v
 
 
+def validation_error_message(error: pydantic.ValidationError, base: str) -> str:
+    errors = "\n  ".join(_formatted_validation_errors(error))
+    return f"{base}\n  {errors}"
+
+
+def _formatted_validation_errors(error: pydantic.ValidationError) -> t.List[str]:
+    result = []
+    for e in error.errors():
+        msg = e["msg"]
+        loc: t.Optional[t.Tuple] = e.get("loc")
+        result.append(f"Invalid field '{loc[0]}':\n    {msg}" if loc else msg)
+    return result
+
+
 def _get_field(
     v: t.Any,
     values: t.Any,
@@ -281,6 +295,9 @@ def cron_validator(v: t.Any) -> str:
 
     from croniter import CroniterBadCronError, croniter
 
+    if not isinstance(v, str):
+        raise ValueError(f"Invalid cron expression '{v}'. Value must be a string.")
+
     try:
         croniter(v)
     except CroniterBadCronError:

tests/core/test_connection_config.py

@@ -973,13 +973,13 @@ def test_databricks(make_config):
     assert oauth_u2m_config.oauth_client_secret is None
 
     # auth_type must match the AuthType enum if specified
-    with pytest.raises(ValueError, match=r".*nonexist does not match a valid option.*"):
+    with pytest.raises(ConfigError, match=r".*nonexist does not match a valid option.*"):
         make_config(
             type="databricks", server_hostname="dbc-test.cloud.databricks.com", auth_type="nonexist"
         )
 
     # if client_secret is specified, client_id must also be specified
-    with pytest.raises(ValueError, match=r"`oauth_client_id` is required.*"):
+    with pytest.raises(ConfigError, match=r"`oauth_client_id` is required.*"):
         make_config(
             type="databricks",
             server_hostname="dbc-test.cloud.databricks.com",
@@ -988,7 +988,7 @@ def test_databricks(make_config):
         )
 
     # http_path is still required when auth_type is specified
-    with pytest.raises(ValueError, match=r"`http_path` is still required.*"):
+    with pytest.raises(ConfigError, match=r"`http_path` is still required.*"):
         make_config(
             type="databricks",
             server_hostname="dbc-test.cloud.databricks.com",