#!/usr/bin/env python3 """Read dbt artifacts and report tests, test results and freshness on critical nodes. Usage: python3 scripts/check_dbt_artifacts.py MANIFEST.json [--run-results RUN_RESULTS.json] [--sources SOURCES_OR_FRESHNESS.json] [--config AUDIT_CONFIG.json] [--critical tag:critical | name1,name2] [--adapter bigquery|snowflake] [--as-of YYYY-MM-DD] [--json] Inputs are files from target/ or from a dbt platform job's artifacts. Never run dbt against production to produce them for the audit. - manifest.json (dbt Core 1.x, schema v9 or later; unknown keys ignored) - run_results.json (unique_id, status, failures) - sources.json or freshness.json (unique_id, max_loaded_at_time_ago_in_s, status, criteria). freshness.json writes Pass, Warn and Error and leaves out a source whose query failed, so a missing result is printed as such Critical nodes are the models the audited outputs read and the sources behind them: - --config takes the model names from outputs[].models in the audit config (the preferred route: tags in a project rarely match what the outputs actually read) - --critical takes tag:NAME (tags or config.tags) or a comma-separated list of model names and source_name.table_name source names - with neither, tag:critical is used - every source a critical model reads, directly or through other models (depends_on), is added to the critical sources, tagged or not - a name that is not in the manifest is listed and makes OPS-01 (models) or FRS-04 (sources) NOT CHECKED unless something else already fails Per critical model: tests attached to it (attached_node, else the model in depends_on), whether the grain is tested (unique or unique_combination_of_columns), whether every grain column has not_null, whether accepted_values and relationships exist, the latest result of each test, and identifier-like columns without not_null (a note, not a grain test). Per critical source: freshness thresholds, and what the check reads. No loaded_at_field or loaded_at_query means dbt uses warehouse metadata where it can; on BigQuery a wildcard identifier such as events_* makes that unreliable. A loaded_at_field that is an ingestion column (_fivetran_synced, _loaded_at and similar) measures the load, not the data. Suggested statuses: OPS-01 (tests), OPS-02 (results), FRS-04 (freshness signal). OPS-02 is NOT CHECKED when any test on a critical model has no result in the artifact: a dbt run, a selection or a later step overwrites run_results.json, so a clean file can hold no test at all. It cannot see job schedules, owners or what a test protects; those stay with the agent. Standard library only. Exit status: 0 when no suggested status is FAIL, 1 otherwise, 2 on unreadable artifacts. """ import argparse import json import re import sys from collections import OrderedDict from datetime import date, datetime, timedelta GRAIN_TESTS = {"unique", "unique_combination_of_columns", "expect_compound_columns_to_be_unique", "expect_column_values_to_be_unique"} NOT_NULL_TESTS = {"not_null", "expect_column_values_to_not_be_null"} INGESTION_FIELD = re.compile( r"^_fivetran_|^_airbyte_|^_sdc_|^_dlt_|(^|_)loaded_at$|_synced(_at)?$|(^|_)ingested_at$|^_etl_", re.I) IDENTIFIER_COLUMN = re.compile(r"(_id$|^gclid$|^gbraid$|^wbraid$|^fbclid$|^msclkid$|^li_fat_id$|^ttclid$)") CLICK_IDS = {"gclid", "gbraid", "wbraid", "fbclid", "msclkid", "li_fat_id", "ttclid"} STALE_AFTER_DAYS = 2 class InputError(Exception): pass def load_json(path, what): try: with open(path, encoding="utf-8-sig") as handle: return json.load(handle) except (OSError, ValueError) as exc: raise InputError("cannot read %s %s: %s" % (what, path, exc)) def load_results(path, what): """A run_results or freshness artifact: a JSON object with a results list.""" data = load_json(path, what) if not isinstance(data, dict) or not isinstance(data.get("results"), list): raise InputError("%s has no results list; is it the %s artifact?" % (path, what)) return data def tags_of(node): tags = list(node.get("tags") or []) for t in (node.get("config") or {}).get("tags") or []: if t not in tags: tags.append(t) return tags def generated(meta): text = (meta or {}).get("generated_at") or "" try: return datetime.strptime(text[:19], "%Y-%m-%dT%H:%M:%S") except ValueError: return None def ref_target(kwargs): m = re.search(r"ref\(\s*['\"]([^'\"]+)['\"]", str((kwargs or {}).get("model", ""))) return m.group(1) if m else None def threshold(fresh, key): item = (fresh or {}).get(key) or {} if item.get("count") and item.get("period"): return "%s %s" % (item["count"], item["period"]) return None def upstream_sources(start, nodes, sources): """Map each source to the start models that read it, directly or through other nodes.""" found = {} for origin in start: seen, stack = set(), [origin] while stack: uid = stack.pop() if uid in seen: continue seen.add(uid) for dep in ((nodes.get(uid) or {}).get("depends_on") or {}).get("nodes") or []: if dep in sources: found.setdefault(dep, []) if origin not in found[dep]: found[dep].append(origin) elif (nodes.get(dep) or {}).get("resource_type") in ("model", "snapshot", "seed"): stack.append(dep) return found def config_models(path): config = load_json(path, "audit config") names = [] for out in (config.get("outputs") or []) if isinstance(config, dict) else []: models = out.get("models") if isinstance(out, dict) else None for name in [models] if isinstance(models, str) else models or []: name = str(name).strip() if name and name not in names: names.append(name) if not names: raise InputError("%s lists no outputs[].models" % path) return names def analyse(args): manifest = load_json(args.manifest, "manifest") if not isinstance(manifest, dict) or "nodes" not in manifest: raise InputError("%s has no nodes; is it a manifest.json?" % args.manifest) meta = manifest.get("metadata") or {} adapter = (args.adapter or meta.get("adapter_type") or "").lower() nodes = manifest.get("nodes") or {} sources = manifest.get("sources") or {} model_by_name = dict((n.get("name"), u) for u, n in nodes.items() if n.get("resource_type") == "model") source_by_name = dict(("%s.%s" % (n.get("source_name"), n.get("name")), u) for u, n in sources.items()) critical = args.critical or ("" if args.config else "tag:critical") selection, tag, names = [], None, [] if args.config: selection.append("audit config outputs[].models") names += [(x, "config") for x in config_models(args.config)] if critical.startswith("tag:"): tag = critical[4:] selection.append(critical) elif critical: selection.append("named in --critical") names += [(x.strip(), "critical") for x in critical.split(",") if x.strip()] crit_models, crit_sources = [], [] if tag: crit_models = [u for u, n in nodes.items() if n.get("resource_type") == "model" and tag in tags_of(n)] crit_sources = [u for u, n in sources.items() if tag in tags_of(n)] missing_models, missing_sources = [], [] for name, origin in names: if origin == "critical" and name in source_by_name: if source_by_name[name] not in crit_sources: crit_sources.append(source_by_name[name]) continue uid = model_by_name.get(name) or model_by_name.get(name.split(".")[-1]) if uid: if uid not in crit_models: crit_models.append(uid) elif origin == "critical" and "." in name: missing_sources.append(name) elif name not in missing_models: missing_models.append(name) read_by = upstream_sources(crit_models, nodes, sources) selected = set(crit_sources) crit_sources += [u for u in sources if u in read_by and u not in selected] tests_by_model = OrderedDict((u, []) for u in crit_models) for uid, n in nodes.items(): if n.get("resource_type") != "test": continue tm = n.get("test_metadata") or {} kwargs = tm.get("kwargs") or {} target = n.get("attached_node") if not target: target = model_by_name.get(ref_target(kwargs)) if not target: target = next((d for d in (n.get("depends_on") or {}).get("nodes", []) if d.startswith("model.")), None) if target not in tests_by_model: continue kind = tm.get("name") or "singular" columns = [] if kind in ("unique_combination_of_columns", "expect_compound_columns_to_be_unique"): columns = list(kwargs.get("combination_of_columns") or kwargs.get("column_list") or []) else: col = n.get("column_name") or kwargs.get("column_name") if col: columns = [col] severity = str((n.get("config") or {}).get("severity") or "ERROR") tests_by_model[target].append({"uid": uid, "kind": kind, "columns": columns, "severity": severity}) results = {} run_meta = None if args.run_results: rr = load_results(args.run_results, "run results") run_meta = rr.get("metadata") or {} for r in rr.get("results") or []: results[r.get("unique_id")] = r fresh_results = {} fresh_meta = None if args.sources: sj = load_results(args.sources, "freshness results") fresh_meta = sj.get("metadata") or {} for r in sj.get("results") or []: fresh_results[r.get("unique_id")] = r models = [] for uid in crit_models: n = nodes[uid] tests = tests_by_model[uid] grain = [t for t in tests if t["kind"] in GRAIN_TESTS] grain_cols = [] for t in grain: for c in t["columns"]: if c not in grain_cols: grain_cols.append(c) not_null_cols = set(c for t in tests if t["kind"] in NOT_NULL_TESTS for c in t["columns"]) missing_not_null = [c for c in grain_cols if c not in not_null_cols] kinds = set(t["kind"] for t in tests) failing, warning = [], [] for t in tests: r = results.get(t["uid"]) if not r: continue status = str(r.get("status") or "").lower() entry = "%s on %s: %s, %s failures" % (t["kind"], ", ".join(t["columns"]) or "model", status, r.get("failures") if r.get("failures") is not None else "no") if status in ("fail", "error"): failing.append(entry) elif status == "warn": warning.append(entry) model_result = results.get(uid) model_status = str((model_result or {}).get("status") or "").lower() if model_result else None notes = [] if tests: for col in (n.get("columns") or {}): if IDENTIFIER_COLUMN.search(col) and col not in not_null_cols and col not in grain_cols: if col in CLICK_IDS: notes.append("%s has no not_null test (not a grain test; click IDs can be empty when " "gbraid or wbraid is used, so measure coverage with IDN-01)" % col) else: notes.append("%s has no not_null test" % col) models.append(OrderedDict([ ("name", n.get("name")), ("tests", ["%s (%s%s)" % (t["kind"], ", ".join(t["columns"]) or "model", ", severity warn" if t["severity"].lower() == "warn" else "") for t in tests]), ("grain_columns", grain_cols), ("has_grain_test", bool(grain)), ("missing_not_null_on_grain", missing_not_null), ("has_accepted_values", "accepted_values" in kinds), ("has_relationships", "relationships" in kinds), ("failing_tests", failing), ("warning_tests", warning), ("model_status", model_status), ("notes", notes)])) srcs = [] for uid in crit_sources: n = sources[uid] cfg = n.get("config") or {} fresh = n.get("freshness") or cfg.get("freshness") or {} warn_t, error_t = threshold(fresh, "warn_after"), threshold(fresh, "error_after") field = n.get("loaded_at_field") or cfg.get("loaded_at_field") query = n.get("loaded_at_query") or cfg.get("loaded_at_query") identifier = n.get("identifier") or n.get("name") if not warn_t and not error_t: signal, verdict = "no freshness thresholds, so dbt does not check this source", "FAIL" elif not field and not query: signal, verdict = "no loaded_at_field or loaded_at_query, so dbt uses warehouse metadata", "FAIL" if adapter == "bigquery" and "*" in str(identifier): signal += "; metadata freshness on a BigQuery wildcard identifier is unreliable" elif field and INGESTION_FIELD.search(field): signal, verdict = "loaded_at_field %s is an ingestion timestamp (measures the load, not the data)" % field, "WARN" else: signal, verdict = "reads %s" % (("loaded_at_field " + field) if field else "a loaded_at_query"), "PASS" r = fresh_results.get(uid) result = None result_status = str(r.get("status") or "").lower() if r else None if r: age = r.get("max_loaded_at_time_ago_in_s") result = "%s, %s hours old at snapshot" % (result_status, round(float(age) / 3600, 1) if age is not None else "unknown") elif args.sources and (warn_t or error_t): result = "no result in the artifact (dbt leaves out a source whose freshness query failed)" srcs.append(OrderedDict([ ("name", "%s.%s" % (n.get("source_name"), n.get("name"))), ("identifier", identifier), ("selected_by", "tag or name" if uid in selected else "depends_on"), ("read_by", [nodes[m].get("name") for m in read_by.get(uid, [])]), ("warn_after", warn_t), ("error_after", error_t), ("signal", signal), ("signal_status", verdict), ("result", result), ("result_status", result_status)])) stale = [] as_of = date.fromisoformat(args.as_of) if args.as_of else date.today() for label, m in (("manifest", meta), ("run results", run_meta), ("freshness results", fresh_meta)): if m is None: continue g = generated(m) if g is None: stale.append("%s has no readable generated_at" % label) elif g.date() < as_of - timedelta(days=STALE_AFTER_DAYS): stale.append("%s generated %s, more than %d days before %s" % (label, g.date(), STALE_AFTER_DAYS, as_of)) statuses = OrderedDict() reasons = OrderedDict() no_grain = [m["name"] for m in models if not m["has_grain_test"]] no_nn = ["%s (%s)" % (m["name"], ", ".join(m["missing_not_null_on_grain"])) for m in models if m["has_grain_test"] and m["missing_not_null_on_grain"]] selection_text = " + ".join(selection) if not models: statuses["OPS-01"], reasons["OPS-01"] = "NOT CHECKED", "no critical models found for %s" % selection_text elif no_grain: statuses["OPS-01"] = "FAIL" reasons["OPS-01"] = "no grain test on %s" % ", ".join( "%s%s" % (x, " (no tests at all)" if not next(m for m in models if m["name"] == x)["tests"] else "") for x in no_grain) elif no_nn: statuses["OPS-01"], reasons["OPS-01"] = "WARN", "grain columns without not_null: %s" % "; ".join(no_nn) else: statuses["OPS-01"], reasons["OPS-01"] = "PASS", "every critical model tests its grain and key not_null" if missing_models: note = ("not in the manifest: %s (find where each is built: a raw connector table makes OPS-01 N/A " "for it, a table built outside dbt has no dbt tests)" % ", ".join(missing_models)) if statuses["OPS-01"] == "PASS": statuses["OPS-01"], reasons["OPS-01"] = "NOT CHECKED", note + "; the models found pass" elif statuses["OPS-01"] == "NOT CHECKED": reasons["OPS-01"] = note else: reasons["OPS-01"] += "; also NOT CHECKED, " + note test_ids = [t["uid"] for u in crit_models for t in tests_by_model[u]] unseen = len([t for t in test_ids if t not in results]) unseen_note = ("%d of %d tests on critical models have no result in this artifact, which may come from a " "dbt run, a selection or another job step" % (unseen, len(test_ids))) if not args.run_results: statuses["OPS-02"], reasons["OPS-02"] = "NOT CHECKED", "no run_results.json given" else: fails = ["%s: %s" % (m["name"], x) for m in models for x in m["failing_tests"]] errored = ["%s %s" % (m["name"], m["model_status"]) for m in models if m["model_status"] in ("error", "skipped")] warns = ["%s: %s" % (m["name"], x) for m in models for x in m["warning_tests"]] run_stale = [s for s in stale if s.startswith("run results")] if fails or errored: statuses["OPS-02"] = "FAIL" reasons["OPS-02"] = "failing on critical models: %s" % "; ".join(fails + errored) elif warns or run_stale: statuses["OPS-02"] = "WARN" reasons["OPS-02"] = "; ".join( (["%d test warning(s) on critical models (%s); confirm an owner and date for each" % (len(warns), "; ".join(warns))] if warns else []) + (["results are stale; compare with the job schedule"] if run_stale else [])) elif not test_ids: statuses["OPS-02"], reasons["OPS-02"] = "NOT CHECKED", "no tests on critical models to read; score OPS-01" elif unseen: statuses["OPS-02"], reasons["OPS-02"] = "NOT CHECKED", unseen_note + "; read the step that ran the tests" else: statuses["OPS-02"], reasons["OPS-02"] = "PASS", "no failing or warning tests on critical models" if unseen and statuses["OPS-02"] in ("FAIL", "WARN"): reasons["OPS-02"] += "; " + unseen_note if not srcs: statuses["FRS-04"], reasons["FRS-04"] = "NOT CHECKED", "no critical sources found for %s" % selection_text else: bad = ["%s (%s)" % (s["name"], s["signal"]) for s in srcs if s["signal_status"] == "FAIL"] weak = ["%s (%s)" % (s["name"], s["signal"]) for s in srcs if s["signal_status"] == "WARN"] if bad: statuses["FRS-04"] = "FAIL" reasons["FRS-04"] = "; ".join(["metadata-based or missing freshness: " + "; ".join(bad)] + (["ingestion timestamp: " + "; ".join(weak)] if weak else [])) elif weak: statuses["FRS-04"], reasons["FRS-04"] = "WARN", "ingestion timestamp: " + "; ".join(weak) else: statuses["FRS-04"], reasons["FRS-04"] = "PASS", "every critical source reads a data timestamp" if missing_sources: note = "sources not in the manifest: %s" % ", ".join(missing_sources) if statuses["FRS-04"] == "PASS": statuses["FRS-04"], reasons["FRS-04"] = "NOT CHECKED", note + "; the sources found pass" elif statuses["FRS-04"] == "NOT CHECKED": reasons["FRS-04"] = note else: reasons["FRS-04"] += "; also NOT CHECKED, " + note return OrderedDict([ ("manifest", OrderedDict([("dbt_version", meta.get("dbt_version")), ("schema", meta.get("dbt_schema_version")), ("generated_at", meta.get("generated_at")), ("adapter", adapter)])), ("run_results_generated_at", (run_meta or {}).get("generated_at")), ("freshness_generated_at", (fresh_meta or {}).get("generated_at")), ("critical", selection_text), ("tag", tag), ("names_given", len(names)), ("missing_models", missing_models), ("missing_sources", missing_sources), ("models", models), ("sources", srcs), ("stale_evidence", stale), ("suggested", statuses), ("reasons", reasons)]) def print_report(args, r): m = r["manifest"] schema = (m["schema"] or "").rsplit("/", 1)[-1].replace(".json", "") out = ["dbt artifact check: %s (dbt %s, manifest %s, generated %s, adapter %s)" % ( args.manifest, m["dbt_version"], schema or "unknown", m["generated_at"], m["adapter"] or "unknown")] if args.run_results: out.append("Run results: %s, generated %s." % (args.run_results, r["run_results_generated_at"])) if args.sources: out.append("Freshness results: %s, generated %s." % (args.sources, r["freshness_generated_at"])) lineage = len([s for s in r["sources"] if s["selected_by"] == "depends_on"]) out.append("Critical nodes (%s): %d models, %d sources (%d of them found through depends_on)." % ( r["critical"], len(r["models"]), len(r["sources"]), lineage)) if r["names_given"]: missing = (["models " + ", ".join(r["missing_models"])] if r["missing_models"] else []) + \ (["sources " + ", ".join(r["missing_sources"])] if r["missing_sources"] else []) out.append("Not in the manifest: %s." % ("; ".join(missing) if missing else "none")) for model in r["models"]: out += ["", "Model %s" % model["name"]] out.append(" Tests: %s" % (", ".join(model["tests"]) or "none")) if model["has_grain_test"]: out.append(" Grain: %s. Key not_null: %s." % ( ", ".join(model["grain_columns"]), "present" if not model["missing_not_null_on_grain"] else "missing on " + ", ".join(model["missing_not_null_on_grain"]))) else: out.append(" Grain: no unique or unique_combination_of_columns test.") absent = [x for x, flag in (("accepted_values", model["has_accepted_values"]), ("relationships", model["has_relationships"])) if not flag] if absent: out.append(" Not found: %s (needed where the model maps categories or joins on a key)" % ", ".join(absent)) for x in model["failing_tests"]: out.append(" Failing: %s" % x) for x in model["warning_tests"]: out.append(" Warning: %s" % x) if model["model_status"] and model["model_status"] != "success": out.append(" Model status in run results: %s" % model["model_status"]) for x in model["notes"]: out.append(" Note: %s" % x) for s in r["sources"]: extra = ["read by " + ", ".join(s["read_by"])] if s["read_by"] else [] if r["tag"] and s["selected_by"] == "depends_on": extra.append("not tagged %s" % r["tag"]) out += ["", "Source %s (identifier %s%s)" % (s["name"], s["identifier"], ("; " + "; ".join(extra)) if extra else "")] out.append(" Freshness: warn after %s, error after %s; %s. Signal: %s" % ( s["warn_after"] or "not set", s["error_after"] or "not set", s["signal"], s["signal_status"])) if s["result"]: out.append(" Latest result: %s" % s["result"]) if r["stale_evidence"]: out += ["", "Stale evidence:"] + ["- " + x for x in r["stale_evidence"]] out += ["", "Suggested statuses"] for cid, st in r["suggested"].items(): out.append("- %s %s: %s" % (cid, st, r["reasons"][cid])) errors = [s["name"] for s in r["sources"] if s["result_status"] in ("error", "runtime error")] if errors: out.append("- Freshness results in error on %s: score the lag under FRS-02." % ", ".join(errors)) elif r["sources"] and all(s["result_status"] == "pass" for s in r["sources"]): out.append("- All %d freshness results pass; a pass on a metadata-based check proves little (FRS-04)." % len(r["sources"])) out += ["These come from the default thresholds. Record the final status in the findings register.", "Nothing was changed."] print("\n".join(out)) def main(argv=None): p = argparse.ArgumentParser(description="Check dbt tests, results and freshness on critical nodes.") p.add_argument("manifest") p.add_argument("--run-results") p.add_argument("--sources", help="sources.json or freshness.json") p.add_argument("--config", help="audit config JSON; its outputs[].models are the critical models") p.add_argument("--critical", help="tag:NAME or comma-separated names (default tag:critical " "when --config is not given)") p.add_argument("--adapter", choices=["bigquery", "snowflake"]) p.add_argument("--as-of", help="YYYY-MM-DD used to judge artifact age (default today)") p.add_argument("--json", action="store_true") args = p.parse_args(argv) try: if args.as_of: date.fromisoformat(args.as_of) result = analyse(args) except (InputError, ValueError) as exc: print("error: %s" % exc, file=sys.stderr) return 2 if args.json: print(json.dumps(result, indent=2)) else: print_report(args, result) return 1 if "FAIL" in result["suggested"].values() else 0 if __name__ == "__main__": sys.exit(main())