SnowflakeDataQualityGateOperator

An Apache Airflow provider that loads data from any cloud storage (S3, GCS, Azure) into Snowflake with a built-in data quality gate. Bad data is caught in a transient staging table before it ever touches your production tables.

Why?

Most cloud-to-Snowflake pipelines blindly COPY INTO production. When bad data lands, you find out hours later from a broken dashboard. This operator enforces a stage → validate → promote pattern in a single, reusable task.

Installation

pip install airflow-provider-s3-snowflake-quality

Quick start

from airflow_s3_snowflake_quality.operators import SnowflakeDataQualityGateOperator
from airflow_s3_snowflake_quality.checks import (
    RowCountCheck, NullCheck, UniquenessCheck,
    FreshnessCheck, AcceptedValuesCheck, CustomSQLCheck,
)

load_orders = SnowflakeDataQualityGateOperator(
    task_id="load_orders",
    source_path="s3://my-data-lake/orders/dt={{ ds }}/",
    file_format="PARQUET",
    snowflake_conn_id="snowflake_prod",
    target_database="ANALYTICS",
    target_schema="PUBLIC",
    target_table="ORDERS",
    load_strategy="merge",
    merge_keys=["ORDER_ID"],
    storage_integration="s3_integration",
    quality_checks=[
        RowCountCheck(min=1000),
        NullCheck(column="ORDER_ID", max_fraction=0.0),
        UniquenessCheck(columns=["ORDER_ID"]),
        FreshnessCheck(column="ORDER_TS", max_age_hours=24),
        AcceptedValuesCheck(column="STATUS", values=["pending", "shipped", "delivered", "cancelled"]),
        CustomSQLCheck(sql="SELECT * FROM {staging_table} WHERE TOTAL_AMOUNT < 0"),
    ],
)

Works with GCS and Azure too — just change the source_path:

source_path="gcs://bucket/orders/dt={{ ds }}/"       # Google Cloud Storage
source_path="azure://container/orders/dt={{ ds }}/"   # Azure Blob

How it works

┌──────────┐     ┌────────────────────┐     ┌──────────┐     ┌────────────┐
│  Cloud   │────▶│  Staging Table     │────▶│ Quality  │────▶│ Production │
│  Storage │     │  (transient)       │     │ Checks   │     │ Table      │
└──────────┘     └────────────────────┘     └──────────┘     └────────────┘
  S3 / GCS                                       │ FAIL
  / Azure                                        ▼
                                            Task fails or
                                            returns report

1. Stage — COPY INTO a transient Snowflake table from cloud storage
2. Validate — Run every configured quality check against the staged data
3. Promote — If all checks pass, move data via append, replace, or merge
4. Cleanup — Staging table is always dropped, even on failure

Operator parameters

Quality checks

Writing custom checks

from airflow_s3_snowflake_quality.checks.base import BaseCheck, CheckResult, CheckStatus

class MyCheck(BaseCheck):
    def validate(self, cursor, staging_table) -> CheckResult:
        count = self._execute_scalar(cursor, f"SELECT COUNT(*) FROM {staging_table} WHERE ...")
        if count > 0:
            return CheckResult(check_name=self.name, status=CheckStatus.FAILED, message=f"Found {count} bad rows")
        return CheckResult(check_name=self.name, status=CheckStatus.PASSED, message="All good")

Testing

python3 -m venv venv && source venv/bin/activate
pip install -e ".[dev]" && pip install duckdb pyarrow

make all              # lint + typecheck + 62 pytest tests (97% coverage)
make test-local-e2e   # 5 end-to-end scenarios with DuckDB

See integration_test/ for full S3 + Snowflake + Airflow Docker-based testing.

License

Apache 2.0