Allow rules to be skipped

docs/configuration.md

@@ -86,6 +86,9 @@ Every rule can be configured with the following option:
 - `severity`: The severity of the rule. Rules have a default severity and can be
   overridden. It's an integer with a minimum value of 1 and a maximum value
   of 4.
+- `model_filter_names`: Filters used by the rule. Takes a list of names that can
+  be found in the same namespace as the rules (see
+  [Package rules](package_rules.md))
 
 Some rules have additional configuration options, e.g.
 [sql_has_reasonable_number_of_lines](rules/generic.md#sql_has_reasonable_number_of_lines).

docs/create_rules.md

@@ -90,3 +90,60 @@ def sql_has_reasonable_number_of_lines(model: Model, max_lines: int = 200) -> Ru
             message=f"SQL query too long: {count_lines} lines (> {max_lines})."
         )
 ```
+
+### Filtering
+
+Custom and standard rules can be configured to have model filters. Filters
+allows setting models of the project to be ignored by a given rule.
+
+Custom rules can skip models "manually":
+
+```python
+from dbt_score import rule, RuleViolation, SkipRule
+
+@rule
+def models_in_x_follow_naming_standard(model: Model) -> RuleViolation | SkipRule | None:
+    if model.schema.lower() != 'x':
+        return SkipRule()
+    if some_regex_fails(model.name):
+        return RuleViolation("Invalid model name.")
+```
+
+Filters can also be set as a first-class citizen. Doing so allow re-use and
+composability, as well as being used by generic rules. First, filters should be
+created using the same discovery mechanism and interface as custom rules, except
+they do not accept parameters. Similar to Python's built-in `filter` function,
+when the filter returns `True` the model should be evaluated, otherwise it
+should be skipped.
+
+```python
+from dbt_score import ModelFilter, model_filter
+
+@model_filter
+def only_schema_x(model: Model) -> bool:
+    """Only applies a rule to schema X."""
+    return model.schema.lower() == 'x'
+
+class SkipSchemaY(ModelFilter):
+    description = "Only applies a rule to every schema but Y."
+    def evaluate(self, model: Model) -> bool:
+      return model.schema.lower() != 'y'
+```
+
+Standard rules can have filters set in the
+[configuration file](configuration.md/#tooldbt-scorerulesrule_namespacerule_name).
+For custom rules, filters are instantiated at the rule definition:
+
+```python
+from dbt_score import Model, rule, RuleViolation, SkipRule
+
+# the filter must be in the namespace,
+# so either define it in the same file
+# or import it.
+
+@rule(model_filters={only_schema_x()})
+def models_in_x_follow_naming_standard(model: Model) -> RuleViolation | SkipRule | None:
+    """Models in schema X must follow the naming standard."""
+    if some_regex_fails(model.name):
+        return RuleViolation("Invalid model name.")
+```

src/dbt_score/__init__.py

@@ -1,6 +1,16 @@
 """Init dbt_score package."""
 
+from dbt_score.model_filter import ModelFilter, model_filter
 from dbt_score.models import Model
-from dbt_score.rule import Rule, RuleViolation, Severity, rule
+from dbt_score.rule import Rule, RuleViolation, Severity, SkipRule, rule
 
-__all__ = ["Model", "Rule", "RuleViolation", "Severity", "rule"]
+__all__ = [
+    "Model",
+    "ModelFilter",
+    "Rule",
+    "RuleViolation",
+    "Severity",
+    "SkipRule",
+    "model_filter",
+    "rule",
+]

src/dbt_score/evaluation.py

@@ -6,15 +6,16 @@
 
 from dbt_score.formatters import Formatter
 from dbt_score.models import ManifestLoader, Model
-from dbt_score.rule import Rule, RuleViolation
+from dbt_score.rule import Rule, RuleViolation, SkipRule
 from dbt_score.rule_registry import RuleRegistry
 from dbt_score.scoring import Score, Scorer
 
 # The results of a given model are stored in a dictionary, mapping rules to either:
 # - None if there was no issue
 # - A RuleViolation if a linting error was found
+# - A SkipRule if the rule doesn't apply to the model
 # - An Exception if the rule failed to run
-ModelResultsType = dict[Type[Rule], None | RuleViolation | Exception]
+ModelResultsType = dict[Type[Rule], None | RuleViolation | SkipRule | Exception]
 
 
 class Evaluation:
@@ -57,7 +58,11 @@ def evaluate(self) -> None:
             self.results[model] = {}
             for rule in rules:
                 try:
-                    result: RuleViolation | None = rule.evaluate(model, **rule.config)
+                    result: RuleViolation | SkipRule | None = None
+                    if rule.should_evaluate(model):
+                        result = rule.evaluate(model, **rule.config)
+                    else:
+                        result = SkipRule()
                 except Exception as e:
                     self.results[model][rule.__class__] = e
                 else:

src/dbt_score/formatters/human_readable_formatter.py

@@ -6,7 +6,7 @@
 from dbt_score.evaluation import ModelResultsType
 from dbt_score.formatters import Formatter
 from dbt_score.models import Model
-from dbt_score.rule import RuleViolation
+from dbt_score.rule import RuleViolation, SkipRule
 from dbt_score.scoring import Score
 
 
@@ -40,6 +40,8 @@ def model_evaluated(
         for rule, result in results.items():
             if result is None:
                 print(f"{self.indent}{self.label_ok} {rule.source()}")
+            elif isinstance(result, SkipRule):
+                continue
             elif isinstance(result, RuleViolation):
                 print(
                     f"{self.indent}{self.label_warning} "

src/dbt_score/model_filter.py

@@ -0,0 +1,115 @@
+"""Model filtering to choose when to apply specific rules."""
+
+from typing import Any, Callable, Type, TypeAlias, overload
+
+from dbt_score.models import Model
+
+FilterEvaluationType: TypeAlias = Callable[[Model], bool]
+
+
+class ModelFilter:
+    """The Filter base class."""
+
+    description: str
+
+    def __init__(self) -> None:
+        """Initialize the filter."""
+        pass
+
+    def __init_subclass__(cls, **kwargs) -> None:  # type: ignore
+        """Initializes the subclass."""
+        super().__init_subclass__(**kwargs)
+        if not hasattr(cls, "description"):
+            raise AttributeError("Subclass must define class attribute `description`.")
+
+    def evaluate(self, model: Model) -> bool:
+        """Evaluates the filter."""
+        raise NotImplementedError("Subclass must implement method `evaluate`.")
+
+    @classmethod
+    def source(cls) -> str:
+        """Return the source of the filter, i.e. a fully qualified name."""
+        return f"{cls.__module__}.{cls.__name__}"
+
+    def __hash__(self) -> int:
+        """Compute a unique hash for a filter."""
+        return hash(self.source())
+
+
+# Use @overload to have proper typing for both @model_filter and @model_filter(...)
+# https://mypy.readthedocs.io/en/stable/generics.html#decorator-factories
+
+
+@overload
+def model_filter(__func: FilterEvaluationType) -> Type[ModelFilter]:
+    ...
+
+
+@overload
+def model_filter(
+    *,
+    description: str | None = None,
+) -> Callable[[FilterEvaluationType], Type[ModelFilter]]:
+    ...
+
+
+def model_filter(
+    __func: FilterEvaluationType | None = None,
+    *,
+    description: str | None = None,
+) -> Type[ModelFilter] | Callable[[FilterEvaluationType], Type[ModelFilter]]:
+    """Model-filter decorator.
+
+    The model-filter decorator creates a filter class (subclass of ModelFilter)
+    and returns it.
+
+    Using arguments or not are both supported:
+    - ``@model_filter``
+    - ``@model_filter(description="...")``
+
+    Args:
+        __func: The filter evaluation function being decorated.
+        description: The description of the filter.
+    """
+
+    def decorator_filter(
+        func: FilterEvaluationType,
+    ) -> Type[ModelFilter]:
+        """Decorator function."""
+        if func.__doc__ is None and description is None:
+            raise AttributeError(
+                "ModelFilter must define `description` or `func.__doc__`."
+            )
+
+        # Get description parameter, otherwise use the docstring
+        filter_description = description or (
+            func.__doc__.split("\n")[0] if func.__doc__ else None
+        )
+
+        def wrapped_func(self: ModelFilter, *args: Any, **kwargs: Any) -> bool:
+            """Wrap func to add `self`."""
+            return func(*args, **kwargs)
+
+        # Create the filter class inheriting from Skip
+        filter_class = type(
+            func.__name__,
+            (ModelFilter,),
+            {
+                "description": filter_description,
+                "evaluate": wrapped_func,
+                # Save provided evaluate function
+                "_orig_evaluate": func,
+                # Forward origin of the decorated function
+                "__qualname__": func.__qualname__,  # https://peps.python.org/pep-3155/
+                "__module__": func.__module__,
+            },
+        )
+
+        return filter_class
+
+    if __func is not None:
+        # The syntax @model_filter is used
+        return decorator_filter(__func)
+    else:
+        # The syntax @model_filter(...) is used
+        return decorator_filter

src/dbt_score/rule.py

@@ -4,8 +4,9 @@
 import typing
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, Callable, Type, TypeAlias, overload
+from typing import Any, Callable, Iterable, Type, TypeAlias, overload
 
+from dbt_score.model_filter import ModelFilter
 from dbt_score.models import Model
 
 
@@ -45,14 +46,22 @@ class RuleViolation:
     message: str | None = None
 
 
-RuleEvaluationType: TypeAlias = Callable[[Model], RuleViolation | None]
+@dataclass
+class SkipRule:
+    """This evaluation of the rule should be skipped."""
+
+    pass
+
+
+RuleEvaluationType: TypeAlias = Callable[[Model], RuleViolation | SkipRule | None]
 
 
 class Rule:
     """The rule base class."""
 
     description: str
     severity: Severity = Severity.MEDIUM
+    model_filters: frozenset[ModelFilter] = frozenset()
     default_config: typing.ClassVar[dict[str, Any]] = {}
 
     def __init__(self, rule_config: RuleConfig | None = None) -> None:
@@ -73,7 +82,7 @@ def process_config(self, rule_config: RuleConfig) -> None:
 
         # Overwrite default rule configuration
         for k, v in rule_config.config.items():
-            if k in self.default_config:
+            if k in self.default_config or k == "model_filter_names":
                 config[k] = v
             else:
                 raise AttributeError(
@@ -85,15 +94,27 @@ def process_config(self, rule_config: RuleConfig) -> None:
         ) if rule_config.severity else rule_config.severity
         self.config = config
 
-    def evaluate(self, model: Model) -> RuleViolation | None:
+    def evaluate(self, model: Model) -> RuleViolation | SkipRule | None:
         """Evaluates the rule."""
         raise NotImplementedError("Subclass must implement method `evaluate`.")
 
+    @classmethod
+    def should_evaluate(cls, model: Model) -> bool:
+        """Checks if all filters in the rule allow evaluation."""
+        if cls.model_filters:
+            return all(f.evaluate(model) for f in cls.model_filters)
+        return True
+
     @classmethod
     def set_severity(cls, severity: Severity) -> None:
         """Set the severity of the rule."""
         cls.severity = severity
 
+    @classmethod
+    def set_filters(cls, model_filters: Iterable[ModelFilter]) -> None:
+        """Set the filters of the rule."""
+        cls.model_filters = frozenset(model_filters)
+
     @classmethod
     def source(cls) -> str:
         """Return the source of the rule, i.e. a fully qualified name."""
@@ -118,6 +139,7 @@ def rule(
     *,
     description: str | None = None,
     severity: Severity = Severity.MEDIUM,
+    model_filters: set[ModelFilter] | None = None,
 ) -> Callable[[RuleEvaluationType], Type[Rule]]:
     ...
 
@@ -127,6 +149,7 @@ def rule(
     *,
     description: str | None = None,
     severity: Severity = Severity.MEDIUM,
+    model_filters: set[ModelFilter] | None = None,
 ) -> Type[Rule] | Callable[[RuleEvaluationType], Type[Rule]]:
     """Rule decorator.
 
@@ -140,6 +163,7 @@ def rule(
         __func: The rule evaluation function being decorated.
         description: The description of the rule.
         severity: The severity of the rule.
+        model_filters: Set of ModelFilter that filters the rule.
     """
 
     def decorator_rule(
@@ -154,7 +178,9 @@ def decorator_rule(
             func.__doc__.split("\n")[0] if func.__doc__ else None
         )
 
-        def wrapped_func(self: Rule, *args: Any, **kwargs: Any) -> RuleViolation | None:
+        def wrapped_func(
+            self: Rule, *args: Any, **kwargs: Any
+        ) -> RuleViolation | SkipRule | None:
             """Wrap func to add `self`."""
             return func(*args, **kwargs)
 
@@ -172,6 +198,7 @@ def wrapped_func(self: Rule, *args: Any, **kwargs: Any) -> RuleViolation | None:
             {
                 "description": rule_description,
                 "severity": severity,
+                "model_filters": model_filters or frozenset(),
                 "default_config": default_config,
                 "evaluate": wrapped_func,
                 # Save provided evaluate function