sqlmesh.core.linter.rule

  1from __future__ import annotations
  2
  3import abc
  4from dataclasses import dataclass, field
  5from pathlib import Path
  6
  7from sqlmesh.core.model import Model
  8
  9from typing import Type
 10
 11import typing as t
 12
 13from sqlmesh.utils.pydantic import PydanticModel
 14
 15
 16if t.TYPE_CHECKING:
 17    from sqlmesh.core.context import GenericContext
 18
 19
 20class RuleLocation(PydanticModel):
 21    """The location of a rule in a file."""
 22
 23    file_path: str
 24    start_line: t.Optional[int] = None
 25
 26
 27@dataclass(frozen=True)
 28class Position:
 29    """The position of a rule violation in a file, the position follows the LSP standard."""
 30
 31    line: int
 32    character: int
 33
 34
 35@dataclass(frozen=True)
 36class Range:
 37    """The range of a rule violation in a file. The range follows the LSP standard."""
 38
 39    start: Position
 40    end: Position
 41
 42
 43@dataclass(frozen=True)
 44class TextEdit:
 45    """A text edit to apply to a file."""
 46
 47    path: Path
 48    range: Range
 49    new_text: str
 50
 51
 52@dataclass(frozen=True)
 53class CreateFile:
 54    """Create a new file with the provided text."""
 55
 56    path: Path
 57    text: str
 58
 59
 60@dataclass(frozen=True)
 61class Fix:
 62    """A fix that can be applied to resolve a rule violation."""
 63
 64    title: str
 65    edits: t.List[TextEdit] = field(default_factory=list)
 66    create_files: t.List[CreateFile] = field(default_factory=list)
 67
 68
 69class _Rule(abc.ABCMeta):
 70    def __new__(cls: Type[_Rule], clsname: str, bases: t.Tuple, attrs: t.Dict) -> _Rule:
 71        attrs["name"] = clsname.lower()
 72        return super().__new__(cls, clsname, bases, attrs)
 73
 74
 75class Rule(abc.ABC, metaclass=_Rule):
 76    """The base class for a rule."""
 77
 78    name = "rule"
 79
 80    def __init__(self, context: GenericContext):
 81        self.context = context
 82
 83    @abc.abstractmethod
 84    def check_model(
 85        self, model: Model
 86    ) -> t.Optional[t.Union[RuleViolation, t.List[RuleViolation]]]:
 87        """The evaluation function that'll check for a violation of this rule."""
 88
 89    @property
 90    def summary(self) -> str:
 91        """A summary of what this rule checks for."""
 92        return self.__doc__ or ""
 93
 94    def violation(
 95        self,
 96        violation_msg: t.Optional[str] = None,
 97        violation_range: t.Optional[Range] = None,
 98        fixes: t.Optional[t.List[Fix]] = None,
 99    ) -> RuleViolation:
100        """Create a RuleViolation instance for this rule"""
101        return RuleViolation(
102            rule=self,
103            violation_msg=violation_msg or self.summary,
104            violation_range=violation_range,
105            fixes=fixes,
106        )
107
108    def get_definition_location(self) -> RuleLocation:
109        """Return the file path and position information for this rule.
110
111        This method returns information about where this rule is defined,
112        which can be used in diagnostics to link to the rule's documentation.
113
114        Returns:
115            A dictionary containing file path and position information.
116        """
117        import inspect
118
119        # Get the file where the rule class is defined
120        file_path = inspect.getfile(self.__class__)
121
122        try:
123            # Get the source code and line number
124            source_lines, start_line = inspect.getsourcelines(self.__class__)
125            return RuleLocation(
126                file_path=file_path,
127                start_line=start_line,
128            )
129        except (IOError, TypeError):
130            # Fall back to just returning the file path if we can't get source lines
131            return RuleLocation(file_path=file_path)
132
133    def __repr__(self) -> str:
134        return self.name
135
136
137class RuleViolation:
138    def __init__(
139        self,
140        rule: Rule,
141        violation_msg: str,
142        violation_range: t.Optional[Range] = None,
143        fixes: t.Optional[t.List[Fix]] = None,
144    ) -> None:
145        self.rule = rule
146        self.violation_msg = violation_msg
147        self.violation_range = violation_range
148        self.fixes = fixes or []
149
150    def __repr__(self) -> str:
151        return f"{self.rule.name}: {self.violation_msg}"