Repository: PicnicSupermarket/dbt-score

Issue:
Improve documentation on how to apply rule filters
It would be nice to have an example how to apply a rule filter under **Rule configuration** on this [documentation page](https://dbt-score.picnic.tech/configuration/).

Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 699109c..df8d47c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,7 +8,15 @@ and this project adheres to
 
 ## [Unreleased]
 
+## [0.10.0] - 2025-01-27
+
 - Add debug mode to help writing new rules. (#91)
+- Fix tests without metadata. (#88)
+- Add new rule to enforce presence of uniqueness test. (#90)
+- Add new rule to enforce single-column PK to be defined at column level. (#90)
+- Add new rule to enforce 1 column uniqueness test to be defined at column
+  level. (#90)
+- Add `constraints` to the model schema. (#90)
 
 ## [0.9.0] - 2024-12-19
 
diff --git a/docs/configuration.md b/docs/configuration.md
index a64df37..e00fea8 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -90,6 +90,13 @@ Every rule can be configured with the following option:
   be found in the same namespace as the rules (see
   [Package rules](package_rules.md)).
 
+  Example: the generic rule `has_example_sql` shall apply only to models materializing a table.
+
+  ```toml
+  [tool.dbt-score.rules."dbt_score.rules.generic.has_example_sql"]
+  rule_filter_names=["dbt_score.rules.filters.is_table"] 
+  ```
+
 Some rules have additional configuration options, e.g.
 [sql_has_reasonable_number_of_lines](rules/generic.md#sql_has_reasonable_number_of_lines).
 Depending on the rule, the options will have different names, types and default
diff --git a/docs/rules/filters.md b/docs/rules/filters.md
new file mode 100644
index 0000000..758fb08
--- /dev/null
+++ b/docs/rules/filters.md
@@ -0,0 +1,3 @@
+# Rule filters
+
+::: dbt_score.rules.filters
diff --git a/mkdocs.yml b/mkdocs.yml
index 6639838..0a23fa9 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -28,6 +28,7 @@ nav:
   - Programmatic invocations: programmatic_invocations.md
   - Rules:
       - rules/generic.md
+      - rules/filters.md
   - Reference:
       - reference/cli.md
       - reference/config.md
diff --git a/src/dbt_score/models.py b/src/dbt_score/models.py
index 79a76b1..a422295 100644
--- a/src/dbt_score/models.py
+++ b/src/dbt_score/models.py
@@ -15,18 +15,20 @@ logger = logging.getLogger(__name__)
 
 @dataclass
 class Constraint:
-    """Constraint for a column.
+    """Constraint for a model or a column.
 
     Attributes:
         type: The type of the constraint, e.g. `foreign_key`.
         name: The name of the constraint.
         expression: The expression of the constraint, e.g. `schema.other_table`.
+        columns: The columns for the constraint (only for model-level constraints).
         _raw_values: The raw values of the constraint in the manifest.
     """
 
     type: str
     name: str | None = None
     expression: str | None = None
+    columns: list[str] | None = None
     _raw_values: dict[str, Any] = field(default_factory=dict)
 
     @classmethod
@@ -36,6 +38,7 @@ class Constraint:
             type=raw_values["type"],
             name=raw_values["name"],
             expression=raw_values["expression"],
+            columns=raw_values.get("columns"),
             _raw_values=raw_values,
         )
 
@@ -63,8 +66,8 @@ class Test:
         """Create a test object from a test node in the manifest."""
         return cls(
             name=test_node["name"],
-            type=test_node["test_metadata"]["name"],
-            kwargs=test_node["test_metadata"].get("kwargs", {}),
+            type=test_node.get("test_metadata", {}).get("name", "generic"),
+            kwargs=test_node.get("test_metadata", {}).get("kwargs", {}),
             tags=test_node.get("tags", []),
             _raw_values=test_node,
         )
@@ -141,7 +144,10 @@ class HasColumnsMixin:
                 [
                     test
                     for test in test_values
-                    if test["test_metadata"]["kwargs"].get("column_name") == name
+                    if test.get("test_metadata", {})
+                    .get("kwargs", {})
+                    .get("column_name")
+                    == name
                 ],
             )
             for name, values in node_values.get("columns", {}).items()
@@ -195,6 +201,7 @@ class Model(HasColumnsMixin):
     tags: list[str] = field(default_factory=list)
     tests: list[Test] = field(default_factory=list)
     depends_on: dict[str, list[str]] = field(default_factory=dict)
+    constraints: list[Constraint] = field(default_factory=list)
     _raw_values: dict[str, Any] = field(default_factory=dict)
     _raw_test_values: list[dict[str, Any]] = field(default_factory=list)
 
@@ -224,9 +231,15 @@ class Model(HasColumnsMixin):
             tests=[
                 Test.from_node(test)
                 for test in test_values
-                if not test["test_metadata"]["kwargs"].get("column_name")
+                if not test.get("test_metadata", {})
+                .get("kwargs", {})
+                .get("column_name")
             ],
             depends_on=node_values["depends_on"],
+            constraints=[
+                Constraint.from_raw_values(constraint)
+                for constraint in node_values["constraints"]
+            ],
             _raw_values=node_values,
             _raw_test_values=test_values,
         )
@@ -359,7 +372,9 @@ class Source(HasColumnsMixin):
             tests=[
                 Test.from_node(test)
                 for test in test_values
-                if not test["test_metadata"]["kwargs"].get("column_name")
+                if not test.get("test_metadata", {})
+                .get("kwargs", {})
+                .get("column_name")
             ],
             _raw_values=node_values,
             _raw_test_values=test_values,
@@ -428,20 +443,17 @@ class ManifestLoader:
         """Index tests based on their associated evaluable."""
         for node_values in self.raw_nodes.values():
             if node_values.get("resource_type") == "test":
-                # tests for models have a non-null value for `attached_node`
+                # Tests for models have a non-null value for `attached_node`
                 if attached_node := node_values.get("attached_node"):
                     self.tests[attached_node].append(node_values)
 
-                # Tests for sources will have a null `attached_node`,
-                # and a non-empty list for `sources`.
-                # They need to be attributed to the source id
+                # Tests for sources or separate tests will have `attached_node` == null.
+                # They need to be attributed to the node id
                 # based on the `depends_on` field.
-                elif node_values.get("sources") and (
-                    source_unique_id := next(
-                        iter(node_values.get("depends_on", {}).get("nodes", [])), None
-                    )
+                elif node_unique_id := next(
+                    iter(node_values.get("depends_on", {}).get("nodes", [])), None
                 ):
-                    self.tests[source_unique_id].append(node_values)
+                    self.tests[node_unique_id].append(node_values)
 
     def _filter_evaluables(self, select: Iterable[str]) -> None:
         """Filter evaluables like dbt's --select."""
diff --git a/src/dbt_score/rules/filters.py b/src/dbt_score/rules/filters.py
new file mode 100644
index 0000000..c48f13f
--- /dev/null
+++ b/src/dbt_score/rules/filters.py
@@ -0,0 +1,10 @@
+"""Rule filters."""
+
+from dbt_score import rule_filter
+from dbt_score.models import Model
+
+
+@rule_filter
+def is_table(model: Model) -> bool:
+    """Models that are tables."""
+    return model.config.get("materialized") in {"table", "incremental"}
diff --git a/src/dbt_score/rules/generic.py b/src/dbt_score/rules/generic.py
index 1e04b1f..d4ce149 100644
--- a/src/dbt_score/rules/generic.py
+++ b/src/dbt_score/rules/generic.py
@@ -1,6 +1,7 @@
 """All generic rules."""
 
 from dbt_score import Model, RuleViolation, Severity, rule
+from dbt_score.rules.filters import is_table
 
 
 @rule
@@ -47,7 +48,65 @@ def sql_has_reasonable_number_of_lines(
 def has_example_sql(model: Model) -> RuleViolation | None:
     """The documentation of a model should have an example query."""
     if model.language == "sql":
-        if "```sql" not in model.description:
+        if "```sql" not in (model.description or ""):
             return RuleViolation(
                 "The model description does not include an example SQL query."
             )
+
+
+@rule(rule_filters={is_table()})
+def single_pk_defined_at_column_level(model: Model) -> RuleViolation | None:
+    """Single-column PK must be defined as a column constraint."""
+    for constraint in model.constraints:
+        if constraint.type == "primary_key":
+            if constraint.columns is not None and len(constraint.columns) == 1:
+                return RuleViolation(
+                    f"Single-column PK {constraint.columns[0]} must be defined as a "
+                    f"column constraint."
+                )
+
+
+@rule(rule_filters={is_table()})
+def single_column_uniqueness_at_column_level(model: Model) -> RuleViolation | None:
+    """Single-column uniqueness test must be defined as a column test."""
+    for data_test in model.tests:
+        if data_test.type == "unique_combination_of_columns":
+            if len(data_test.kwargs.get("combination_of_columns", [])) == 1:
+                return RuleViolation(
+                    "Single-column uniqueness test must be defined as a column test."
+                )
+
+
+@rule(rule_filters={is_table()})
+def has_uniqueness_test(model: Model) -> RuleViolation | None:
+    """Model has uniqueness test for primary key."""
+    # ruff: noqa: C901 [too-complex]
+
+    # Single-column PK
+    for column in model.columns:
+        for column_constraint in column.constraints:
+            if column_constraint.type == "primary_key":
+                for data_test in column.tests:
+                    if data_test.type == "unique":
+                        return None
+                return RuleViolation(
+                    f"No unique constraint defined on PK column {column.name}."
+                )
+
+    # Composite PK
+    pk_columns: list[str] = []
+    for model_constraint in model.constraints:
+        if model_constraint.type == "primary_key":
+            pk_columns = model_constraint.columns or []
+            break
+
+    if not pk_columns: # No PK, no need for uniqueness test
+        return None
+
+    for data_test in model.tests:
+        if data_test.type == "unique_combination_of_columns":
+            if set(data_test.kwargs.get("combination_of_columns")) == set(pk_columns): # type: ignore
+                return None
+    return RuleViolation(
+        f"No uniqueness test defined and matching PK {','.join(pk_columns)}."
+    )


Test patch:
diff --git a/tests/resources/manifest.json b/tests/resources/manifest.json
index af5908b..e1d5e2f 100644
--- a/tests/resources/manifest.json
+++ b/tests/resources/manifest.json
@@ -26,6 +26,7 @@
           "tags": []
         }
       },
+      "constraints": [],
       "package_name": "package",
       "database": "db",
       "schema": "schema",
@@ -56,6 +57,7 @@
           "tags": []
         }
       },
+      "constraints": [],
       "package_name": "package",
       "database": "db",
       "schema": "schema",
@@ -86,6 +88,7 @@
           "tags": []
         }
       },
+      "constraints": [],
       "package_name": "package2",
       "database": "db",
       "schema": "schema",
@@ -125,6 +128,13 @@
       "resource_type": "test",
       "package_name": "package"
     },
+    "test.package.test4": {
+      "resource_type": "test",
+      "depends_on": { "nodes": ["model.package.model1"] },
+      "name": "test4",
+      "package_name": "package",
+      "tags": []
+    },
     "test.package.source_test1": {
       "resource_type": "test",
       "package_name": "package",
diff --git a/tests/test_models.py b/tests/test_models.py
index b0fac9d..849ccea 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -20,6 +20,7 @@ def test_manifest_load(mock_read_text, raw_manifest):
             ]
         )
         assert loader.models[0].tests[0].name == "test2"
+        assert loader.models[0].tests[1].name == "test4"
         assert loader.models[0].columns[0].tests[0].name == "test1"
 
         assert len(loader.sources) == len(