vnpt_data_governance

DBGraph (VNPT Data Governance SDK) helps you explore and find relevant data assets in a large, complex relational database. It introspects a schema, builds a navigable graph of tables/columns, classifies PII, flags quality issues, infers relationships, optionally enriches it with LLM-generated descriptions, and lets you search/render/traverse that graph.

This package is a standalone SDK extracted from the internal dbgraph project, packaged for reuse across teams/services (originally built by minhdenthedev, packaged as an SDK by hainamnguyen192).

Install

Core install (graph building/traversal/PNG rendering; no network DB drivers or LLM client):

pip install vnpt_data_governance

Pick the extras you actually need — each one only pulls in the dependency for that piece:

pip install "vnpt_data_governance[postgres]"   # PostgresDataGateway
pip install "vnpt_data_governance[mysql]"      # MySQLDataGateway
pip install "vnpt_data_governance[trino]"      # TrinoDataGateway
pip install "vnpt_data_governance[openai]"     # OAICompatibleLLM
pip install "vnpt_data_governance[search]"     # BM25SearchEngine
pip install "vnpt_data_governance[all]"        # everything above

SQLite is supported out of the box (Python's built-in sqlite3), no extra needed.

Quick start

The one-call entry point — introspects the database and returns a fully enriched graph (schema, statistics, PII classification, quality issues, normalization hints, inferred soft-FK/lineage relationships). Point it at a sqlite file (or a postgres:///mysql:///trino:// DSN) and it does the rest, including writing a JSON dump and a PNG diagram next to the source:

import vnpt_data_governance

graph = vnpt_data_governance.analyze("data/northwind.db")
# -> data/northwind.json, data/northwind.png written automatically

For a network database, pass its DSN instead of a file path — same one-liner:

graph = vnpt_data_governance.analyze("postgres://user:pass@host:5432/mydb")
graph = vnpt_data_governance.analyze("trino://user:pass@host:443/catalog/schema")

Pass an explicit save_json=/render_png= path to control where those land (or False to skip one), and db_schema= when a trino:///postgres:// DSN doesn't carry a schema segment:

graph = vnpt_data_governance.analyze(
    "trino://user:pass@host:443/catalog",  # no /schema in the DSN
    db_schema="app",
    save_json="out/graph.json",
    render_png=False,
)

Full control: pass a gateway instead of a path

For anything the DSN auto-detection can't cover (a nonstandard sqlite db_schema, connection pooling, non-default ports baked into a gateway object, ...), pass a ready-made gateway instead of a string — analyze() then behaves exactly like before: no JSON/PNG unless you ask for them:

from pathlib import Path
from vnpt_data_governance import OAICompatibleLLM

graph = vnpt_data_governance.analyze(
    vnpt_data_governance.SqliteDataGateway(Path("data/northwind.db")),
    llm=OAICompatibleLLM(model=..., base_url=..., api_key=...),  # requires the `openai` extra
    save_json=Path("data/northwind-graph.json"),
    search_index_dir=Path("data/northwind-index"),  # requires the `search` extra
)

# load a previously saved graph back later
graph = vnpt_data_governance.JSONGraphLoader(json_path=Path("data/northwind-graph.json")).load()

# ...and re-render its PNG straight from that JSON, without rebuilding the graph
vnpt_data_governance.GraphRenderer.render_json_file(
    "data/northwind-graph.json", "data/northwind-graph.png"
)

# search it (requires the `search` extra)
search_engine = vnpt_data_governance.BM25SearchEngine(Path("data/northwind-index"))
asset_ids = search_engine.search("restricted PII columns in the Sales domain")

Advanced: composing the pipeline yourself

analyze() is a thin wrapper around individually usable pieces — reach for these instead if you need custom prompts, want to skip/reorder steps, or want finer control than analyze()'s options give you:

from pathlib import Path

from vnpt_data_governance import (
    JSONGraphWriter,
    RGraphBuilder,
    SqliteDataGateway,
    SoftForeignKeyDetector,
    LineageDetector,
)

graph_builder = RGraphBuilder(SqliteDataGateway(Path("data/northwind.db")))
graph = graph_builder.build_graph()
graph = SoftForeignKeyDetector(SqliteDataGateway(Path("data/northwind.db"))).detect(graph)
graph = LineageDetector(SqliteDataGateway(Path("data/northwind.db"))).detect(graph)

# optional: generate semantic descriptions for assets via an LLM, with your
# own prompts
# from vnpt_data_governance import GraphDescriptorV1, OAICompatibleLLM
#
# graph_descriptor = GraphDescriptorV1(
#     llm=OAICompatibleLLM(model=..., base_url=..., api_key=...),
#     system_prompt=..., formating_prompt=..., target_prompt=...,
# )
# graph = graph_descriptor.rfill_semantic_aspects(graph)

JSONGraphWriter(json_path=Path("data/northwind-graph.json"), indent=2).write(graph)

For visualization purposes, here is a graph saved as JSON:

{
  "assets": [
    {
      "asset_id": "8ab5a624-0596-497e-a0ee-3996d95dbe63",
      "name": "Categories",
      "type": "table",
      "aspects": {
        "schema_properties": { "name": "Categories_table_schema", "pks": ["CategoryID"], "indices": {} },
        "statistical_properties": { "name": "Categories_table_stats", "num_columns": 4, "num_rows": 8 },
        "semantic_properties": {
          "name": "Categories_semantic",
          "description": "Stores product category definitions and metadata, serving as a lookup table for classifying products in the inventory system.",
          "keywords": ["categories", "product classification", "category definitions", "inventory groups", "product types"]
        }
      }
    }
  ],
  "links": [
    {
      "link_id": "db6bea93-a02c-4426-a2db-449e4a7bba8f",
      "name": "Categories_CategoryID",
      "type": "contain",
      "source_id": "8ab5a624-0596-497e-a0ee-3996d95dbe63",
      "destination_id": "04c20046-2808-4021-bbf1-99876e0eea6e",
      "aspects": {}
    }
  ]
}

Use cases

• Manipulating database schema — build the schema graph, store it, and use it to traverse the database, find join paths, get referenced tables, ...
• Profiling database — the Aspect concept represents different kinds of properties attached to a data asset (schema, statistics, semantics, ...).
• Render graph — output a schema graph as a PNG diagram (GraphRenderer) or Markdown/text to feed as LLM context.
• LLM assistance — use an LLM to generate data assets' descriptions/keywords, and as input for downstream SQL generation.
• Search for data assets — search assets via BM25 indexing/retrieval, over a document built from every enriched aspect (description/keywords, PII classification, quality issues, domain), not just the description.

Architecture

DBGraph is designed to be easy to extend:

1. Core classes (entities) hold the shared business logic of database graphs (traversal, neighborhoods, ...) and core operations (building graphs, profiling databases, ...). The prefix R... stands for "Relational" (the only paradigm currently supported); D..., V..., G... are reserved for Document/Vector/Graph paradigms.
2. Interfaces (extensions) mark the parts of the system meant to be pluggable:
   • RGraphBuilder works against any RDBMS via the RDataGateway abstraction — implementations ship for SQLite, PostgreSQL, MySQL and Trino.
   • LLM abstracts the model provider — OAICompatibleLLM is the bundled implementation ([openai] extra); bring your own by implementing LLM.generate/agenerate.
   • GraphWriter/GraphLoader abstract graph persistence — JSONGraphWriter/JSONGraphLoader are the bundled implementation.
   • GraphRenderer renders any DatabaseGraph (including one just loaded via JSONGraphLoader) as a PNG schema diagram; also runnable as a script: python -m vnpt_data_governance.io.graph_renderer graph.json graph.png.
   • SearchEngine abstracts indexing/retrieval — BM25SearchEngine is the bundled implementation ([search] extra).

Development

uv sync --group dev --all-extras
uv run pytest
uv run pylint vnpt_data_governance
uv run mypy vnpt_data_governance
uv run flake8 vnpt_data_governance

Tests that talk to Postgres/MySQL/Trino/OpenAI need real credentials (see tests/) and are skipped/fail without them; the SQLite, entity and loader/writer tests run standalone.