feat(lsp): return load time errors as diagnostics

- adds path in places to make this as possible with
- refreshes well on typing

Author: benfdking

sqlmesh/core/loader.py

@@ -349,7 +349,7 @@ def _load(path: Path) -> t.List[Model]:
                         for row in YAML().load(file.read())
                     ]
             except Exception as ex:
-                raise ConfigError(self._failed_to_load_model_error(path, ex))
+                raise ConfigError(self._failed_to_load_model_error(path, ex), path)
 
         for path in paths_to_load:
             self._track_file(path)
@@ -363,7 +363,8 @@ def _load(path: Path) -> t.List[Model]:
                         raise ConfigError(
                             self._failed_to_load_model_error(
                                 path, f"Duplicate external model name: '{model.name}'."
-                            )
+                            ),
+                            path,
                         )
                     models[model.fqn] = model
 
@@ -375,7 +376,8 @@ def _load(path: Path) -> t.List[Model]:
                             raise ConfigError(
                                 self._failed_to_load_model_error(
                                     path, f"Duplicate external model name: '{model.name}'."
-                                )
+                                ),
+                                path,
                             )
                         models.update({model.fqn: model})
 
@@ -402,13 +404,15 @@ def _load_requirements(self) -> t.Tuple[t.Dict[str, str], t.Set[str]]:
                     args = [k.strip() for k in line.split("==")]
                     if len(args) != 2:
                         raise ConfigError(
-                            f"Invalid lock file entry '{line.strip()}'. Only 'dep==ver' is supported"
+                            f"Invalid lock file entry '{line.strip()}'. Only 'dep==ver' is supported",
+                            requirements_path,
                         )
                     dep, ver = args
                     other_ver = requirements.get(dep, ver)
                     if ver != other_ver:
                         raise ConfigError(
-                            f"Conflicting requirement {dep}: {ver} != {other_ver}. Fix your {c.REQUIREMENTS} file."
+                            f"Conflicting requirement {dep}: {ver} != {other_ver}. Fix your {c.REQUIREMENTS} file.",
+                            requirements_path,
                         )
                     requirements[dep] = ver
 
@@ -619,13 +623,14 @@ def _load_sql_models(
                                 raise ConfigError(
                                     self._failed_to_load_model_error(
                                         path, f"Duplicate SQL model name: '{model.name}'."
-                                    )
+                                    ),
+                                    path,
                                 )
                             elif model.enabled:
                                 model._path = path
                                 models[model.fqn] = model
                     except Exception as ex:
-                        raise ConfigError(self._failed_to_load_model_error(path, ex))
+                        raise ConfigError(self._failed_to_load_model_error(path, ex), path)
 
         return models
 
@@ -678,7 +683,7 @@ def _load_python_models(
                             if model.enabled:
                                 models[model.fqn] = model
                 except Exception as ex:
-                    raise ConfigError(self._failed_to_load_model_error(path, ex))
+                    raise ConfigError(self._failed_to_load_model_error(path, ex), path)
 
         finally:
             model_registry._dialect = None
@@ -782,7 +787,9 @@ def _load_metrics(self) -> UniqueKeyDict[str, MetricMeta]:
                         metric = load_metric_ddl(expression, path=path, dialect=dialect)
                         metrics[metric.name] = metric
                 except SqlglotError as ex:
-                    raise ConfigError(f"Failed to parse metric definitions at '{path}': {ex}.")
+                    raise ConfigError(
+                        f"Failed to parse metric definitions at '{path}': {ex}.", path
+                    )
 
         return metrics
 
@@ -1005,7 +1012,7 @@ def _load_scripts(self) -> t.Tuple[MacroRegistry, JinjaMacroRegistry]:
                         package=package,
                     )
                 except Exception as e:
-                    raise ConfigError(f"Failed to load macro file: {path}", e)
+                    raise ConfigError(f"Failed to load macro file:  {e}", path)
 
         self._macros_max_mtime = macros_max_mtime
 

sqlmesh/core/model/common.py

@@ -41,7 +41,7 @@ def make_python_env(
     macros: MacroRegistry,
     variables: t.Optional[t.Dict[str, t.Any]] = None,
     used_variables: t.Optional[t.Set[str]] = None,
-    path: t.Optional[str | Path] = None,
+    path: t.Optional[Path] = None,
     python_env: t.Optional[t.Dict[str, Executable]] = None,
     strict_resolution: bool = True,
     blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
@@ -265,12 +265,14 @@ def validate_extra_and_required_fields(
     klass: t.Type[PydanticModel],
     provided_fields: t.Set[str],
     entity_name: str,
+    path: t.Optional[Path] = None,
 ) -> None:
     missing_required_fields = klass.missing_required_fields(provided_fields)
     if missing_required_fields:
         field_names = "'" + "', '".join(missing_required_fields) + "'"
         raise_config_error(
-            f"Please add required field{'s' if len(missing_required_fields) > 1 else ''} {field_names} to the {entity_name}."
+            f"Please add required field{'s' if len(missing_required_fields) > 1 else ''} {field_names} to the {entity_name}.",
+            path,
         )
 
     extra_fields = klass.extra_fields(provided_fields)
@@ -293,7 +295,8 @@ def validate_extra_and_required_fields(
             similar_msg = "\n\n  " + "\n  ".join(similar) if similar else ""
 
         raise_config_error(
-            f"Invalid field name{'s' if len(extra_fields) > 1 else ''} present in the {entity_name}: {extra_field_names}{similar_msg}"
+            f"Invalid field name{'s' if len(extra_fields) > 1 else ''} present in the {entity_name}: {extra_field_names}{similar_msg}",
+            path,
         )
 
 

sqlmesh/core/model/definition.py

@@ -2434,7 +2434,10 @@ def _create_model(
     **kwargs: t.Any,
 ) -> Model:
     validate_extra_and_required_fields(
-        klass, {"name", *kwargs} - {"grain", "table_properties"}, "MODEL block"
+        klass,
+        {"name", *kwargs} - {"grain", "table_properties"},
+        "MODEL block",
+        path,
     )
 
     for prop in PROPERTIES:

sqlmesh/lsp/context.py

@@ -1,6 +1,5 @@
 from dataclasses import dataclass
 from pathlib import Path
-import uuid
 from sqlmesh.core.context import Context
 import typing as t
 
@@ -35,14 +34,12 @@ class LSPContext:
     map: t.Dict[Path, t.Union[ModelTarget, AuditTarget]]
     _render_cache: t.Dict[Path, t.List[RenderModelEntry]]
     _lint_cache: t.Dict[Path, t.List[AnnotatedRuleViolation]]
-    _version_id: str
     """
     This is a version ID for the context. It is used to track changes to the context. It can be used to 
     return a version number to the LSP client.
     """
 
     def __init__(self, context: Context) -> None:
-        self._version_id = str(uuid.uuid4())
         self.context = context
         self._render_cache = {}
         self._lint_cache = {}
@@ -70,11 +67,6 @@ def __init__(self, context: Context) -> None:
             **audit_map,
         }
 
-    @property
-    def version_id(self) -> str:
-        """Get the version ID for the context."""
-        return self._version_id
-
     def render_model(self, uri: URI) -> t.List[RenderModelEntry]:
         """Get rendered models for a file, using cache when available.
 

sqlmesh/lsp/errors.py

@@ -0,0 +1,51 @@
+from lsprotocol.types import Diagnostic, DiagnosticSeverity, Range, Position
+
+from sqlmesh.lsp.uri import URI
+from sqlmesh.utils.errors import (
+    ConfigError,
+)
+import typing as t
+
+type ContextFailedError = str | ConfigError | Exception
+
+
+def context_error_to_diagnostic(
+    error: t.Union[Exception, ContextFailedError],
+    uri_filter: t.Optional[URI] = None,
+) -> t.Tuple[t.Optional[t.Tuple[str, Diagnostic]], ContextFailedError]:
+    """
+    Convert an error to a diagnostic message.
+    If the error is a ConfigError, it will be converted to a diagnostic message.
+
+    uri_filter is used to filter diagnostics by URI. If present, only diagnostics
+    with a matching URI will be returned.
+    """
+    if isinstance(error, ConfigError):
+        return config_error_to_diagnostic(error), error
+    return None, str(error)
+
+
+def config_error_to_diagnostic(
+    error: ConfigError,
+    uri_filter: t.Optional[URI] = None,
+) -> t.Optional[t.Tuple[str, Diagnostic]]:
+    if error.location is None:
+        return None
+    uri = URI.from_path(error.location).value
+    if uri_filter and uri != uri_filter.value:
+        return None
+    return uri, Diagnostic(
+        range=Range(
+            start=Position(
+                line=0,
+                character=0,
+            ),
+            end=Position(
+                line=0,
+                character=0,
+            ),
+        ),
+        message=str(error),
+        severity=DiagnosticSeverity.Error,
+        source="SQLMesh",
+    )