Wymuszanie dokumentacji YAML w dbt za pomocą GitHub Actions

Wprowadzenie

Poniżej znajdziesz skrypt w Pythonie wykorzystujący bibliotekę pydantic. Skanuje wszystkie definicje YAML modeli i sprawdza ograniczenia pól, które definiujesz. Proponowana struktura to taka, która w moim doświadczeniu wymusza minimalną dokumentację, nie będąc przy tym zbyt restrykcyjna — ułatwia szybszy development, gdy jest to potrzebne. Możesz dostosować te reguły do własnych standardów organizacyjnych.

Wymagane importy

Oto lista bibliotek potrzebnych do tego zadania:

import glob # to iterate over your files
import pprint # to lint your printed errors
import re # to match with regular expressions
from pathlib import Path # to scan your files' paths
from typing import Any, Dict, Optional # to standardise checks input

import yaml # to evaluate your YAML files
from pydantic import ( # to execute model testing based on predefined schema
    BaseModel,
    Field,
    ValidationError,
    conlist,
    field_validator,
)

Budowa modelu bazowego

Pierwszą rzeczą jest nadanie pydanticowi „schematu”, którego będzie szukał podczas skanowania plików YAML. Ten schemat definiuje minimalne pola, które chcesz mieć w każdej definicji modelu.

Przykład:

class DBTModel(BaseModel):
    name: str = Field(min_length=3) # setting min length of name to 3 chars
    description: str = Field(min_length=10) # setting min length of description to 10chars
    meta: DBTModelMeta
    additional_args: Optional[Dict[str, Any]] = {}

Mogłeś zauważyć definicję meta: DBTModelMeta, która odpowiada osobno zbudowanej klasie. To świetne podejście do standaryzacji sekcji meta w dokumentacji YAML. Pozwala wymusić właściciela modelu, tagowanie lub inne wymiary. Jest niezwykle pomocne przy korzystaniu z dbt docs lub propagowaniu meta do np. Atlan. Definicja poniżej.

Ograniczenia dla meta

Prosty przykład standaryzacji tworzenia właściciela modelu oraz ograniczenia dozwolonych wartości. Możesz na tej podstawie rozbudować dalsze pola do sprawdzania.

1. Podajemy listę dozwolonych wartości — każdy właściciel w YAML musi pochodzić z tej listy.
2. Tworzymy osobną klasę do walidacji meta, gdzie niestandardowa funkcja Python sprawdza długość listy właścicieli oraz zgodność z dozwoloną listą. Zwróć uwagę na dekorator field_validator, który każe pydantic wykonać te sprawdzenia.

ALLOWED_DATA_DOMAINS_LIST = [
    "data-engineering",
    "data-science",
    "analytics-engineering",
    "business-analytics",
    "product-analytics",
]


# Define a Pydantic model for the data in schema .yml files
class DBTModelMeta(BaseModel):
    owners: conlist(str, min_length=1)

    @field_validator("owners")
    def validate_owner_team(cls, value):
        if len(value) > 1 and value[1] not in ALLOWED_DATA_DOMAINS_LIST:
            raise ValueError(
                f"Owner team should be one of the following: {ALLOWED_DATA_DOMAINS_LIST}"
            )
        return value

Możesz tworzyć kolejne klasy i rozszerzać sprawdzenia według potrzeb (np. osobna klasa do walidacji dokumentacji kolumn). Uwaga — jeśli potrzebujesz szybkich poprawek lub pilnego developmentu, a sprawdzenia są bardzo dokładne — będziesz miał wiele nieudanych checków CI i ewentualnie sfrustrowany zespół, zwłaszcza osoby rzadziej pracujące z dbt.

Sprawdzenie, czy wszystkie modele mają odpowiadające pliki YAML

To sprawdzenie działa, gdy stosujesz best practices — osobny plik YAML dla każdego modelu, o tej samej nazwie, ale innym rozszerzeniu. Jeśli trzymasz wszystko w jednym pliku lub inaczej — możesz ten fragment dostosować lub pominąć.

Przy tworzeniu uwzględniłem scenariusz z wersjonowaniem modeli dbt. Dla wersjonowanych modeli jest jeden plik YAML źródłowy. Uwzględniam to przez usuwanie postfixów _v<n> podczas iteracji po plikach.

Kroki:

1. Ustaw zmienną errors na false — ułatwia rzucanie błędów w całym skrypcie.
2. Ustaw wzorzec postfixu — dla wersjonowanych modeli.
3. Iteruj po wszystkich plikach .sql w katalogu models i usuń wzorzec postfixu.
4. Sprawdź, czy dla każdego znalezionego .sql istnieje plik YAML.
5. Rzuć błąd, jeśli brakuje definicji YAML.

errors = False

# Regular expression to match postfixes like _v1, _v2, etc.
postfix_pattern = re.compile(r"_v\d+$")

# Glob through all SQL files in the models directory
for sql_file in glob.glob(
    "<dbt_root_directory>/models/**/*.sql",
    recursive=True,
):
    sql_file_path = Path(sql_file)
    # Strip the postfix from the file name
    base_name = sql_file_path.stem
    base_name_no_postfix = postfix_pattern.sub("", base_name)
    yml_file_path = sql_file_path.with_name(base_name_no_postfix).with_suffix(
        ".yml"
    )
    if not yml_file_path.exists():
        errors = True
        print(f"Corresponding YAML file not found for: {sql_file_path}")

# If any SQL file is missing its corresponding YAML file
if errors:
    raise Exception(
        "Please ensure corresponding YAML files exist for all SQL files. They must have the exact same name as the SQL model file name."
    )

W tych sprawdzeniach jest wiele elementów. To działająca opcja z parametrami i sprawdzeniami opisanymi powyżej. Przy innej strukturze lub dalszych wymaganiach — zmodyfikuj skrypt.

Iteracja po wszystkich modelach YAML i walidacja pól

I teraz finał — pętla, która wykonuje sprawdzenia zdefiniowane w pydantic. Działa z jednym lub wieloma modelami w jednym pliku YAML. Pomija definicje sources i innych obiektów — sprawdza tylko modele. Błędy są sformatowane tak, by szybko zlokalizować problem. Kroki:

1. Iteruj po wszystkich modelach w plikach YAML w katalogu models.
2. Wczytaj pliki YAML do strumienia.
3. Iteruj po każdej definicji modelu i porównuj ze schematem z klas pydantic.
4. Rzuć błąd, jeśli którekolwiek pole nie pasuje.

# Glob through all schema .yml files in the models directory
for filename in glob.glob(
    "<dbt_root_directory>/models/**/*.yml", recursive=True
):
    with open(filename, "r") as stream:
        try:
            yaml_file = yaml.safe_load(stream)
        except yaml.YAMLError as exc:
            print(exc)

        try:
            yaml_models = yaml_file.get("models", [])
        # continuing the loop in case the YAML file is for sources or something else
        except Exception:
            continue

        # adding safety check in case more than one model are defined
        for (
            model
        ) in yaml_models:  # Assuming 'models' key contains a list of models
            try:
                schema = DBTModel(**model)
            except ValidationError as e:
                errors = True
                print("##############################")
                print(
                    f"ERROR in model: {model.get('name')} in file: {filename}"
                )
                for x in e.errors():
                    print(f"- Field \"{x['loc'][0]}\": {x['type']}")
                    print(f"\t - {x['msg']}")

# If any model is missing any required field
if errors:
    print("Expected Yaml schema for models:")
    pprint.pprint(DBTModel.model_fields, indent=3)
    raise Exception("Please update the models specified above.")

Pełny skrypt do uruchomienia

Poniżej wszystkie powyższe kroki połączone w skrypt wykonywalny. W następnej sekcji jest też fragment GitHub Action do automatyzacji uruchomienia. Skopiuj do własnego wdrożenia. Masz sugestie ulepszeń — zostaw komentarz poniżej!

import glob
import pprint
import re
from pathlib import Path
from typing import Any, Dict, Optional

import yaml
from pydantic import (
    BaseModel,
    Field,
    ValidationError,
    conlist,
    field_validator,
)

ALLOWED_DATA_DOMAINS_LIST = [
    "data-engineering",
    "data-science",
    "analytics-engineering",
    "business-analytics",
    "product-analytics",
]


# Define a Pydantic model for the data in schema .yml files
class DBTModelMeta(BaseModel):
    owners: conlist(str, min_length=1)

    @field_validator("owners")
    def validate_owner_team(cls, value):
        if len(value) > 1 and value[1] not in ALLOWED_DATA_DOMAINS_LIST:
            raise ValueError(
                f"Owner team should be one of the following: {ALLOWED_DATA_DOMAINS_LIST}"
            )
        return value

class DBTModel(BaseModel):
    name: str = Field(min_length=3)
    description: str = Field(min_length=10)
    meta: DBTModelMeta
    additional_args: Optional[Dict[str, Any]] = {}

errors = False

# Regular expression to match postfixes like _v1, _v2, etc.
postfix_pattern = re.compile(r"_v\d+$")

# Glob through all SQL files in the models directory
for sql_file in glob.glob(
    "<dbt_root_directory>/models/**/*.sql",
    recursive=True,
):
    sql_file_path = Path(sql_file)
    # Strip the postfix from the file name
    base_name = sql_file_path.stem
    base_name_no_postfix = postfix_pattern.sub("", base_name)
    yml_file_path = sql_file_path.with_name(base_name_no_postfix).with_suffix(
        ".yml"
    )
    if not yml_file_path.exists():
        errors = True
        print(f"Corresponding YAML file not found for: {sql_file_path}")

# If any SQL file is missing its corresponding YAML file
if errors:
    raise Exception(
        "Please ensure corresponding YAML files exist for all SQL files. They must have the exact same name as the SQL model file name."
    )


# Glob through all schema .yml files in the models directory
for filename in glob.glob(
    "<dbt_root_directory>/models/**/*.yml", recursive=True
):
    with open(filename, "r") as stream:
        try:
            yaml_file = yaml.safe_load(stream)
        except yaml.YAMLError as exc:
            print(exc)

        try:
            yaml_models = yaml_file.get("models", [])
        # continuing the loop in case the YAML file is for sources or something else
        except Exception:
            continue

        yaml_model_names = set()
        # adding safety check in case more than one model are defined
        for (
            model
        ) in yaml_models:  # Assuming 'models' key contains a list of models
            try:
                schema = DBTModel(**model)
            except ValidationError as e:
                errors = True
                print("##############################")
                print(
                    f"ERROR in model: {model.get('name')} in file: {filename}"
                )
                for x in e.errors():
                    print(f"- Field \"{x['loc'][0]}\": {x['type']}")
                    print(f"\t - {x['msg']}")

# If any model is missing any required field
if errors:
    print("Expected Yaml schema for models:")
    pprint.pprint(DBTModel.model_fields, indent=3)
    raise Exception("Please update the models specified above.")

Implementacja w GitHub Action

Prosta definicja, jak powyższe sprawdzenia wdrożyć w pipeline GitHub Actions. Część kroków jest opcjonalna, ale uszczupla wdrożenie i oszczędza minuty GitHub Actions, uruchamiając workflow tylko gdy jest to potrzebne. Kroki:

1. (Opcjonalnie) dodaj workflow_dispatch do ręcznego uruchomienia.
2. (Opcjonalnie) określ typy pull requestów, przy których workflow się uruchomi.
3. (Opcjonalnie) określ gałęzie.
4. (Opcjonalnie) określ ścieżki zmian — workflow tylko przy zmianie w podanej ścieżce.
5. (Opcjonalnie) określ concurrency — co się dzieje przy kolejnym pushu podczas trwającego joba. Tutaj anulujemy bieżący i startujemy nowy.
6. Określ środowisko uruchomienia; u mnie ubuntu-latest.
7. Checkout kodu — natywna akcja GitHub.
8. Skonfiguruj Pythona — natywna akcja GitHub.
9. Zainstaluj zależności.
10. Uruchom zdefiniowany wcześniej skrypt.

name: Validate model metadata

on:
 workflow_dispatch:
 pull_request:
   types:
     - opened
     - reopened
     - synchronize
   branches:
     - main
   paths:
     - '<your_dbt_path>/models/**/*.yml'

concurrency:
  group: "validate-model-metadata-pr-${{ github.event.pull_request.number }}"
  cancel-in-progress: true

jobs:
 run-python-script:
   runs-on: ubuntu-latest

   steps:
   - name: Check out code
     uses: actions/checkout@v4

   - name: Set up Python
     uses: actions/setup-python@v5
     with:
       python-version: '3.x'

   - name: Install required dependencies
     run: |
       python -m pip install --upgrade pip
       pip install -U pyyaml
       pip install -U pydantic
       pip install -U typing

   - name: Validate model metadata in schema.yml
     run: python <path_to_your_executable_file>

Kontakt

Dziękuję za przeczytanie. Podoba Ci się przekazywana wiedza, ale brakuje czasu lub kompetencji, aby uporządkować analytics engineering? Sprawdź moje dane kontaktowe.

• Piotr Siemiński | Linktree
• Lortech Solutions

Wymuszanie dokumentacji YAML w dbt za pomocą GitHub Actions został pierwotnie opublikowany w Lortech Solutions Blog na Medium, gdzie rozmowa trwa dalej dzięki podświetleniom i odpowiedziom czytelników.