soda-core-bigquery

import os from soda.scan import Scan # Configure your GCP project ID. For local execution, ensure: # 1. 'GOOGLE_APPLICATION_CREDENTIALS' env var points to a service account key JSON, # OR 2. 'gcloud auth application-default login' has been run. # os.environ['BIGQUERY_PROJECT_ID'] = 'your-gcp-project-id' # Initialize a Soda Scan scan = Scan() # Add BigQuery data source configuration via a YAML string # Replace 'my-dummy-project' with your actual GCP project ID scan.add_configuration_yaml_str(f''' data_sources: bigquery_source: type: bigquery project_id: {os.environ.get('BIGQUERY_PROJECT_ID', 'my-dummy-project')} ''') # Define data quality checks via a YAML string # Replace 'my_dataset.my_table' with an actual BigQuery dataset.table for real checks scan.add_checks_yaml_str(''' checks for my_dataset.my_table: - row_count > 0: # Checks if the table has any rows name: 'Table should not be empty' - missing_count(id) = 0: # Checks for missing values in 'id' column name: 'No missing IDs' - duplicate_count(id) = 0: # Checks for duplicate values in 'id' column name: 'No duplicate IDs' ''') print("Running Soda Scan...") scan.execute() if scan.has_failures(): print("\nSoda Scan finished with failures.") else: print("\nSoda Scan finished with no failures.") # You can inspect the scan results for detailed outcomes # print(scan.get_scan_results())