dbt-artifacts-parser

This is a dbt artifacts parser in python. It enables us to deal with catalog.json, manifest.json, run-results.json and sources.json as python objects.

This package is primarily designed for dbt-core, enabling seamless interaction with dbt artifacts as Python objects. While dbt Cloud provides additional artifact types beyond those of dbt-core, this package does not offer comprehensive support for them.

Supported Versions and Compatibility

⚠️ Important Note:

• Pydantic v1 will not be supported for dbt 1.9 or later.
• To parse dbt 1.9 or later, please migrate your code to pydantic v2.
• We will reassess version compatibility upon the release of pydantic v3.

Installation

Requires Python 3.10 or newer. Tested on CPython 3.10, 3.11, 3.12, and 3.13.

pip install -U dbt-artifacts-parser

Development dependencies are declared as a PEP 735 dev dependency group in pyproject.toml (locked in uv.lock) and mirrored under [project.optional-dependencies].dev so pip install dbt-artifacts-parser[dev] keeps working. After cloning, run make setup or bash dev/setup.sh --deps development (uses uv sync). To add a dev tool: uv add --group dev <package>, then update [project.optional-dependencies].dev to match, and run uv lock.

Python classes

Those are the classes to parse dbt artifacts.

Catalog

• CatalogV1 for catalog.json v1

Manifest

• ManifestV1 for manifest.json v1
• ManifestV2 for manifest.json v2
• ManifestV3 for manifest.json v3
• ManifestV4 for manifest.json v4
• ManifestV5 for manifest.json v5
• ManifestV6 for manifest.json v6
• ManifestV7 for manifest.json v7
• ManifestV8 for manifest.json v8
• ManifestV9 for manifest.json v9
• ManifestV10 for manifest.json v10
• ManifestV11 for manifest.json v11
• ManifestV12 for manifest.json v12

Run Results

• RunResultsV1 for run_results.json v1
• RunResultsV2 for run_results.json v2
• RunResultsV3 for run_results.json v3
• RunResultsV4 for run_results.json v4
• RunResultsV5 for run_results.json v5
• RunResultsV6 for run_results.json v6

Sources

• SourcesV1 for sources.json v1
• SourcesV2 for sources.json v2
• SourcesV3 for sources.json v3

Examples

Parse catalog.json

import json

# parse any version of catalog.json
from dbt_artifacts_parser.parser import parse_catalog

with open("path/to/catalog.json", "r") as fp:
    catalog_dict = json.load(fp)
    catalog_obj = parse_catalog(catalog=catalog_dict)

# parse catalog.json v1
from dbt_artifacts_parser.parser import parse_catalog_v1

with open("path/to/catalog.json", "r") as fp:
    catalog_dict = json.load(fp)
    catalog_obj = parse_catalog_v1(catalog=catalog_dict)

Parse manifest.json

import json

# parse any version of manifest.json
from dbt_artifacts_parser.parser import parse_manifest

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest(manifest=manifest_dict)

# Opt-in best-effort parse when the schema version is newer than this
# package supports. Fields forbidden by the latest schema are dropped;
# nested extras allowed by the schema (e.g. custom dbt config keys) are
# kept. Prefer refreshing parsers for full support of new versions.
manifest_obj = parse_manifest(manifest=manifest_dict, fallback_to_latest=True)

# parse manifest.json v1
from dbt_artifacts_parser.parser import parse_manifest_v1

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v1(manifest=manifest_dict)

# parse manifest.json v2
from dbt_artifacts_parser.parser import parse_manifest_v2

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v2(manifest=manifest_dict)

# parse manifest.json v3
from dbt_artifacts_parser.parser import parse_manifest_v3

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v3(manifest=manifest_dict)

# parse manifest.json v4
from dbt_artifacts_parser.parser import parse_manifest_v4

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v4(manifest=manifest_dict)

# parse manifest.json v5
from dbt_artifacts_parser.parser import parse_manifest_v5

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v5(manifest=manifest_dict)

# parse manifest.json v6
from dbt_artifacts_parser.parser import parse_manifest_v6

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v6(manifest=manifest_dict)

# parse manifest.json v7
from dbt_artifacts_parser.parser import parse_manifest_v7

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v7(manifest=manifest_dict)

# parse manifest.json v8
from dbt_artifacts_parser.parser import parse_manifest_v8

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v8(manifest=manifest_dict)

# parse manifest.json v9
from dbt_artifacts_parser.parser import parse_manifest_v9

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v9(manifest=manifest_dict)

# parse manifest.json v10
from dbt_artifacts_parser.parser import parse_manifest_v10

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v10(manifest=manifest_dict)

# parse manifest.json v11
from dbt_artifacts_parser.parser import parse_manifest_v11

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v11(manifest=manifest_dict)

# parse manifest.json v12
from dbt_artifacts_parser.parser import parse_manifest_v12

with open("path/to/manifest.json", "r") as fp:
    manifest_dict = json.load(fp)
    manifest_obj = parse_manifest_v12(manifest=manifest_dict)

Parse run-results.json

import json

# parse any version of run-results.json
from dbt_artifacts_parser.parser import parse_run_results

with open("path/to/run-results.json", "r") as fp:
    run_results_dict = json.load(fp)
    run_results_obj = parse_run_results(run_results=run_results_dict)

# parse run-results.json v1
from dbt_artifacts_parser.parser import parse_run_results_v1

with open("path/to/run-results.json", "r") as fp:
    run_results_dict = json.load(fp)
    run_results_obj = parse_run_results_v1(run_results=run_results_dict)

# parse run-results.json v2
from dbt_artifacts_parser.parser import parse_run_results_v2

with open("path/to/run-results.json", "r") as fp:
    run_results_dict = json.load(fp)
    run_results_obj = parse_run_results_v2(run_results=run_results_dict)

# parse run-results.json v3
from dbt_artifacts_parser.parser import parse_run_results_v3

with open("path/to/run-results.json", "r") as fp:
    run_results_dict = json.load(fp)
    run_results_obj = parse_run_results_v3(run_results=run_results_dict)

# parse run-results.json v4
from dbt_artifacts_parser.parser import parse_run_results_v4

with open("path/to/run-results.json", "r") as fp:
    run_results_dict = json.load(fp)
    run_results_obj = parse_run_results_v4(run_results=run_results_dict)

# parse run-results.json v5
from dbt_artifacts_parser.parser import parse_run_results_v5

with open("path/to/run-results.json", "r") as fp:
    run_results_dict = json.load(fp)
    run_results_obj = parse_run_results_v5(run_results=run_results_dict)

# parse run-results.json v6
from dbt_artifacts_parser.parser import parse_run_results_v6

with open("path/to/run-results.json", "r") as fp:
    run_results_dict = json.load(fp)
    run_results_obj = parse_run_results_v6(run_results=run_results_dict)

Parse sources.json

import json

# parse any version of sources.json
from dbt_artifacts_parser.parser import parse_sources

with open("path/to/sources.json", "r") as fp:
    sources_dict = json.load(fp)
    sources_obj = parse_sources(sources=sources_dict)

# parse sources.json v1
from dbt_artifacts_parser.parser import parse_sources_v1

with open("path/to/sources.json", "r") as fp:
    sources_dict = json.load(fp)
    sources_obj = parse_sources_v1(sources=sources_dict)

# parse sources.json v2
from dbt_artifacts_parser.parser import parse_sources_v2

with open("path/to/sources.json", "r") as fp:
    sources_dict = json.load(fp)
    sources_obj = parse_sources_v2(sources=sources_dict)

# parse sources.json v3
from dbt_artifacts_parser.parser import parse_sources_v3

with open("path/to/sources.json", "r") as fp:
    sources_dict = json.load(fp)
    sources_obj = parse_sources_v3(sources=sources_dict)

Contributors

Thank you for your contributions!

If you are interested in contributing to this package, please check out the CONTRIBUTING.md.

Yu Ishikawa	Null	Brent Brewington	Dinis Louseiro	Jairus Martinez	Judah Rand
Kilian Tscherny	Matteo Renoldi	Mayur Singal	Mohamed Nabil Mahmoud Hafez	Onkar Ravgan	Pádraic Slattery
asafalina	Null