Skip to content

Evaluation

This module is responsible for evaluating rules.

Evaluation

Evaluate a set of rules on a set of nodes.

Source code in src/dbt_score/evaluation.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class Evaluation:
    """Evaluate a set of rules on a set of nodes."""

    def __init__(
        self,
        rule_registry: RuleRegistry,
        manifest_loader: ManifestLoader,
        formatter: Formatter,
        scorer: Scorer,
    ) -> None:
        """Create an Evaluation object.

        Args:
            rule_registry: A rule registry to access rules.
            manifest_loader: A manifest loader to access dbt metadata.
            formatter: A formatter to display results.
            scorer: A scorer to compute scores.
        """
        self._rule_registry = rule_registry
        self._manifest_loader = manifest_loader
        self._formatter = formatter
        self._scorer = scorer

        # For each evaluable, its results
        self.results: dict[Evaluable, EvaluableResultsType] = {}

        # For each evaluable, its computed score
        self.scores: dict[Evaluable, Score] = {}

        # The aggregated project score
        self.project_score: Score

    def evaluate(self) -> None:
        """Evaluate all rules."""
        rules = self._rule_registry.rules.values()

        for evaluable in chain(
            self._manifest_loader.models, self._manifest_loader.sources
        ):
            # type inference on elements from `chain` is wonky
            # and resolves to superclass HasColumnsMixin
            evaluable = cast(Evaluable, evaluable)
            self.results[evaluable] = {}
            for rule in rules:
                try:
                    if rule.should_evaluate(evaluable):
                        result = rule.evaluate(evaluable, **rule.config)
                        self.results[evaluable][rule.__class__] = result
                except Exception as e:
                    self.results[evaluable][rule.__class__] = e

            self.scores[evaluable] = self._scorer.score_evaluable(
                self.results[evaluable]
            )
            self._formatter.evaluable_evaluated(
                evaluable, self.results[evaluable], self.scores[evaluable]
            )

        # Compute score for project
        self.project_score = self._scorer.score_aggregate_evaluables(
            list(self.scores.values())
        )

        # Add null check before calling project_evaluated
        if self._manifest_loader.models or self._manifest_loader.sources:
            self._formatter.project_evaluated(self.project_score)

__init__(rule_registry, manifest_loader, formatter, scorer)

Create an Evaluation object.

Parameters:

Name Type Description Default
rule_registry RuleRegistry

A rule registry to access rules.

 required
manifest_loader ManifestLoader

A manifest loader to access dbt metadata.

 required
formatter Formatter

A formatter to display results.

 required
scorer Scorer

A scorer to compute scores.

 required
Source code in src/dbt_score/evaluation.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(
    self,
    rule_registry: RuleRegistry,
    manifest_loader: ManifestLoader,
    formatter: Formatter,
    scorer: Scorer,
) -> None:
    """Create an Evaluation object.

    Args:
        rule_registry: A rule registry to access rules.
        manifest_loader: A manifest loader to access dbt metadata.
        formatter: A formatter to display results.
        scorer: A scorer to compute scores.
    """
    self._rule_registry = rule_registry
    self._manifest_loader = manifest_loader
    self._formatter = formatter
    self._scorer = scorer

    # For each evaluable, its results
    self.results: dict[Evaluable, EvaluableResultsType] = {}

    # For each evaluable, its computed score
    self.scores: dict[Evaluable, Score] = {}

    # The aggregated project score
    self.project_score: Score

evaluate()

Evaluate all rules.

Source code in src/dbt_score/evaluation.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def evaluate(self) -> None:
    """Evaluate all rules."""
    rules = self._rule_registry.rules.values()

    for evaluable in chain(
        self._manifest_loader.models, self._manifest_loader.sources
    ):
        # type inference on elements from `chain` is wonky
        # and resolves to superclass HasColumnsMixin
        evaluable = cast(Evaluable, evaluable)
        self.results[evaluable] = {}
        for rule in rules:
            try:
                if rule.should_evaluate(evaluable):
                    result = rule.evaluate(evaluable, **rule.config)
                    self.results[evaluable][rule.__class__] = result
            except Exception as e:
                self.results[evaluable][rule.__class__] = e

        self.scores[evaluable] = self._scorer.score_evaluable(
            self.results[evaluable]
        )
        self._formatter.evaluable_evaluated(
            evaluable, self.results[evaluable], self.scores[evaluable]
        )

    # Compute score for project
    self.project_score = self._scorer.score_aggregate_evaluables(
        list(self.scores.values())
    )

    # Add null check before calling project_evaluated
    if self._manifest_loader.models or self._manifest_loader.sources:
        self._formatter.project_evaluated(self.project_score)