Adding support for exposures (#112)

As discussed in [this
issue](#111),
adding exposures as a type of evaluable. I believe I've got it plugged
into all the right places - tests pass, and I can run rules locally
along the lines of the below made-up example:
```
@rule(severity=Severity.CRITICAL)
def exposures_must_have_owner(exposure: Exposure) -> RuleViolation | None:
    """Exposures may not be owned by Data Team"""
    if exposure.owner["name"] == "Data Team":
        return RuleViolation(
            message=f"Exposure {exposure.name} may not have an owner {exposure.owner['name']}"
        )
    return None
```

Let me know if I've missed anywhere obvious?

Author: ross-whatnot

CHANGELOG.md

@@ -8,6 +8,8 @@ and this project adheres to
 
 ## [Unreleased]
 
+- Support linting of exposures (#112)
+
 ## [0.12.0] - 2025-05-06
 
 - Add support for linting and scoring dbt seeds (#110)

docs/create_rules.md

@@ -32,7 +32,7 @@ function is its description. Therefore, it is important to use a
 self-explanatory name for the function and document it well.
 
 The type annotation for the rule's argument dictates whether the rule should be
-applied to dbt models, sources or snapshots.
+applied to dbt models, sources, snapshots, seeds, or exposures.
 
 Here is the same example rule, applied to sources:
 
@@ -139,8 +139,8 @@ class SkipSchemaY(RuleFilter):
       return model.schema.lower() != 'y'
 ```
 
-Filters also rely on type-annotations to dictate whether they apply to models
-sources or snapshots:
+Filters also rely on type-annotations to dictate whether they apply to models,
+sources, snapshots, seeds, or exposures:
 
 ```python
 from dbt_score import RuleFilter, rule_filter, Source

docs/index.md

@@ -14,6 +14,8 @@ encourage) good practices. The dbt entities that `dbt-score` is able to lint
 - Models
 - Sources
 - Snapshots
+- Exposures
+- Seeds
 
 ## Example
 

src/dbt_score/__init__.py

@@ -1,10 +1,11 @@
 """Init dbt_score package."""
 
-from dbt_score.models import Model, Seed, Snapshot, Source
+from dbt_score.models import Exposure, Model, Seed, Snapshot, Source
 from dbt_score.rule import Rule, RuleViolation, Severity, rule
 from dbt_score.rule_filter import RuleFilter, rule_filter
 
 __all__ = [
+    "Exposure",
     "Model",
     "Source",
     "Snapshot",
@@ -13,6 +14,6 @@
     "Rule",
     "RuleViolation",
     "Severity",
-    "rule_filter",
     "rule",
+    "rule_filter",
 ]

src/dbt_score/dbt_utils.py

@@ -75,6 +75,7 @@ def dbt_ls(select: Iterable[str] | None) -> Iterable[str]:
         "model",
         "source",
         "snapshot",
+        "exposure",
         "seed",
         "--output",
         "name",

src/dbt_score/evaluation.py

@@ -64,6 +64,7 @@ def evaluate(self) -> None:
             self._manifest_loader.models.values(),
             self._manifest_loader.sources.values(),
             self._manifest_loader.snapshots.values(),
+            self._manifest_loader.exposures.values(),
             self._manifest_loader.seeds.values(),
         ):
             # type inference on elements from `chain` is wonky
@@ -98,6 +99,7 @@ def evaluate(self) -> None:
             self._manifest_loader.models
             or self._manifest_loader.sources
             or self._manifest_loader.snapshots
+            or self._manifest_loader.exposures
             or self._manifest_loader.seeds
         ):
             self._formatter.project_evaluated(self.project_score)

src/dbt_score/formatters/human_readable_formatter.py

@@ -4,7 +4,7 @@
 
 from dbt_score.evaluation import EvaluableResultsType
 from dbt_score.formatters import Formatter
-from dbt_score.models import Evaluable, Model, Seed, Snapshot, Source
+from dbt_score.models import Evaluable, Exposure, Model, Seed, Snapshot, Source
 from dbt_score.rule import RuleViolation
 from dbt_score.scoring import Score
 
@@ -37,6 +37,8 @@ def pretty_name(evaluable: Evaluable) -> str:
                 return evaluable.selector_name
             case Snapshot():
                 return evaluable.name
+            case Exposure():
+                return evaluable.name
             case Seed():
                 return evaluable.name
             case _:

src/dbt_score/formatters/manifest_formatter.py

@@ -36,4 +36,8 @@ def project_evaluated(self, score: Score) -> None:
                 source_manifest = manifest["sources"][evaluable_id]
                 source_manifest["meta"]["score"] = evaluable_score.value
                 source_manifest["meta"]["badge"] = evaluable_score.badge
+            if evaluable_id.startswith("exposure"):
+                exposure_manifest = manifest["exposures"][evaluable_id]
+                exposure_manifest["meta"]["score"] = evaluable_score.value
+                exposure_manifest["meta"]["badge"] = evaluable_score.badge
         print(json.dumps(manifest, indent=2))

src/dbt_score/models.py

@@ -156,7 +156,7 @@ def _get_columns(
 
 # Type annotation for parent references
 ParentType = Union["Model", "Source", "Snapshot", "Seed"]
-ChildType = Union["Model", "Snapshot"]
+ChildType = Union["Model", "Snapshot", "Exposure"]
 
 
 @dataclass
@@ -499,6 +499,71 @@ def __hash__(self) -> int:
         return hash(self.unique_id)
 
 
+@dataclass
+class Exposure:
+    """Represents a dbt exposure.
+
+    Attributes:
+        unique_id: The unique id of the exposure (e.g. `exposure.package.exposure1`).
+        name: The name of the exposure.
+        description: The description of the exposure.
+        label: The label of the exposure.
+        url: The url of the exposure.
+        maturity: The maturity of the exposure.
+        original_file_path: The path to the exposure file
+            (e.g. `models/exposures/exposures.yml`).
+        type: The type of the exposure, e.g. `application`.
+        owner: The owner of the exposure,
+            e.g. `{"name": "owner", "email": "[email protected]"}`.
+        config: The config of the exposure.
+        meta: The meta of the exposure.
+        tags: The list of tags attached to the exposure.
+        depends_on: The depends_on of the exposure.
+        parents: The list of models, sources, and snapshot this exposure depends on.
+        _raw_values: The raw values of the exposure in the manifest.
+    """
+
+    unique_id: str
+    name: str
+    description: str
+    label: str
+    url: str
+    maturity: str
+    original_file_path: str
+    type: str
+    owner: dict[str, Any]
+    config: dict[str, Any]
+    meta: dict[str, Any]
+    tags: list[str]
+    depends_on: dict[str, list[str]] = field(default_factory=dict)
+    parents: list[ParentType] = field(default_factory=list)
+    _raw_values: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_node(cls, node_values: dict[str, Any]) -> "Exposure":
+        """Create an exposure object from a node in the manifest."""
+        return cls(
+            unique_id=node_values["unique_id"],
+            name=node_values["name"],
+            description=node_values["description"],
+            label=node_values["label"],
+            url=node_values["url"],
+            maturity=node_values["maturity"],
+            original_file_path=node_values["original_file_path"],
+            type=node_values["type"],
+            owner=node_values["owner"],
+            config=node_values["config"],
+            meta=node_values["meta"],
+            tags=node_values["tags"],
+            depends_on=node_values["depends_on"],
+            _raw_values=node_values,
+        )
+
+    def __hash__(self) -> int:
+        """Compute a unique hash for an exposure."""
+        return hash(self.unique_id)
+
+
 @dataclass
 class Seed(HasColumnsMixin):
     """Represents a dbt seed.
@@ -579,11 +644,11 @@ def __hash__(self) -> int:
         return hash(self.unique_id)
 
 
-Evaluable: TypeAlias = Model | Source | Snapshot | Seed
+Evaluable: TypeAlias = Model | Source | Snapshot | Seed | Exposure
 
 
 class ManifestLoader:
-    """Load the models, sources, snapshots, seeds and tests from the manifest."""
+    """Load the evaluables from the manifest."""
 
     def __init__(self, file_path: Path, select: Iterable[str] | None = None):
         """Initialize the ManifestLoader.
@@ -604,25 +669,38 @@ def __init__(self, file_path: Path, select: Iterable[str] | None = None):
             for source_id, source_values in self.raw_manifest.get("sources", {}).items()
             if source_values["package_name"] == self.project_name
         }
+        self.raw_exposures = {
+            exposure_id: exposure_values
+            for exposure_id, exposure_values in self.raw_manifest.get(
+                "exposures", {}
+            ).items()
+            if exposure_values["package_name"] == self.project_name
+        }
 
         self.models: dict[str, Model] = {}
         self.tests: dict[str, list[dict[str, Any]]] = defaultdict(list)
         self.sources: dict[str, Source] = {}
         self.snapshots: dict[str, Snapshot] = {}
+        self.exposures: dict[str, Exposure] = {}
         self.seeds: dict[str, Seed] = {}
 
         self._reindex_tests()
         self._load_models()
         self._load_sources()
         self._load_snapshots()
+        self._load_exposures()
         self._load_seeds()
         self._populate_relatives()
 
         if select:
             self._filter_evaluables(select)
 
         if (
-            len(self.models) + len(self.sources) + len(self.snapshots) + len(self.seeds)
+            len(self.models)
+            + len(self.sources)
+            + len(self.snapshots)
+            + len(self.seeds)
+            + len(self.exposures)
         ) == 0:
             logger.warning("Nothing to evaluate!")
 
@@ -647,6 +725,13 @@ def _load_snapshots(self) -> None:
                 snapshot = Snapshot.from_node(node_values, self.tests.get(node_id, []))
                 self.snapshots[node_id] = snapshot
 
+    def _load_exposures(self) -> None:
+        """Load the exposures from the manifest."""
+        for node_id, node_values in self.raw_exposures.items():
+            if node_values.get("resource_type") == "exposure":
+                exposure = Exposure.from_node(node_values)
+                self.exposures[node_id] = exposure
+
     def _load_seeds(self) -> None:
         """Load the seeds from the manifest."""
         for node_id, node_values in self.raw_nodes.items():
@@ -672,7 +757,11 @@ def _reindex_tests(self) -> None:
 
     def _populate_relatives(self) -> None:
         """Populate `parents` and `children` for all evaluables."""
-        for node in list(self.models.values()) + list(self.snapshots.values()):
+        for node in (
+            list(self.models.values())
+            + list(self.snapshots.values())
+            + list(self.exposures.values())
+        ):
             for parent_id in node.depends_on.get("nodes", []):
                 if parent_id in self.models:
                     node.parents.append(self.models[parent_id])
@@ -704,4 +793,5 @@ def _filter_evaluables(self, select: Iterable[str]) -> None:
             k: s for k, s in self.sources.items() if s.selector_name in selected
         }
         self.snapshots = {k: s for k, s in self.snapshots.items() if s.name in selected}
+        self.exposures = {k: e for k, e in self.exposures.items() if e.name in selected}
         self.seeds = {k: s for k, s in self.seeds.items() if s.name in selected}

src/dbt_score/rule.py

@@ -14,7 +14,7 @@
     overload,
 )
 
-from dbt_score.models import Evaluable, Model, Seed, Snapshot, Source
+from dbt_score.models import Evaluable, Exposure, Model, Seed, Snapshot, Source
 from dbt_score.more_itertools import first_true
 from dbt_score.rule_filter import RuleFilter
 
@@ -66,11 +66,13 @@ class RuleViolation:
 ModelRuleEvaluationType: TypeAlias = Callable[[Model], RuleViolation | None]
 SourceRuleEvaluationType: TypeAlias = Callable[[Source], RuleViolation | None]
 SnapshotRuleEvaluationType: TypeAlias = Callable[[Snapshot], RuleViolation | None]
+ExposureRuleEvaluationType: TypeAlias = Callable[[Exposure], RuleViolation | None]
 SeedRuleEvaluationType: TypeAlias = Callable[[Seed], RuleViolation | None]
 RuleEvaluationType: TypeAlias = (
     ModelRuleEvaluationType
     | SourceRuleEvaluationType
     | SnapshotRuleEvaluationType
+    | ExposureRuleEvaluationType
     | SeedRuleEvaluationType
 )
 
@@ -210,6 +212,11 @@ def rule(__func: SnapshotRuleEvaluationType) -> Type[Rule]:
     ...
 
 
+@overload
+def rule(__func: ExposureRuleEvaluationType) -> Type[Rule]:
+    ...
+
+
 @overload
 def rule(__func: SeedRuleEvaluationType) -> Type[Rule]:
     ...