Skip to content

Manifest formatter

Formatter for a manifest.json.

ManifestFormatter

Bases: Formatter

Formatter to generate manifest.json with score metadata.

Source code in src/dbt_score/formatters/manifest_formatter.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class ManifestFormatter(Formatter):
    """Formatter to generate manifest.json with score metadata."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Instantiate a manifest formatter."""
        self._evaluable_scores: dict[str, Score] = {}
        super().__init__(*args, **kwargs)

    def evaluable_evaluated(
        self, evaluable: Evaluable, results: EvaluableResultsType, score: Score
    ) -> None:
        """Callback when an evaluable item has been evaluated."""
        self._evaluable_scores[evaluable.unique_id] = score

    def project_evaluated(self, score: Score) -> None:
        """Callback when a project has been evaluated."""
        manifest = copy.copy(self._manifest_loader.raw_manifest)
        for evaluable_id, evaluable_score in self._evaluable_scores.items():
            if evaluable_id.startswith("model") or evaluable_id.startswith("snapshot"):
                nodes_manifest = manifest["nodes"][evaluable_id]
                nodes_manifest["meta"]["score"] = evaluable_score.value
                nodes_manifest["meta"]["badge"] = evaluable_score.badge
            if evaluable_id.startswith("source"):
                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))

__init__(*args, **kwargs)

Instantiate a manifest formatter.

Source code in src/dbt_score/formatters/manifest_formatter.py 
16
17
18
19
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Instantiate a manifest formatter."""
    self._evaluable_scores: dict[str, Score] = {}
    super().__init__(*args, **kwargs)

evaluable_evaluated(evaluable, results, score)

Callback when an evaluable item has been evaluated.

Source code in src/dbt_score/formatters/manifest_formatter.py 
21
22
23
24
25
def evaluable_evaluated(
    self, evaluable: Evaluable, results: EvaluableResultsType, score: Score
) -> None:
    """Callback when an evaluable item has been evaluated."""
    self._evaluable_scores[evaluable.unique_id] = score

project_evaluated(score)

Callback when a project has been evaluated.

Source code in src/dbt_score/formatters/manifest_formatter.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def project_evaluated(self, score: Score) -> None:
    """Callback when a project has been evaluated."""
    manifest = copy.copy(self._manifest_loader.raw_manifest)
    for evaluable_id, evaluable_score in self._evaluable_scores.items():
        if evaluable_id.startswith("model") or evaluable_id.startswith("snapshot"):
            nodes_manifest = manifest["nodes"][evaluable_id]
            nodes_manifest["meta"]["score"] = evaluable_score.value
            nodes_manifest["meta"]["badge"] = evaluable_score.badge
        if evaluable_id.startswith("source"):
            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))