Testing

Soda Core Data Quality Testing: Validate Data in CI/CD Pipelines

HelpMeTest

17 May 2026 — 3 min read

Soda Core is an open-source data quality framework that lets you define checks in SodaCL (Soda Checks Language) — a YAML-based DSL — and run them against your database or data lake. Unlike Great Expectations, Soda Core uses declarative YAML instead of Python code, making checks readable by data analysts and engineers alike. This guide covers how to integrate Soda Core into your data pipeline testing.

Installation

pip install soda-core soda-core-postgres soda-core-duckdb

Basic Configuration

# configuration.yml
data_source orders_db:
  type: postgres
  connection:
    host: ${DATABASE_HOST}
    port: 5432
    username: ${DATABASE_USER}
    password: ${DATABASE_PASSWORD}
    database: orders
    schema: public

Writing SodaCL Checks

# checks/orders.yml
checks for orders:
  - row_count > 0:
      name: "Orders table is not empty"
  - missing_count(order_id) = 0:
      name: "No missing order IDs"
  - missing_count(customer_id) = 0:
      name: "No missing customer IDs"
  - invalid_count(status) = 0:
      name: "All statuses are valid"
      valid values:
        - pending
        - processing
        - completed
        - cancelled
        - refunded
  - min(amount) >= 0:
      name: "No negative order amounts"
  - duplicate_count(order_id) = 0:
      name: "Order IDs are unique"
  - freshness(created_at) < 25h:
      name: "Orders table updated within 25 hours"
  - schema:
      name: "Required columns present with correct types"
      fail:
        when required column missing:
          - order_id
          - customer_id
          - amount
          - status
          - created_at
        when wrong column type:
          order_id: varchar
          amount: numeric

Running Checks

soda scan -d orders_db -c configuration.yml checks/orders.yml

Exit codes: 0 = all passed, 1 = checks failed, 2 = scan error.

Advanced SodaCL: Cross-Dataset Referential Integrity

# checks/referential_integrity.yml
checks for order_items:
  - reference:
      name: "All order items reference valid orders"
      column: order_id
      must exist in:
        dataset: orders
        column: order_id
  - reference:
      name: "All order items reference valid products"
      column: product_id
      must exist in:
        dataset: products
        column: id

Python API for Pytest Integration

# test_data_quality.py
import pytest
from soda.scan import Scan

def run_soda_checks(checks_yaml: str, connection_config: str) -> dict:
    scan = Scan()
    scan.set_scan_definition_name("pytest-checks")
    scan.set_data_source_name("test_db")
    scan.add_configuration_yaml_str(connection_config)
    scan.add_sodacl_yaml_str(checks_yaml)
    scan.execute()
    return {
        'passed': scan.get_checks_pass_count(),
        'failed': scan.get_checks_fail_count(),
        'all_passed': scan.get_checks_fail_count() == 0,
        'errors': scan.get_error_logs_text(),
    }

PG_CONFIG = """
data_source test_db:
  type: postgres
  connection:
    host: localhost
    port: 5432
    username: test
    password: test
    database: testdb
"""

def test_orders_completeness():
    results = run_soda_checks("""
        checks for orders:
          - row_count > 0
          - missing_count(order_id) = 0
          - missing_count(customer_id) = 0
    """, PG_CONFIG)
    assert results['all_passed'], f"{results['failed']} checks failed: {results['errors']}"

def test_orders_validity():
    results = run_soda_checks("""
        checks for orders:
          - invalid_count(status) = 0:
              valid values: [pending, processing, completed, cancelled]
          - min(amount) >= 0
          - duplicate_count(order_id) = 0
    """, PG_CONFIG)
    assert results['all_passed']

def test_orders_freshness():
    results = run_soda_checks("""
        checks for orders:
          - freshness(created_at) < 25h
    """, PG_CONFIG)
    assert results['all_passed'], "Orders data is stale"

Testing with DuckDB (No Server Required)

# test_dq_duckdb.py
import duckdb
from soda.scan import Scan

DUCKDB_CONFIG = """
data_source test_db:
  type: duckdb
  database: ":memory:"
"""

def test_invalid_status_fails_check():
    # Create table with bad data
    conn = duckdb.connect(':memory:')
    conn.execute("""
        CREATE TABLE orders (order_id VARCHAR, status VARCHAR, amount DECIMAL);
        INSERT INTO orders VALUES ('o1', 'invalid_status', 100);
    """)

    scan = Scan()
    scan.set_scan_definition_name("test")
    scan.set_data_source_name("test_db")
    scan.add_configuration_yaml_str(DUCKDB_CONFIG)
    scan.add_sodacl_yaml_str("""
        checks for orders:
          - invalid_count(status) = 0:
              valid values: [pending, completed, cancelled]
    """)
    scan.execute()

    assert scan.get_checks_fail_count() == 1
    conn.close()

def test_valid_data_passes_all_checks():
    conn = duckdb.connect(':memory:')
    conn.execute("""
        CREATE TABLE orders (order_id VARCHAR, customer_id VARCHAR, amount DECIMAL(10,2), status VARCHAR, created_at TIMESTAMP);
        INSERT INTO orders VALUES
            ('o1', 'c1', 100.0, 'completed', CURRENT_TIMESTAMP),
            ('o2', 'c2', 200.0, 'pending', CURRENT_TIMESTAMP);
    """)

    scan = Scan()
    scan.set_scan_definition_name("test")
    scan.set_data_source_name("test_db")
    scan.add_configuration_yaml_str(DUCKDB_CONFIG)
    scan.add_sodacl_yaml_str("""
        checks for orders:
          - row_count > 0
          - missing_count(order_id) = 0
          - min(amount) >= 0
          - duplicate_count(order_id) = 0
    """)
    scan.execute()

    assert scan.get_checks_fail_count() == 0
    assert scan.get_checks_pass_count() == 4
    conn.close()

CI/CD Integration

# .github/workflows/data-quality.yml
name: Data Quality Checks
on:
  push:
    paths: ['data/**', 'checks/**']
  schedule:
    - cron: '0 6 * * *'

jobs:
  data-quality:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: test
          POSTGRES_DB: orders
        ports: ['5432:5432']
        options: --health-cmd pg_isready --health-interval 10s
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install soda-core soda-core-postgres
      - name: Seed database
        run: PGPASSWORD=test psql -h localhost -U postgres -d orders -f tests/seed.sql
      - name: Run data quality checks
        run: |
          soda scan \
            -d orders_db \
            -c checks/configuration.yml \
            checks/orders.yml \
            checks/referential_integrity.yml
        env:
          DATABASE_HOST: localhost
          DATABASE_USER: postgres
          DATABASE_PASSWORD: test

Post-dbt Integration

#!/bin/bash
set -e
dbt run --profiles-dir . --target prod
soda scan -d production_db -c soda/configuration.yml soda/checks/after_dbt_run.yml
echo "All data quality checks passed"

# soda/checks/after_dbt_run.yml
checks for fct_orders:
  - row_count > 0
  - duplicate_count(order_id) = 0
  - missing_count(customer_id) = 0
  - invalid_count(order_status) = 0:
      valid values: [placed, shipped, delivered, returned]
  - freshness(dbt_updated_at) < 2h:
      name: "dbt model was run within the last 2 hours"

checks for dim_customers:
  - row_count > 0
  - duplicate_count(customer_id) = 0
  - missing_percent(email) < 5

Soda Core vs Alternatives

Tool	Check Format	Language	Best For
Soda Core	YAML (SodaCL)	Any	YAML-first, analyst-friendly
Great Expectations	Python code	Python	Complex Python-based checks
dbt tests	YAML + SQL	SQL	Tests inside dbt projects
Amazon Deequ	Scala/Python	Spark	Large-scale Spark datasets
Pandera	Python schemas	Python	DataFrame validation in code

Soda Core's YAML-first approach makes checks reviewable in PRs by analysts and engineers without Python knowledge. The checks become executable data contracts that you can enforce in CI before data reaches production.