Create rule registry and discovery

src/dbt_score/exceptions.py

@@ -0,0 +1,11 @@
+"""dbt-score exceptions."""
+
+
+class DuplicatedRuleException(Exception):
+    """Two rules with the same name are defined."""
+
+    def __init__(self, rule_name: str):
+        """Instantiate exception."""
+        super().__init__(
+            f"Rule {rule_name} is defined twice. Rules must have unique names."
+        )

src/dbt_score/rule_registry.py

@@ -0,0 +1,57 @@
+"""Rule registry.
+
+This module implements rule discovery.
+"""
+
+import importlib
+import pkgutil
+from typing import Iterator, Type
+
+from dbt_score.exceptions import DuplicatedRuleException
+from dbt_score.rule import Rule
+
+THIRD_PARTY_RULES_NAMESPACE = "dbt_score_rules"
+
+
+class RuleRegistry:
+    """A container for configured rules."""
+
+    def __init__(self) -> None:
+        """Instantiate a rule registry."""
+        self._rules: dict[str, Type[Rule]] = {}
+
+    @property
+    def rules(self) -> dict[str, Type[Rule]]:
+        """Get all rules."""
+        return self._rules
+
+    def _walk_packages(self, namespace_name: str) -> Iterator[str]:
+        """Walk packages and sub-packages recursively."""
+        try:
+            namespace = importlib.import_module(namespace_name)
+        except ImportError:  # no custom rule in Python path
+            return
+
+        def onerror(module_name: str) -> None:
+            print(f"Failed to import {module_name}.")
+
+        for package in pkgutil.walk_packages(namespace.__path__, onerror=onerror):
+            yield f"{namespace_name}.{package.name}"
+            if package.ispkg:
+                yield from self._walk_packages(f"{namespace_name}.{package.name}")
+
+    def _load(self, namespace_name: str) -> None:
+        """Load rules found in a given namespace."""
+        for module_name in self._walk_packages(namespace_name):
+            module = importlib.import_module(module_name)
+            for obj_name in dir(module):
+                obj = module.__dict__[obj_name]
+                if type(obj) is type and issubclass(obj, Rule):
+                    if obj_name in self.rules:
+                        raise DuplicatedRuleException(obj_name)
+                    self._rules[obj_name] = obj
+
+    def load_all(self) -> None:
+        """Load all rules, core and third-party."""
+        self._load("dbt_score.rules")
+        self._load(THIRD_PARTY_RULES_NAMESPACE)

src/dbt_score/rules/generic.py

@@ -1,4 +1,5 @@
 """All generic rules."""
+
 from dbt_score.models import Model
 from dbt_score.rule import RuleViolation, rule
 

tests/rules/example.py

@@ -0,0 +1,9 @@
+"""Example rules."""
+
+from dbt_score.models import Model
+from dbt_score.rule import RuleViolation, rule
+
+
+@rule()
+def rule_test_example(model: Model) -> RuleViolation | None:
+    """An example rule."""

tests/rules/nested/__init__.py

@@ -0,0 +1 @@
+"""Nested package for testing rule discovery."""

tests/rules/nested/example.py

@@ -0,0 +1,9 @@
+"""Example rules."""
+
+from dbt_score.models import Model
+from dbt_score.rule import RuleViolation, rule
+
+
+@rule()
+def rule_test_nested_example(model: Model) -> RuleViolation | None:
+    """An example rule."""

tests/test_rule_registry.py

@@ -0,0 +1,27 @@
+"""Unit tests for the rule registry."""
+
+import pytest
+from dbt_score.exceptions import DuplicatedRuleException
+from dbt_score.rule_registry import RuleRegistry
+
+
+def test_rule_registry_discovery():
+    """Ensure rules can be found in a given namespace recursively."""
+    r = RuleRegistry()
+    r._load("tests.rules")
+    assert sorted(r.rules.keys()) == ["rule_test_example", "rule_test_nested_example"]
+
+
+def test_rule_registry_no_duplicates():
+    """Ensure no duplicate rule names can coexist."""
+    r = RuleRegistry()
+    r._load("tests.rules")
+    with pytest.raises(DuplicatedRuleException):
+        r._load("tests.rules")
+
+
+def test_rule_registry_core_rules():
+    """Ensure core rules are automatically discovered."""
+    r = RuleRegistry()
+    r.load_all()
+    assert len(r.rules) > 0