Lint your dbt metadata with dbt-score

Picnic has recently open-sourced dbt-score, a linter for your dbt metadata. In this post I walk through some of its features and how you can incorporate it in your dbt project and CI/CD pipelines.

Setup

Setting up dbt-score is straightforward. We start with a pip install, and include it in our requirements.txt file.

pip install dbt-score
pip freeze > requirements.txt

Although not required, it is advised to also include dbt-score in your pyproject.toml file for more granular control. An example setup below.

[tool.dbt-score]
rule_namespaces = ["dbt_score.rules", "rules"]
disabled_rules = ["dbt_score.rules.generic.has_owner"]
inject_cwd_in_python_path = true

[tool.dbt-score.badges]
first.threshold = 10.0
first.icon = "🥇"
second.threshold = 8.0
second.icon = "🥈"
third.threshold = 6.0
third.icon = "🥉"
wip.icon = "🏗️"

[tool.dbt-score.rules."dbt_score.rules.generic.sql_has_reasonable_number_of_lines"]
severity = 1
max_lines = 300

Custom rules

One of the most powerful features of dbt-score is the ability to create your own custom rules. As the toml above shows, I assigned therules folder for this.

For my project, I wanted to check the naming conventions of the models within the various layers. To do this, I created the rules/ModelNameSpecification.py file, with the code below.

from dbt_score import Model, rule, RuleViolation, Severity

@rule(description="A model should follow one of the specified formats.", severity=Severity.HIGH)
def evaluate(model: Model) -> RuleViolation | None:
    """Evaluate the rule."""

    if model.schema.endswith('mart'):
        if not model.name.startswith('fct__') and not model.name.startswith('dim__'):
            return RuleViolation(message="Model name not compliant.")

    elif model.schema.endswith('int'):
        if not model.name.startswith('int_'):
            return RuleViolation(message="Model name not compliant.")
        
    elif model.schema.endswith('stg'):
        if not model.name.startswith('stg_') and not model.name.startswith('base_'):
            return RuleViolation(message="Model name not compliant.")

This rule makes use of the dbt-score @rule decorator. Though you could opt to inherit from the Rule class, this is the simpler & preferred way.
This rule does some (simplified) checks on the model naming conventions, based on the schema the model is in. Since we think conventions are important, the severity is set to high.
Note: you can have as many rules as you want!

CI/CD Integration

The next step is to automatically run the linter. You could achieve this through a pre-commit hook, or in your CI/CD pipeline. We use bitbucket pipelines, so that is what’s used in this example. The method should be portable.

pipelines:
  branches:
    main:
      - step:
          name: 'Lint & Test'
          image: python:3.10-slim
          script:
            - pip install -r requirements.txt
            - dbt deps
            - dbt-score lint --run-dbt-parse -f manifest > lint.json
            - python -m pytest tests/documentation/test.py --junitxml=./test-reports/unit.xml

A breakdown of what happens:

All the commands are run on the python:3.10-slim image
We first install the required libraries from the requirements file
Then, we install the necessary dbt dependencies
We run the dbt-score lint CLI command, and make sure to export the results as json to the lint.json file in the project root, by adding the -f manifest flag. We also parse the dbt project through the — run-dbt-parse flag. Note that all the required environment variables need to be available.
Lastly, we run pytest, and export the test results in junit XML, to make the results available in the bitbucket pipeline interface. An example of a simple test can be found below

import json
import pytest

with open('lint.json') as json_file:
    data = json.load(json_file)

def test_total_score():
    # Calculate scores
    scores = []
    for node in data['nodes']:
        try:
            scores.append(data['nodes'][node]['meta']['score'])
        except KeyError:
            next

    total_score = sum(scores) / len(scores)
    assert total_score >= 9

This example test grabs the result from the generated lint results, calculates the final score, and checks if it’s ≥ 9.

Success!

Success!

In conclusion

dbt-score is a simple, yet powerful package that makes it easy to manage your dbt model metadata. Shoutout to Jochem van Dooren and colleagues for sharing it with the community!

Hi, I’m Bastiaan 👋🏼 Data Lead at a scale-up. I write about the Modern Data Workflow, where I explore tools & processes to supercharge your data capabilities. Follow me for more!