Support configuration using pyproject.toml

.github/workflows/ci.yml

@@ -18,14 +18,14 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python: ["3.10", "3.11"]
+        python: ["3.11", "3.12"]
 
     steps:
       - uses: actions/checkout@v4
       - name: Set up PDM
         uses: pdm-project/setup-pdm@v4
         with:
-          python-version: "3.10"
+          python-version: "3.11"
       - name: Install dependencies
         run: |
           pdm sync -d

README.md

@@ -8,7 +8,7 @@ Linter for dbt model metadata.
 
 You'll need the following prerequisites:
 
-- Any Python version starting from 3.10
+- Any Python version starting from 3.11
 - [pre-commit](https://pre-commit.com/)
 - [PDM](https://pdm-project.org/2.12/)
 

docs/reference/config.md

@@ -0,0 +1,3 @@
+# Config
+
+::: dbt_score.config

mkdocs.yml

@@ -8,6 +8,7 @@ nav:
   - Home: index.md
   - Reference:
       - reference/cli.md
+      - reference/config.md
       - reference/exceptions.md
       - reference/evaluation.md
       - reference/models.md

pyproject.toml

@@ -22,7 +22,7 @@ dependencies = [
     "dbt-core>=1.5",
     "click>=7.1.1, <9.0.0",
 ]
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 readme = "README.md"
 license = {text = "MIT"}
 
@@ -92,7 +92,7 @@ max-args = 6
 
 [tool.ruff.lint.per-file-ignores]
 "tests/**/*.py" = [
-    "PLR2004",  # magic value comparisons
+    "PLR2004",  # Magic value comparisons
 ]
 
 ### Coverage ###

src/dbt_score/cli.py

@@ -7,6 +7,7 @@
 from click.core import ParameterSource
 from dbt.cli.options import MultiOption
 
+from dbt_score.config import Config
 from dbt_score.lint import lint_dbt_project
 from dbt_score.parse import dbt_parse, get_default_manifest_path
 
@@ -57,7 +58,10 @@ def lint(select: tuple[str], manifest: Path, run_dbt_parse: bool) -> None:
     if manifest_provided and run_dbt_parse:
         raise click.UsageError("--run-dbt-parse cannot be used with --manifest.")
 
+    config = Config()
+    config.load()
+
     if run_dbt_parse:
         dbt_parse()
 
-    lint_dbt_project(manifest)
+    lint_dbt_project(manifest, config)

src/dbt_score/config.py

@@ -0,0 +1,72 @@
+"""This module is responsible for loading configuration."""
+
+import logging
+import tomllib
+from pathlib import Path
+from typing import Any, Final
+
+from dbt_score.rule import RuleConfig
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_CONFIG_FILE = "pyproject.toml"
+
+
+class Config:
+    """Configuration for dbt-score."""
+
+    _main_section: Final[str] = "tool.dbt-score"
+    _options: Final[list[str]] = ["rule_namespaces", "disabled_rules"]
+    _rules_section: Final[str] = f"{_main_section}.rules"
+
+    def __init__(self) -> None:
+        """Initialize the Config object."""
+        self.rule_namespaces: list[str] = ["dbt_score_rules"]
+        self.disabled_rules: list[str] = []
+        self.rules_config: dict[str, RuleConfig] = {}
+        self.config_file: Path | None = None
+
+    def set_option(self, option: str, value: Any) -> None:
+        """Set an option in the config."""
+        setattr(self, option, value)
+
+    def _load_toml_file(self, file: str) -> None:
+        """Load the options from a TOML file."""
+        with open(file, "rb") as f:
+            toml_data = tomllib.load(f)
+
+        tools = toml_data.get("tool", {})
+        dbt_score_config = tools.get("dbt-score", {})
+        rules_config = dbt_score_config.pop("rules", {})
+
+        # Main configuration
+        for option, value in dbt_score_config.items():
+            if option in self._options:
+                self.set_option(option, value)
+            elif not isinstance(
+                value, dict
+            ):  # If value is a dictionary, it's another section
+                logger.warning(
+                    f"Option {option} in {self._main_section} not supported."
+                )
+
+        # Rule configuration
+        self.rules_config = {
+            name: RuleConfig.from_dict(config) for name, config in rules_config.items()
+        }
+
+    @staticmethod
+    def get_config_file(directory: Path) -> Path | None:
+        """Get the config file."""
+        candidates = [directory]
+        candidates.extend(directory.parents)
+        for path in candidates:
+            config_file = path / DEFAULT_CONFIG_FILE
+            if config_file.exists():
+                return config_file
+
+    def load(self) -> None:
+        """Load the config."""
+        config_file = self.get_config_file(Path.cwd())
+        if config_file:
+            self._load_toml_file(str(config_file))

src/dbt_score/evaluation.py

@@ -51,15 +51,13 @@ def __init__(
 
     def evaluate(self) -> None:
         """Evaluate all rules."""
-        # Instantiate all rules. In case they keep state across calls, this must be
-        # done only once.
-        rules = [rule_class() for rule_class in self._rule_registry.rules.values()]
+        rules = self._rule_registry.rules.values()
 
         for model in self._manifest_loader.models:
             self.results[model] = {}
             for rule in rules:
                 try:
-                    result: RuleViolation | None = rule.evaluate(model)
+                    result: RuleViolation | None = rule.evaluate(model, **rule.config)
                 except Exception as e:
                     self.results[model][rule.__class__] = e
                 else:

src/dbt_score/lint.py

@@ -2,19 +2,20 @@
 
 from pathlib import Path
 
+from dbt_score.config import Config
 from dbt_score.evaluation import Evaluation
 from dbt_score.formatters.human_readable_formatter import HumanReadableFormatter
 from dbt_score.models import ManifestLoader
 from dbt_score.rule_registry import RuleRegistry
 from dbt_score.scoring import Scorer
 
 
-def lint_dbt_project(manifest_path: Path) -> None:
+def lint_dbt_project(manifest_path: Path, config: Config) -> None:
     """Lint dbt manifest."""
     if not manifest_path.exists():
         raise FileNotFoundError(f"Manifest not found at {manifest_path}.")
 
-    rule_registry = RuleRegistry()
+    rule_registry = RuleRegistry(config)
     rule_registry.load_all()
 
     manifest_loader = ManifestLoader(manifest_path)

src/dbt_score/rule.py

@@ -1,6 +1,8 @@
 """Rule definitions."""
 
-from dataclasses import dataclass
+import inspect
+import typing
+from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, Callable, Type, TypeAlias, overload
 
@@ -16,6 +18,26 @@ class Severity(Enum):
     CRITICAL = 4
 
 
+@dataclass
+class RuleConfig:
+    """Configuration for a rule."""
+
+    severity: Severity | None = None
+    config: dict[str, Any] = field(default_factory=dict)
+
+    @staticmethod
+    def from_dict(rule_config: dict[str, Any]) -> "RuleConfig":
+        """Create a RuleConfig from a dictionary."""
+        config = rule_config.copy()
+        severity = (
+            Severity(config.pop("severity", None))
+            if "severity" in rule_config
+            else None
+        )
+
+        return RuleConfig(severity=severity, config=config)
+
+
 @dataclass
 class RuleViolation:
     """The violation of a rule."""
@@ -31,17 +53,47 @@ class Rule:
 
     description: str
     severity: Severity = Severity.MEDIUM
+    default_config: typing.ClassVar[dict[str, Any]] = {}
+
+    def __init__(self, rule_config: RuleConfig | None = None) -> None:
+        """Initialize the rule."""
+        self.config: dict[str, Any] = {}
+        if rule_config:
+            self.process_config(rule_config)
 
     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 process_config(self, rule_config: RuleConfig) -> None:
+        """Process the rule config."""
+        config = self.default_config.copy()
+
+        # Overwrite default rule configuration
+        for k, v in rule_config.config.items():
+            if k in self.default_config:
+                config[k] = v
+            else:
+                raise AttributeError(
+                    f"Unknown rule parameter: {k} for rule {self.source()}."
+                )
+
+        self.set_severity(
+            rule_config.severity
+        ) if rule_config.severity else rule_config.severity
+        self.config = config
+
     def evaluate(self, model: Model) -> RuleViolation | None:
         """Evaluates the rule."""
         raise NotImplementedError("Subclass must implement method `evaluate`.")
 
+    @classmethod
+    def set_severity(cls, severity: Severity) -> None:
+        """Set the severity of the rule."""
+        cls.severity = severity
+
     @classmethod
     def source(cls) -> str:
         """Return the source of the rule, i.e. a fully qualified name."""
@@ -106,13 +158,21 @@ def wrapped_func(self: Rule, *args: Any, **kwargs: Any) -> RuleViolation | None:
             """Wrap func to add `self`."""
             return func(*args, **kwargs)
 
+        # Get default parameters from the rule definition
+        default_config = {
+            key: val.default
+            for key, val in inspect.signature(func).parameters.items()
+            if val.default != inspect.Parameter.empty
+        }
+
         # Create the rule class inheriting from Rule
         rule_class = type(
             func.__name__,
             (Rule,),
             {
                 "description": rule_description,
                 "severity": severity,
+                "default_config": default_config,
                 "evaluate": wrapped_func,
                 # Forward origin of the decorated function
                 "__qualname__": func.__qualname__,  # https://peps.python.org/pep-3155/