Skip to content
StagbyteStagbyte
Back to projects
experimental

Data Reliability & AI QA Suite

A Python-based data-quality and observability framework for validating schemas, business rules, warehouse integrity, anomalies, drift, and lineage.

Repository

9

Validation stages

Declarative

Contract checks

Prometheus

Metrics export

About

The Data Reliability Suite automates data validation across ingestion, transformation, and warehouse layers. It uses declarative YAML contracts to enforce schemas, business rules, and referential integrity, and generates Prometheus-compatible metrics for observability.

Problem

Data pipelines often lack automated quality checks. Schema changes, data drift, and business rule violations go undetected until they impact downstream consumers. This framework provides a structured approach to data validation with alerting and lineage tracking.

Architecture

The framework processes data through sequential validation stages. Each stage is independent and can be enabled or disabled via configuration. Failed records are quarantined with metadata for investigation. Metrics are exposed via a Prometheus endpoint for integration with monitoring systems.

Pipeline

DATA_RELIABILITY_SUITE_PIPELINE

01
Ingestion
02
Volume Check
03
Schema Validation
04
Rule Validation
05
Quarantine
06
Transform
07
Warehouse
08
Integrity
09
Metrics

Workflow

01Raw Data → Ingestion layer
02Volume Anomaly Detection (z-score)
03Schema Validation against data contracts
04Data Contract Validation (types, formats, nulls, uniqueness)
05Business-Rule Validation (cross-field, referential integrity)
06Invalid-Record Quarantine
07Transformation
08PostgreSQL Warehouse Load
09Warehouse Integrity Checks
10Lineage and Metrics Export

Code

YAMLData contract YAML
# contracts/customers.yaml
name: customer_contract
version: 1.0
columns:
  - name: customer_id
    type: uuid
    nullable: false
    unique: true
  - name: email
    type: string
    format: email
    nullable: false
  - name: age
    type: integer
    nullable: true
    rules:
      - min: 18
      - max: 120
  - name: signup_date
    type: datetime
    nullable: false
    rules:
      - not_future: true
PythonValidation pipeline execution
from dataqa.contracts import load_contract
from dataqa.validators import (
    SchemaValidator,
    BusinessRuleValidator,
    IntegrityValidator,
)

contract = load_contract("contracts/customers.yaml")
df = pd.read_parquet("data/raw/customers.parquet")

schema_validator = SchemaValidator(contract)
schema_errors = schema_validator.validate(df)

rule_validator = BusinessRuleValidator(contract)
rule_errors = rule_validator.validate(df)

integrity = IntegrityValidator(db_connection)
referential_errors = integrity.check_referential(
    table="customers", foreign_keys=["region_id"]
)

all_errors = schema_errors + rule_errors + referential_errors
quarantine = InvalidRecordQuarantine()
quarantine.store(df.iloc[all_errors.index], errors=all_errors)

Quick start

Terminal
git clone https://github.com/ErvinAB/dataQA.git
cd dataQA && python -m venv venv && source venv/bin/activate
pip install -r requirements.txt
cp .env.example .env and configure database connection
python run_pipeline.py --source "data/sample.parquet"
docker compose up to start Prometheus metrics endpoint

Current functionality

  • Declarative YAML data contracts
  • Schema enforcement with type validation
  • Regex and format validation
  • Cross-field business rules
  • Referential integrity checks
  • Invalid-row quarantine with metadata
  • Z-score volume anomaly detection
  • Statistical drift detection
  • Prometheus-compatible metrics
  • Data lineage tracking
  • Docker containerisation
  • GitHub Actions CI/CD

Limitations

  • Designed for PostgreSQL warehouse targets
  • Anomaly detection uses basic statistical methods
  • Lineage is file-based, not integrated with data catalog tools

Planned improvements

  • Support additional warehouse targets
  • Add ML-based anomaly detection
  • Integrate with data catalog tools
  • Add dashboard visualisation
  • Expand contract library

Technology

PythonpytestPandasPydanticPostgreSQLSQLAlchemyDockerGitHub ActionsPrometheus

Status

experimental

This project is an active experimental framework. It is not production-ready and should be evaluated for your specific use case.

Interested in this project?

Stagbyte builds practical automation systems. If this project aligns with a problem you are solving, reach out to discuss how it can be adapted or extended.

Start the conversation