Support configuration using pyproject.toml

docs/reference/config_parser.md

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

mkdocs.yml

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

pyproject.toml

@@ -92,7 +92,8 @@ max-args = 6
 
 [tool.ruff.lint.per-file-ignores]
 "tests/**/*.py" = [
-    "PLR2004",  # magic value comparisons
+    "PLR2004",  # Magic value comparisons
+    "PLR0913"   # Too many arguments in function definition
 ]
 
 ### 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_parser import DEFAULT_CONFIG_FILE, DbtScoreConfig
 from dbt_score.lint import lint_dbt_project
 from dbt_score.parse import dbt_parse, get_default_manifest_path
 
@@ -57,7 +58,11 @@ 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 = DbtScoreConfig()
+    if Path(DEFAULT_CONFIG_FILE).exists():
+        config.load_toml_file(DEFAULT_CONFIG_FILE)
+
     if run_dbt_parse:
         dbt_parse()
 
-    lint_dbt_project(manifest)
+    lint_dbt_project(manifest, config)

src/dbt_score/config_parser.py

@@ -0,0 +1,80 @@
+"""This module is responsible for parsing configuration."""
+
+import configparser
+import json
+import logging
+from dataclasses import dataclass, field
+from typing import Any, ClassVar
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_CONFIG_FILE = "pyproject.toml"
+
+
+@dataclass
+class RuleConfig:
+    """Configuration for a rule."""
+
+    severity: int | None = None
+    description: str | None = None
+    params: dict[str, Any] = field(default_factory=dict)
+
+    @staticmethod
+    def from_dict(rule_config: dict[str, Any]) -> "RuleConfig":
+        """Create a RuleConfig from a dictionary."""
+        severity = rule_config.pop("severity", None)
+        description = rule_config.pop("description", None)
+
+        return RuleConfig(
+            severity=severity, description=description, params=rule_config
+        )
+
+
+class DbtScoreConfig:
+    """Configuration for dbt-score."""
+
+    _main_section = "tool.dbt-score"
+    _options: ClassVar[list[str]] = ["rule_namespaces", "disabled_rules"]
+    _rules_section = f"{_main_section}.rules"
+
+    def __init__(self) -> None:
+        """Initialize the DbtScoreConfig object."""
+        self.rule_namespaces: list[str] = ["dbt_score_rules"]
+        self.disabled_rules: list[str] = []
+        self.rules_config: dict[str, RuleConfig] = {}
+
+    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."""
+        config = configparser.ConfigParser()
+        config.read(file)
+
+        # Main configuration
+        if config.has_section(self._main_section):
+            for option in config.options(self._main_section):
+                if option in self._options:
+                    self.set_option(
+                        option, json.loads(config.get(self._main_section, option))
+                    )
+                else:
+                    logger.warning(
+                        f"Option {option} in {self._main_section} not supported."
+                    )
+
+        # Rule configuration
+        rules_sections = list(
+            filter(
+                lambda section: section.startswith(self._rules_section),
+                config.sections(),
+            )
+        )
+
+        for rule_section in rules_sections:
+            rule_name = rule_section.replace(f"{self._rules_section}.", "")
+            rule_config = {
+                param: json.loads(val) for param, val in config.items(rule_section)
+            }
+            self.rules_config[rule_name] = RuleConfig.from_dict(rule_config)

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.params)
                 except Exception as e:
                     self.results[model][rule.__class__] = e
                 else:

src/dbt_score/lint.py

@@ -2,19 +2,23 @@
 
 from pathlib import Path
 
+from dbt_score.config_parser import DbtScoreConfig
 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: DbtScoreConfig | None = None) -> None:
     """Lint dbt manifest."""
     if not manifest_path.exists():
         raise FileNotFoundError(f"Manifest not found at {manifest_path}.")
 
-    rule_registry = RuleRegistry()
+    if config is None:
+        config = DbtScoreConfig()
+
+    rule_registry = RuleRegistry(config)
     rule_registry.load_all()
 
     manifest_loader = ManifestLoader(manifest_path)

src/dbt_score/rule.py

@@ -1,9 +1,11 @@
 """Rule definitions."""
-
+import inspect
+import typing
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Callable, Type, TypeAlias, overload
 
+from dbt_score.config_parser import RuleConfig
 from dbt_score.models import Model
 
 
@@ -31,17 +33,50 @@ class Rule:
 
     description: str
     severity: Severity = Severity.MEDIUM
+    default_params: typing.ClassVar[dict[str, Any]] = {}
+
+    def __init__(self, rule_config: RuleConfig | None = None) -> None:
+        """Initialize the rule."""
+        self.params = self.process_config(rule_config) if rule_config else {}
 
     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) -> dict[str, Any]:
+        """Process the rule config."""
+        rule_params = self.default_params.copy()
+
+        # Overwrite default rule params
+        for k, v in rule_config.params.items():
+            if k in self.default_params:
+                rule_params[k] = v
+            else:
+                raise AttributeError(f"Unknown rule parameter: {k}.")
+
+        self.set_severity(rule_config.severity or self.severity)
+        self.set_description(rule_config.description or self.description)
+
+        return rule_params
+
     def evaluate(self, model: Model) -> RuleViolation | None:
         """Evaluates the rule."""
         raise NotImplementedError("Subclass must implement method `evaluate`.")
 
+    @classmethod
+    def set_severity(cls, severity: int | Severity) -> None:
+        """Set the severity of the rule."""
+        if isinstance(severity, int):
+            severity = Severity(severity)
+        cls.severity = severity
+
+    @classmethod
+    def set_description(cls, description: str) -> None:
+        """Set the description of the rule."""
+        cls.description = description
+
     @classmethod
     def source(cls) -> str:
         """Return the source of the rule, i.e. a fully qualified name."""
@@ -106,13 +141,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_params = {
+            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_params": default_params,
                 "evaluate": wrapped_func,
                 # Forward origin of the decorated function
                 "__qualname__": func.__qualname__,  # https://peps.python.org/pep-3155/

src/dbt_score/rule_registry.py

@@ -8,25 +8,32 @@
 import pkgutil
 from typing import Iterator, Type
 
+from dbt_score.config_parser import DbtScoreConfig, RuleConfig
 from dbt_score.exceptions import DuplicatedRuleException
 from dbt_score.rule import Rule
 
 logger = logging.getLogger(__name__)
 
-THIRD_PARTY_RULES_NAMESPACE = "dbt_score_rules"
-
 
 class RuleRegistry:
     """A container for configured rules."""
 
-    def __init__(self) -> None:
+    def __init__(self, config: DbtScoreConfig | None = None) -> None:
         """Instantiate a rule registry."""
+        self.config = config or DbtScoreConfig()
         self._rules: dict[str, Type[Rule]] = {}
+        self._initialized_rules: dict[str, Rule] = {}
+
+    def init_rules(self) -> None:
+        """Initialize rules."""
+        for rule_name, rule_class in self._rules.items():
+            rule_config = self.config.rules_config.get(rule_name, RuleConfig())
+            self._initialized_rules[rule_name] = rule_class(rule_config=rule_config)
 
     @property
-    def rules(self) -> dict[str, Type[Rule]]:
+    def rules(self) -> dict[str, Rule]:
         """Get all rules."""
-        return self._rules
+        return self._initialized_rules
 
     def _walk_packages(self, namespace_name: str) -> Iterator[str]:
         """Walk packages and sub-packages recursively."""
@@ -50,14 +57,18 @@ def _load(self, namespace_name: str) -> None:
             for obj_name in dir(module):
                 obj = module.__dict__[obj_name]
                 if type(obj) is type and issubclass(obj, Rule) and obj is not Rule:
-                    self._add_rule(obj_name, obj)
+                    self._add_rule(f"{module_name}.{obj_name}", obj)
 
     def _add_rule(self, name: str, rule: Type[Rule]) -> None:
-        if name in self.rules:
+        """Add a rule."""
+        if name in self._rules:
             raise DuplicatedRuleException(name)
-        self._rules[name] = rule
+        if name not in self.config.disabled_rules:
+            self._rules[name] = rule
 
     def load_all(self) -> None:
         """Load all rules, core and third-party."""
         self._load("dbt_score.rules")
-        self._load(THIRD_PARTY_RULES_NAMESPACE)
+        for namespace in self.config.rule_namespaces:
+            self._load(namespace)
+        self.init_rules()

tests/conftest.py

@@ -5,8 +5,36 @@
 from typing import Any, Type
 
 from dbt_score import Model, Rule, RuleViolation, Severity, rule
+from dbt_score.config_parser import DbtScoreConfig, RuleConfig
 from pytest import fixture
 
+# Configuration
+
+
+@fixture
+def valid_config_path() -> Path:
+    """Return the path of the configuration."""
+    return Path(__file__).parent / "resources" / "pyproject.toml"
+
+
+@fixture
+def invalid_config_path() -> Path:
+    """Return the path of the configuration."""
+    return Path(__file__).parent / "resources" / "invalid_pyproject.toml"
+
+
+@fixture
+def default_config() -> DbtScoreConfig:
+    """Return a DbtScoreConfig object."""
+    return DbtScoreConfig()
+
+
+@fixture
+def default_rule_config() -> RuleConfig:
+    """Return an empty RuleConfig object."""
+    return RuleConfig()
+
+
 # Manifest
 
 
@@ -156,6 +184,21 @@ def rule_severity_critical(model: Model) -> RuleViolation | None:
     return rule_severity_critical
 
 
+@fixture
+def rule_with_params() -> Type[Rule]:
+    """An example rule with additional input params."""
+
+    @rule
+    def rule_with_params(
+        model: Model, model_name: str = "model1"
+    ) -> RuleViolation | None:
+        """Rule with additional input params."""
+        if model.name != model_name:
+            return RuleViolation(message=model_name)
+
+    return rule_with_params
+
+
 @fixture
 def rule_error() -> Type[Rule]:
     """An example rule which fails to run."""

tests/resources/invalid_pyproject.toml

@@ -0,0 +1,2 @@
+[tool.dbt-score]
+foo = "bar"

tests/resources/pyproject.toml

@@ -0,0 +1,10 @@
+[tool.dbt-score]
+rule_namespaces = ["namespace_foo"]
+disabled_rules = ["foo", "bar"]
+
+
+[tool.dbt-score.rules.foobar]
+severity=4
+
+[tool.dbt-score.rules.rule_with_params]
+model_name="model2"