DQVF � Data Quality Validation Framework

A Python-based data quality validation framework for database migrations. DQVF orchestrates a multi-stage, agent-driven validation pipeline over a source and target database pair, uses LLM reasoning for borderline failure assessment and root-cause analysis, and surfaces results through a CLI, a REST API, and an embedded web dashboard.

Overview

DQVF validates a database migration by running six specialised validation agents across one or more tables. Each agent performs a focused category of checks � structural, columnar, pattern-based, row-level reconciliation, semantic, and freshness. The agents are executed in an ordered, staged pipeline managed by LangGraph. After all stages complete, a root-cause analysis node uses an LLM to synthesise findings and produce a go/no-go recommendation. Results are written as JSON reports and optionally as PDF and Slack notifications.

The framework supports two execution modes:

Cutover mode � strict thresholds for a final one-shot migration. Any HIGH-severity failure blocks go-live.
Live-sync mode � widened thresholds for migrations running alongside an active source. Row-count drift and hash divergence are reported as warnings rather than failures.

Architecture

CLI / Web UI
     |
     v
LangGraph Validation Graph
     |
     +-- initialize_node        (load config, parse user instruction, resolve intent)
     |
     +-- stage_1                (table_profiling)            [sequential, blocking]
     |      |
     |      +-- assess_stage_1  (LLM or rule-based gate)
     |              |
     |              +-- proceed/warn  --> stage_2
     |              +-- blocked       --> human_review
     |
     +-- stage_2                (column_profiling)           [sequential, blocking]
     |      |
     |      +-- assess_stage_2  (LLM or rule-based gate)
     |              |
     |              +-- proceed/warn  --> stage_3
     |              +-- blocked       --> human_review
     |
     +-- stage_3                (pattern_analysis, reconciliation)  [parallel]
     |
     +-- stage_4                (semantic_validation, freshness)    [parallel]
     |
     +-- root_cause_node        (LLM synthesis: actions, SQL, risk, go-live rec.)
     |
     +-- report_node            (write JSON, PDF, send Slack)
     |
     v
    END

The graph topology is generated dynamically at startup by graph_builder.py from AGENT_REGISTRY. Adding a new agent requires only one new file and one line in manifest.yaml � no changes to the graph or node files.

Human review is modelled as an interrupt. When an assess node decides a stage is blocked, the graph pauses and waits for an external --resume call carrying an approved or rejected decision. The decision is recorded in state with the approver's identity for audit compliance.

Validation Agents

All agents extend BaseAgent and are registered at startup via agents/registry.py using the manifest allow-list.

Stage 1 � Table Profiling (blocking)

Module: dqvf/agents/table_profiling.py

Checks structural integrity at the table level before any column or row-level validation proceeds.

Row count comparison between source and target within the configured tolerance.
Schema presence: verifies all expected tables exist in the target.
Primary key existence and uniqueness.
Null-rate comparison per column against the source baseline.
Volume baseline check.

A HIGH failure here blocks all downstream stages until a human reviewer approves continuation.

Stage 2 � Column Profiling (blocking)

Module: dqvf/agents/column_profiling.py

Validates column-level fidelity after structural checks pass.

Column presence: detects missing or unexpected columns, accounting for declared column_map renames and column_transformations.
Data-type compatibility between source and target.
Null-rate delta per column against configured null_rate_delta_pct.
Cardinality change detection within cardinality_change_pct.
Distribution drift using a statistical p-value test (distribution_drift_pvalue).

A HIGH failure here also triggers the human-review interrupt.

Stage 3a � Pattern Analysis (parallel)

Module: dqvf/agents/pattern_analysis.py

Runs in parallel with reconciliation. Detects format and pattern anomalies.

Regex pattern consistency for string columns (email, phone, identifiers).
Fuzzy duplicate detection using RapidFuzz with a configurable similarity threshold.
Levenshtein distance checks for near-identical values that may represent encoding errors.

Stage 3b � Reconciliation (parallel)

Module: dqvf/agents/reconciliation.py

Performs row-level data fidelity checks in parallel with pattern analysis.

Chunk-based MD5/SHA hash comparison between source and target rows, bounded by chunk_size.
Primary-key join to detect orphaned rows (rows present in source but absent in target).
Column-level value comparison for sampled rows.
CDC lag check: warns if target data is older than cdc_lag_max_minutes.
Table-merge transformation validation: verifies denormalised tables preserve the expected combined row counts.

Stage 4a � Semantic Validation (parallel)

Module: dqvf/agents/semantic_validation.py

LLM-assisted semantic checks that complement structural rules.

Business-rule evaluation: runs table-level SQL expressions declared in the config (e.g., created_at <= NOW()).
Transformation-aware validation: interprets declared column_transformations to validate computed, merged, split, renamed, and unit-converted columns.
Transformation auto-detection: when enabled, samples rows and prompts the LLM to infer undeclared transformations.

Stage 4b � Freshness (parallel)

Module: dqvf/agents/freshness.py

Validates data currency.

Compares the maximum timestamp column value in source and target.
Raises HIGH if the target is stale beyond the table-level sla_freshness_hours.
Raises MEDIUM if freshness data is absent or the timestamp column cannot be inferred.

LangGraph Orchestration

Graph State

The shared state type is ValidationGraphState (a TypedDict defined in dqvf/langgraph/state.py). It carries:

Input: run_id, config_path, parsed config object.
Agent results: one key per agent (table_profiling, column_profiling, pattern_analysis, reconciliation, semantic_validation, freshness).
Routing decisions: stage_N_decision values written by assess nodes.
Human review audit: human_review_decision with approver identity and reason.
Root-cause fields: root_cause_summary, recommended_actions, schema_remediation_queries, business_impact, risk_assessment, llm_go_live_recommendation.
Final output: report (SupervisorReport), go_live_blocked.

The state is fully serialisable. Connectors (database, LLM) are instantiated per-node and never stored in state, which is required for checkpoint persistence.

Checkpointing

The graph uses AsyncSqliteSaver backed by dqvf.db for development. Every node completion is checkpointed. If a run is interrupted (human review), the state is saved and can be resumed by run ID without re-executing completed stages.

For production, replace the saver with AsyncPostgresSaver.

Intent Parser

dqvf/langgraph/intent.py translates a free-text --instruction string into a structured Intent object that specifies which agents to run, urgency level, and which tables to focus on. It has two modes:

LLM parsing � if an LLM is configured and available, the instruction is sent to the model with the list of available agents and tables. The response is parsed as structured JSON.
Heuristic parsing � falls back to keyword matching against each agent's declared agent_keywords list and urgency keyword lists (critical, audit, etc.).

Assess Nodes

Each blocking stage has an assess node generated by make_assess_node(). The decision logic is:

Condition	Decision
No HIGH or MEDIUM failures	proceed
MEDIUM failures only	warn (continue)
More than 5 HIGH failures	blocked (fast path, no LLM call)
1-5 HIGH failures, LLM available	LLM reasoning (proceed / blocked / warn)
1-5 HIGH failures, no LLM	blocked (fallback)

The --force flag bypasses all assess nodes and proceeds regardless of failure count.

Root Cause Node

After all stages complete, root_cause_node collects every failure across all agents and invokes the LLM to produce:

A narrative root-cause analysis.
An ordered list of recommended corrective actions.
Ready-to-run SQL remediation queries.
A business impact narrative.
A risk assessment summary.
A go-live recommendation: PROCEED, PROCEED_WITH_CAUTION, or BLOCK.

If no LLM is configured, the node generates deterministic text summaries from the raw failure counts.

Human Review Node

When an assess node decides blocked, execution halts at human_review_node. The node logs the interrupt and exits. The caller receives None from run_validation_graph() and the CLI prints a resume command.

To resume:

python -m dqvf --resume --run-id <run-id> --approve approved --approved-by "your-name" --approval-reason "reason"

The resume injects the decision into the checkpointed state and re-invokes the graph from the interrupted node.

Connectors

All connectors extend BaseConnector (dqvf/connectors/base_connector.py) and are constructed by connector_factory.py from the source and target config blocks.

Connector	Config type	Notes
`PostgresConnector`	`postgres`, `neon`	asyncpg-based, full async, SSL, chunked streaming
`SqlServerConnector`	`sqlserver`, `mssql`	pyodbc-based with asyncio bridge, Windows Auth support
`MongoDBConnector`	`mongodb`	motor-based, ObjectId-aware, aggregation pipelines

All connectors implement a common interface: execute_query, fetch_many, get_table_info, stream_chunks, and close.

Configuration Reference

Runs are configured via YAML files. Example configs are provided in configs/.

migration_run_id: "run-2026-06-23"

source:
  type: mongodb
  connection_string: "${SOURCE_DB_URI}"
  database: "${SOURCE_DB_NAME}"

target:
  type: neon              # or postgres, sqlserver
  host: "${TARGET_DB_HOST}"
  port: 5432
  user: "${TARGET_DB_USER}"
  password: "${TARGET_DB_PASSWORD}"
  database: "${TARGET_DB_NAME}"
  ssl: true

tables:
  - name: users
    pk: id                        # target primary key column
    source_pk: _id                # source primary key column (e.g., MongoDB ObjectId field)
    schema: "${TARGET_DB_SCHEMA:-target}"
    sla_freshness_hours: 24       # freshness SLA for this table
    column_map:
      _id: id                     # declare known renames so column checks do not false-positive
    skip_columns: []
    business_rules:
      - name: no_future_created_at
        expression: "created_at <= NOW()"
        description: "created_at should never be in the future"

thresholds:
  row_count_tolerance_pct: 0.0       # 0 = exact match required
  null_rate_delta_pct: 1.0           # max allowed null-rate increase per column
  distribution_drift_pvalue: 0.05    # p-value below which a distribution shift is flagged
  cardinality_change_pct: 10.0       # max allowed cardinality change per column
  chunk_size: 100000                 # rows per chunk for hash reconciliation
  cdc_lag_max_minutes: 15.0          # max CDC lag before a WARN is raised
  hash_algorithm: md5
  volume_baseline_pct: 10.0
  fuzzy_duplicate_threshold: 0.85    # RapidFuzz similarity score for duplicate detection
  levenshtein_max_distance: 3
  parallel_chunk_workers: 4          # concurrent table and chunk operations

agents:
  table_profiling: true
  column_profiling: true
  pattern_analysis: true
  reconciliation: true
  semantic: true
  freshness: true

llm:
  provider: openrouter
  model: "${LLM_MODEL:-google/gemini-2.5-flash-lite}"
  base_url: "${LLM_BASE_URL:-https://openrouter.ai/api/v1}"
  max_tokens: 4096
  temperature: 0.0
  timeout_seconds: 30               # fallback to rule-based decisions after this many seconds

notifications:
  slack_webhook: null               # set to a Slack webhook URL to enable
  alert_on:
    - HIGH
    - MEDIUM

validation_mode: cutover            # or live_sync

# Declare intentional column transformations to suppress false positives
column_transformations:
  - id: full_name_merge
    type: merge
    table: customers
    source_columns: [first_name, middle_name, last_name]
    target_columns: [full_name]
    sql_expression: "TRIM(CONCAT_WS(' ', first_name, NULLIF(middle_name,''), last_name))"
    tolerance_pct: 1.0

# Declare table-merge (denormalisation) transformations
table_transformations:
  - id: order_denormalization
    type: table_merge
    source_tables: [orders, order_lines]
    target_table: order_details
    join_key: order_id
    tolerance_pct: 1.0

transformation_auto_detection:
  enabled: false
  confidence_threshold: 0.75
  sample_size: 100

CLI Reference

python -m dqvf [options]

Subcommands

Command	Purpose
`dqvf serve`	Launch the embedded Flask web dashboard

Common options

Flag	Description
`-c, --config FILE`	Path to a YAML config file (required unless `--auto`)
`-i, --instruction TEXT`	Natural-language instruction to scope the run
`-o, --report-dir DIR`	Directory for output reports (default: `./reports`)
`--auto`	Auto-discover source schema, generate config, then run validation
`--discover`	Auto-discover schema and print the generated YAML; do not validate
`--auto-config-out FILE`	Save the auto-generated YAML to this file
`--env NAME`	Load `.env.<name>` in addition to `.env`
`--run-id ID`	Override the auto-generated run identifier
`--ci`	CI mode: write a compact JSON summary to stdout; exit 2 on go-live block
`--fail-on SEV`	Minimum severity that sets a non-zero exit code (`HIGH`, `MEDIUM`, `LOW`)
`--force`	Run all agents, ignoring HIGH failures in assess nodes
`--schema SCHEMA`	Override the target schema for all tables
`--exclude TABLES`	Comma-separated table names to skip
`--include TABLES`	Comma-separated table names to process (all others skipped)
`-a, --agent AGENT`	Debug mode: run only a single named agent
`--agents AGENTS`	Comma-separated list of agents to run
`-t, --table TABLE`	Restrict validation to a single table
`--log-level LEVEL`	Log verbosity: `DEBUG`, `INFO`, `WARN`, `ERROR`
`--no-checkpoint`	Disable SQLite checkpointing for this run
`--skip-reports`	Do not write report files

Resume flags

Flag	Description
`--resume`	Resume a paused (human-review interrupted) run; requires `--run-id`
`--approve DECISION`	`approved` or `rejected`
`--approved-by IDENTITY`	Approver identity recorded in the audit trail
`--approval-reason TEXT`	Justification for the decision

Exit codes

Code	Meaning
0	Validation passed, go-live clear
1	Validation failed (at least one agent FAIL)
2	Go-live blocked (HIGH severity failures remain unresolved)
3	Run paused, awaiting human review

Usage examples

# Full validation with an LLM instruction
python -m dqvf --config configs/example-run.yml --instruction "critical full validation before cutover"

# Auto-discover the source schema and run validation
python -m dqvf --auto --report-dir ./reports

# Discover only � print the generated YAML and exit
python -m dqvf --discover --auto-config-out configs/generated.yml

# CI mode � compact JSON summary, exit 2 on go-live block
python -m dqvf --config configs/example-run.yml --ci

# Run only two specific agents
python -m dqvf --config configs/example-run.yml --agents reconciliation,freshness

# Resume a paused human-review run
python -m dqvf --resume --run-id run-2026-06-23T10-00-00 \
  --approve approved \
  --approved-by "eng-lead" \
  --approval-reason "Known schema renames, covered by column_map"

# Launch the web dashboard
python -m dqvf serve --port 4173

Web UI and REST API

The embedded Flask application is started with dqvf serve. It provides a browser dashboard for triggering runs, viewing reports, and submitting human-review decisions, as well as a REST API for programmatic access.

Default address: http://127.0.0.1:4173

Serve options

Flag	Default	Description
`--host HOST`	`127.0.0.1`	Bind address
`--port PORT`	`4173`	Port
`--report-dir DIR`	`./reports`	Directory scanned for report JSON files
`--config-dir DIR`	`./configs`	Directory scanned for YAML config files
`--debug`	off	Enable Flask debug/reloader mode

Read-only API endpoints (no authentication required)

Method	Path	Description
GET	`/`	Dashboard index � lists all runs and available configs
GET	`/report/<run_id>`	Rendered HTML report for a single run
GET	`/api/runs`	JSON array of all run summaries
GET	`/api/run/<run_id>`	Full JSON report for a specific run
GET	`/api/status`	Server health, engine name, latest run status
GET	`/download/<run_id>.pdf`	Download the PDF report for a run

Mutating API endpoints (require X-DQVF-API-Key header)

Method	Path	Body	Description
POST	`/api/run`	`{"config": "example-run.yml", "instruction": "...", "force": false}`	Trigger a new validation run asynchronously; returns `runId`
POST	`/api/run-report`	Same as above	Trigger a run and block until the PDF is generated; returns download URL
POST	`/api/approve/<run_id>`	`{"decision": "approved", "reason": "..."}`	Resume a paused human-review run; requires approver role

Authentication and role checking are implemented in dqvf/ui/auth.py. The API key and approver role configuration is read from environment variables.

Reporting

Each completed run produces up to three output artefacts in the report directory.

JSON report (`<run_id>.json`)

The canonical output. Contains:

run_id, timestamp, overall_status (PASS, FAIL, WARN), go_live_blocked.
agents: per-agent status, rules checked, rules failed, and all failure details.
root_cause_analysis, recommended_actions, schema_remediation_queries.
business_impact, risk_assessment, go_live_recommendation.
human_review_decision (if a human review occurred).
execution_plan, duration_ms, schema_version.

PDF report (`<run_id>.pdf`)

A formatted PDF generated by ReportLab. Includes all sections from the JSON report with a summary table, per-agent breakdown, failure detail tables, and the LLM-generated root-cause narrative.

Slack notification

If notifications.slack_webhook is set, a notification is sent to the webhook after the run completes. Alert severity filtering is controlled by notifications.alert_on.

Extending the Framework

Adding a new validation agent

Create dqvf/agents/<your_agent>.py with a class that:
- Extends BaseAgent.
- Declares agent_name, agent_stage (1-4), agent_keywords, and _COMPUTATION_LOCATION ("db" or "python_bounded").
- Implements async _validate_table(self, ctx, table) -> list[RuleCheckResult].
Add the module name (without .py) to dqvf/agents/manifest.yaml.

That is the complete change required. The registry, graph builder, intent parser, root-cause node, and report node all discover and integrate the new agent automatically.

Adding a new validation rule

Rules are organised under dqvf/rules/<agent_name>/. Each rule file defines a function or class invoked by the corresponding agent. The rules/registry.py loads rules by convention. Adding a rule file makes it available to the agent without changing any other file.

Adding a new database connector

Create dqvf/connectors/<name>_connector.py with a class extending BaseConnector.
Register the new type string in connector_factory.py.

Memory safety contract

All agent implementations must satisfy one of two computation locations:

"db" � all heavy computation is pushed to the database as aggregate queries. Python only processes counts and summaries.
"python_bounded" � Python-side computation is limited to at most thresholds.chunk_size rows per operation.

The BaseAgent.run() method enforces this at startup by inspecting _COMPUTATION_LOCATION and raising an AssertionError if the contract is missing.

Installation

Python 3.11 or later is required.

# Clone and enter the repository
cd dqvf_python

# Create and activate a virtual environment
python -m venv .venv
.venv\Scripts\activate          # Windows
source .venv/bin/activate       # Linux / macOS

# Install dependencies
pip install -r requirements.txt

# Install the package in editable mode
pip install -e .

Verify the installation:

python -m dqvf --help

Environment Variables

Copy .env and fill in the values for your environment. All variables support ${VAR:-default} expansion inside YAML configs.

Variable	Description
`SOURCE_DB_URI`	MongoDB connection string for the source database
`SOURCE_DB_NAME`	Source MongoDB database name
`TARGET_DB_HOST`	Target PostgreSQL / Neon host
`TARGET_DB_USER`	Target database username
`TARGET_DB_PASSWORD`	Target database password
`TARGET_DB_NAME`	Target database name
`TARGET_DB_SCHEMA`	Target schema (default: `target`)
`LLM_MODEL`	LLM model identifier (default: `google/gemini-2.5-flash-lite`)
`LLM_BASE_URL`	LLM API base URL (default: OpenRouter)
`OPENAI_API_KEY`	API key when using OpenAI directly
`ANTHROPIC_API_KEY`	API key when using Anthropic directly
`DQVF_API_KEY`	Secret key for the web UI mutating API endpoints

Multiple environments are supported via named env files. --env staging loads .env.staging on top of .env.

Project Layout

dqvf_python/
+-- dqvf/
�   +-- main.py                    # CLI entry point
�   +-- __main__.py                # python -m dqvf shim
�   +-- agents/
�   �   +-- manifest.yaml          # agent allow-list for auto-import
�   �   +-- registry.py            # agent registry builder
�   �   +-- base_agent.py          # abstract base class for all agents
�   �   +-- table_profiling.py
�   �   +-- column_profiling.py
�   �   +-- pattern_analysis.py
�   �   +-- reconciliation.py
�   �   +-- semantic_validation.py
�   �   +-- freshness.py
�   +-- langgraph/
�   �   +-- graph.py               # public API shim
�   �   +-- graph_builder.py       # dynamic graph topology builder
�   �   +-- runner.py              # run / resume entry point
�   �   +-- state.py               # ValidationGraphState TypedDict
�   �   +-- intent.py              # free-text instruction parser
�   �   +-- checkpointer.py        # checkpoint helpers
�   �   +-- connectors.py          # shared connector factory for nodes
�   �   +-- nodes/
�   �   �   +-- initialize_node.py
�   �   �   +-- stage_runner.py
�   �   �   +-- assess_node.py
�   �   �   +-- human_review_node.py
�   �   �   +-- root_cause_node.py
�   �   �   +-- report_node.py
�   �   +-- edges/
�   �   �   +-- routing.py
�   �   +-- tools/
�   +-- connectors/
�   �   +-- base_connector.py
�   �   +-- connector_factory.py
�   �   +-- postgres_connector.py
�   �   +-- sqlserver_connector.py
�   �   +-- mongodb_connector.py
�   +-- rules/
�   �   +-- registry.py
�   �   +-- base_rule.py
�   �   +-- table_profiling/
�   �   +-- column_profiling/
�   �   +-- pattern_analysis/
�   �   +-- reconciliation/
�   �   +-- freshness/
�   +-- reporters/
�   �   +-- json_reporter.py
�   �   +-- pdf_reporter.py
�   �   +-- slack_notifier.py
�   +-- config/
�   �   +-- schema.py              # Pydantic config models
�   �   +-- loader.py
�   �   +-- discovery.py
�   �   +-- env_loader.py
�   +-- llm/                       # LLM client wrappers
�   +-- transformations/           # column and table transformation validators
�   +-- types/                     # shared Pydantic types (AgentContext, report types)
�   +-- utils/
�   �   +-- logger.py
�   +-- supervisor/                # legacy supervisor package (deprecated)
�   +-- ui/
�       +-- server.py              # Flask application
�       +-- auth.py                # API key and role checking
�       +-- templates/             # Jinja2 HTML templates
�       +-- static/                # CSS, JS, assets
+-- configs/
�   +-- example-run.yml
�   +-- example-with-rules.yml
�   +-- excel-seed-run.yml
�   +-- full-run.yml
+-- reports/                       # generated JSON and PDF reports (git-ignored)
+-- tests/
+-- scripts/
+-- pyproject.toml
+-- requirements.txt
+-- .env                           # local environment variables (git-ignored)

Dependencies

Package	Purpose
`pydantic >= 2.7`	Config schema validation and structured LLM output parsing
`pyyaml >= 6.0`	YAML config parsing
`python-dotenv >= 1.0`	`.env` file loading
`asyncpg >= 0.29`	Async PostgreSQL driver
`pyodbc >= 5.1`	SQL Server / ODBC driver
`motor >= 3.4`	Async MongoDB driver
`pymongo >= 4.7`	MongoDB sync utilities and ObjectId handling
`anthropic >= 0.30`	Anthropic Claude LLM client
`openai >= 1.35`	OpenAI-compatible LLM client (also used for OpenRouter)
`scipy >= 1.13`	Statistical distribution drift tests
`numpy >= 1.26`	Numerical array operations
`rapidfuzz >= 3.9`	Fuzzy string matching for duplicate detection
`reportlab >= 4.2`	PDF report generation
`aiohttp >= 3.9`	Async HTTP for Slack webhook notifications
`flask >= 3.0`	Embedded web dashboard
`langgraph >= 0.2`	Stateful multi-agent graph orchestration
`pytest >= 8.2`	Test runner
`pytest-asyncio >= 0.23`	Async test support

Name		Name	Last commit message	Last commit date
Latest commit History 12 Commits
configs		configs
dqvf.egg-info		dqvf.egg-info
dqvf		dqvf
reports		reports
scripts		scripts
tests		tests
.gitignore		.gitignore
README.md		README.md
dqvf.db		dqvf.db
dqvf.db-shm		dqvf.db-shm
dqvf.db-wal		dqvf.db-wal
extract.py		extract.py
pyproject.toml		pyproject.toml
read_fails.py		read_fails.py
requirements.txt		requirements.txt
test_llm.py		test_llm.py

Folders and files

Latest commit

History

Repository files navigation