feat(doc): add script to override dbt docs

[AxelThevenot]

Add a script to overide dbt docs generate results (from manifest.json ?) in order to add documentation about the assertions in the .yml definition which are ignored by dbt.

This can be a simple Python script.

I think a resulting Markdown table will be a good idea.

Example

assertion	description	expression	null_as_error
unique	Rows must be unique over columns (key_1, key_2)	n/a	false
open_date_is_past	Open date must be in the past	open_date <= CURRENT_DATE()	false
...	...	...	...

[AxelThevenot]

Why not to add some gitlab/github/cloudbuild CI to execute this script

[AxelThevenot]

import os
import json

manifest_file_path = 'target/manifest.json'

def _extract_columns(columns, parent_column=None, depends_on=None):

    if not columns:
        return {}
    
    result = {
        parent_column: {'columns': [], 'depends_on': depends_on or []}
    }

    for column in columns:

        if isinstance(column, str):
            result[parent_column]['columns'].append(column)
        
        elif isinstance(column, dict):

            for nested_colum, sub_columns in column.items():
                result.update(_extract_columns(
                        columns=sub_columns,
                        parent_column=nested_colum,
                        depends_on=depends_on + [parent_column] if depends_on is not None else []
                    )
                )

    return result

def _build_unique_description(unique_columns):

    result = {}
    
    layered_unique_columns = _extract_columns(unique_columns)

    for parent_column, layer_unique_columns in layered_unique_columns.items():

        columns = layer_unique_columns['columns']
        depends_on = layer_unique_columns['depends_on']

        if parent_column is None:
            result.update({
                'unique': {
                    'description': 'Row must be unique over the unique keys.',
                    'expression': ', '.join(columns),
                }
            })

        else:

            parent_unique_columns = []
            for parent_column in depends_on:
                parent_unique_columns.extend(columns[parent_column]['columns'])

            result.update({
                '.'.join(depends_on + [parent_column]) + '__unique': {
                    'description': 'Row must be unique over the unique keys.',
                    'expression': ', '.join([*parent_unique_columns, *columns]),
                }
            })

    return result

def _build_not_null_description(not_null_columns):

    result = {}

    layered_not_null_columns = _extract_columns(not_null_columns)

    for parent_column, layer_not_null_columns in layered_not_null_columns.items():

        columns = layer_not_null_columns['columns']
        depends_on = layer_not_null_columns['depends_on']

        if parent_column is None:

            for column in columns:

                result.update({
                    column + '__not_null': {
                        'description': column + ' is not null.',
                        'expression': column + ' IS NOT NULL'
                    }
                })

        else:

            for column in columns:

                result.update({
                    '.'.join(depends_on + [parent_column, column]) + ' __unique': {
                        'description': '.'.join(depends_on + [parent_column, column]) + ' are not null within the row.',
                        'expression': '',
                    }
                })

    return result

def build_description(assertions):

    unique_columns = assertions.pop('__unique__', None)
    not_null_columns = assertions.pop('__not_null__', None)
    if not_null_columns and '__unique__' in not_null_columns:
        not_null_columns = unique_columns.copy()

    table_md = '\n'.join((
        '',
        '|assertions|description|expression|null as exception|',
        '|----------|-----------|----------|-----------------|',
    ))
    assertions = {
        **_build_unique_description(unique_columns),
        **_build_not_null_description(not_null_columns),
        **assertions,
    }

    for assertion_id, assertion in assertions.items():
        description = assertion.get('description', '')
        expression = assertion.get('expression', '').replace('\n', ' ')
        null_as_exception = assertion.get('null_as_exception', '')
        table_md += f'\n|{assertion_id}|{description}|{expression}|{null_as_exception}|'

    return table_md

if __name__ == '__main__':

    with open(manifest_file_path, 'r') as f:
        manifest = json.load(f)

    nodes = manifest.get('nodes', {})
    sources = manifest.get('sources', {})

    for resource_type, resources in (('nodes', nodes), ('sources', sources)):
        for node_id, node in resources.items():

            columns = node.get('columns', {})

            for column_name, column in columns.items():
                assertions = column.get('assertions', [])

                if not assertions:
                    continue

                assertions_description = build_description(assertions)
                # print(assertions_description)
                manifest[resource_type][node_id]['columns'][column_name]['description'] = assertions_description
    
    with open(manifest_file_path, 'w') as f:
        json.dump(manifest, f)