Validate Manufacturing Data Pipelines with Great Expectations and DVC

                      
                     
import os
import dvc.api
from great_expectations.data_context import DataContext
from great_expectations.exceptions import GreatExpectationsError

# Configuration
class Config:
    def __init__(self):
        self.dvc_repo_url = os.getenv('DVC_REPO_URL')
        self.data_source = os.getenv('DATA_SOURCE')
        self.ge_project_path = os.getenv('GE_PROJECT_PATH')

# Initialize DataContext for Great Expectations
config = Config()

try:
    data_context = DataContext(config.ge_project_path)
except GreatExpectationsError as e:
    print(f"Error initializing Great Expectations: {str(e)}")

# Function to validate data pipeline
def validate_data_pipeline():
    try:
        # Load data with DVC
        with dvc.api.open(config.dvc_repo_url) as data:
            # Validate data using Great Expectations
            batch = data_context.get_batch(data)
            validator = data_context.get_expectation_suite('your_suite_name')
            validation_result = data_context.run_validation_operator(
                'default',
                run_id='batch_validation',
                assets_to_validate=[batch],
                expectation_suite_names=[validator]
            )
            return validation_result
    except Exception as e:
        print(f"Validation failed: {str(e)}")

if __name__ == '__main__':
    result = validate_data_pipeline()
    if result['success']:
        print("Validation successful!")
    else:
        print("Validation failed!")