Stop Hardcoding Database Checks: Building a Metadata-Driven Data Quality Framework

import os
import json
import logging
from datetime import datetime
import pandas as pd
from sqlalchemy import create_engine
import great_expectations as gx
from great_expectations.core.batch import RuntimeBatchRequest

# Configure structured logging for production auditing
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class MetadataDataQualityEngine:
    def __init__(self, connection_string: str):
        self.engine = create_engine(connection_string)
        # Initialize Great Expectations ephemeral context for programmatic runtime control
        self.context = gx.get_context(context_root_dir=None)

    def fetch_active_metadata(self) -> pd.DataFrame:
        """Harvests active validation configurations from the centralized store."""
        query = """
            SELECT table_name, column_name, expectation_type, expectation_kwargs 
            FROM data_quality_rules 
            WHERE is_active = TRUE
        """
        try:
            df = pd.read_sql(query, self.engine)
            logger.info(f"Successfully harvested {len(df)} active validation rules.")
            return df
        except Exception as e:
            logger.error(f"Failed to query metadata repository: {str(e)}")
            raise

    def compile_expectation_suite(self, table_name: str, rules_df: pd.DataFrame):
        """Assembles validation rules into a Great Expectations suite on the fly."""
        suite_name = f"{table_name}_suite"
        suite = self.context.add_or_update_expectation_suite(expectation_suite_name=suite_name)
        
        # Filter metadata constraints for the specific target asset
        table_rules = rules_df[rules_df['table_name'] == table_name]
        
        for _, row in table_rules.iterrows():
            # Parse JSON kwargs configuration gracefully
            kwargs = row['expectation_kwargs']
            if isinstance(kwargs, str):
                kwargs = json.loads(kwargs)
                
            kwargs['column'] = row['column_name']
            
            # Programmatically map string values to structured GX expectation objects
            expectation_config = gx.core.ExpectationConfiguration(
                expectation_type=row['expectation_type'],
                kwargs=kwargs,
                meta={"notes": f"Automated constraint enforcement for column: {row['column_name']}"}
            )
            suite.add_expectation(expectation_config)
            
        self.context.add_or_update_expectation_suite(suite=suite)
        return suite

    def execute_quality_checkpoint(self, table_name: str, target_df: pd.DataFrame):
        """Builds a runtime batch request and evaluates data against the generated suite."""
        timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
        suite_name = f"{table_name}_suite"
        checkpoint_name = f"{table_name}_checkpoint"

        # Unique runtime composite signature prevents processing trace collisions
        batch_request = RuntimeBatchRequest(
            datasource_name="lakehouse_runtime_datasource",
            data_connector_name="runtime_data_connector",
            data_asset_name=f"{table_name}_{timestamp}",
            runtime_parameters={"batch_data": target_df},
            batch_identifiers={"table_name": table_name, "execution_timestamp": timestamp}
        )

        # Register and fire a dynamic checkpoint execution
        self.context.add_or_update_checkpoint(
            name=checkpoint_name,
            config_version=1,
            class_name="SimpleCheckpoint",
            validations=[{
                "batch_request": batch_request, 
                "expectation_suite_name": suite_name
            }]
        )

        logger.info(f"Launching data quality checkpoint for table: {table_name}")
        return self.context.run_checkpoint(checkpoint_name=checkpoint_name)

# Production Loop Execution Pattern
if __name__ == "__main__":
    SF_CONN = "snowflake://<user>:<pass>@<account>/<db>/<schema>?warehouse=COMPUTE_WH&role=SYSADMIN"
    
    dq_engine = MetadataDataQualityEngine(connection_string=SF_CONN)
    metadata_rules = dq_engine.fetch_active_metadata()
    distinct_target_tables = metadata_rules['table_name'].unique()

    for current_table in distinct_target_tables:
        try:
            # Stage current batch dataset from target engine
            raw_data_df = pd.read_sql(f"SELECT * FROM {current_table}", dq_engine.engine)
            
            # Step 1: Build suite dynamically from relational rules
            dq_engine.compile_expectation_suite(table_name=current_table, rules_df=metadata_rules)
            
            # Step 2: Validate batch data and extract metrics payload
            eval_result = dq_engine.execute_quality_checkpoint(table_name=current_table, target_df=raw_data_df)
            
            if not eval_result["success"]:
                logger.warning(f"Data Quality anomalies detected on asset: {current_table}")
            else:
                logger.info(f"Asset {current_table} successfully cleared all metadata expectations.")
                
        except Exception as err:
            # Fault isolation ensures an asset failure never crashes cascading pipeline steps
            logger.error(f"Processing loop interrupted on asset {current_table}: {str(err)}")
            continue