DEV Community: Vignesh

How to Add a Data Quality Gate to Your Airflow Pipeline in 5 Minutes

Vignesh — Fri, 17 Apr 2026 01:01:30 +0000

Your Airflow DAG ran successfully last night. Your dashboard is broken this morning.

Somewhere between the upstream API and your warehouse, a field went null, a type changed, or a schema drifted — and nothing caught it. dbt tests and Great Expectations are great, but they run after the data is already written. By the time they flag an issue, bad rows have been in production for hours.

This post shows how to add a real-time quality gate before the write happens — using DataScreenIQ inside an Airflow DAG.

The problem with post-insert validation

Most data quality tools follow the same pattern:

Extract → Load → Transform → Test

The test step catches issues — but only after the data is already in your warehouse. The feedback loop is slow. The damage is already done.

What we actually want is:

Extract → Screen → Load (only if clean) → Transform

Setup

pip install datascreeniq apache-airflow

Get a free API key at datascreeniq.com (500K rows/month,
no credit card).

Set your environment variable:

export DATASCREENIQ_API_KEY=dsiq_live_...

The basic quality gate task

import datascreeniq as dsiq
from datascreeniq.exceptions import DataQualityError

def quality_gate(rows: list, source: str) -> dict:
    """Drop this into any DAG as a quality gate."""
    client = dsiq.Client()  # reads DATASCREENIQ_API_KEY from env
    report = client.screen(rows, source=source)

    if report.is_blocked:
        raise ValueError(
            f"Data quality gate FAILED for '{source}': "
            f"health={report.health_pct}, "
            f"issues={report.type_mismatches or report.null_rates}"
        )

    if report.is_warn:
        import logging
        logging.warning(
            f"Data quality WARNING for '{source}': {report.summary()}"
        )

    return report.to_dict()

Full Airflow DAG example

from airflow import DAG
from airflow.decorators import task
from airflow.utils.dates import days_ago
import datascreeniq as dsiq
import requests

default_args = {"owner": "airflow", "retries": 1}

with DAG(
    dag_id="orders_pipeline_with_quality_gate",
    default_args=default_args,
    schedule_interval="@daily",
    start_date=days_ago(1),
    catchup=False,
) as dag:

    @task
    def extract_orders() -> list:
        """Extract rows from upstream API."""
        response = requests.get("https://api.example.com/orders")
        return response.json()["data"]

    @task
    def screen_data_quality(rows: list) -> dict:
        """
        Quality gate — screens rows before they reach the warehouse.
        Raises ValueError on BLOCK, logs warning on WARN.
        """
        client = dsiq.Client()
        report = client.screen(rows, source="orders")

        if report.is_blocked:
            raise ValueError(
                f"Pipeline blocked: {report.summary()}\n"
                f"Type mismatches: {report.type_mismatches}\n"
                f"Null rates: {report.null_rates}"
            )

        if report.is_warn:
            print(f"WARNING: {report.summary()}")

        return report.to_dict()

    @task
    def load_to_warehouse(rows: list, quality_report: dict):
        """
        Only runs if quality gate passed.
        quality_report is passed for logging/auditing.
        """
        print(f"Loading {len(rows)} rows — quality: {quality_report['status']}")
        # your warehouse load logic here

    # DAG wiring
    rows = extract_orders()
    report = screen_data_quality(rows)
    load_to_warehouse(rows, report)

What the report looks like

When a batch has issues, you get a structured report back:

# report.summary() output:
🚨 BLOCK | Health: 34.0% | Rows: 1200 | 
Type mismatches: amount | Null rate: email=67% | (8ms)

# Structured access:
report.status           # "BLOCK"
report.health_pct       # "34.0%"
report.type_mismatches  # ["amount"]
report.null_rates       # {"email": 0.67}
report.drift            # schema drift events vs baseline

Handling the BLOCK in production

When the gate blocks, you have a few options depending on
your pipeline:

@task
def screen_data_quality(rows: list) -> dict:
    client = dsiq.Client()
    report = client.screen(rows, source="orders")

    if report.is_blocked:
        # Option 1: Halt the pipeline (raises, DAG fails)
        raise ValueError(f"Quality gate failed: {report.summary()}")

        # Option 2: Send to dead-letter queue instead
        # send_to_dlq(rows, reason=report.summary())
        # return {"status": "QUARANTINED", "report": report.to_dict()}

        # Option 3: Alert and continue (not recommended for critical data)
        # alert_slack(report.summary())

    return report.to_dict()

Drift detection across batches

After the first run, every subsequent batch is compared against your stored schema baseline.

This catches issues like:

A field that was always number suddenly arriving as string
A column that was never null now showing 40% nulls
A new field appearing that wasn't in the original schema

# drift events in the report
for event in report.drift:
    print(f"{event['kind']}: {event['field']} — {event['detail']}")

# field_added: user_age — New field not in previous schema
# type_changed: amount — number → string  
# null_spike: email — null rate increased from 2% to 67%

Custom thresholds per source

Different sources have different tolerance levels. Override
the defaults inline:

report = client.screen(
    rows,
    source="orders",
    options={
        "thresholds": {
            "null_rate_warn": 0.05,   # warn if >5% nulls
            "null_rate_block": 0.2,   # block if >20% nulls
            "type_mismatch_block": 0.05,
        }
    }
)

Screening a CSV or DataFrame instead

If your DAG extracts files rather than API rows:

# CSV file
report = client.screen_file("orders.csv", source="orders")

# pandas DataFrame
import pandas as pd
df = pd.read_csv("orders.csv")
report = client.screen_dataframe(df, source="orders")

Summary

The pattern is simple:

Extract your data as usual
Pass rows through client.screen() before the load step
Raise on BLOCK, log on WARN, proceed on PASS
Every subsequent batch gets drift-checked against your baseline automatically

No infrastructure changes. No new database. No config files.
Just a quality gate that takes 5 minutes to add and catches
issues your downstream tests never will.

Install:

pip install datascreeniq

Free tier: 500K rows/month — datascreeniq.com

SDK + more examples:
github.com/AppDevIQ/datascreeniq-python

Have questions or a different integration pattern you'd like
covered? Drop a comment below.

How I stopped bad data from reaching my warehouse using a single Airflow task

Vignesh — Tue, 07 Apr 2026 03:02:37 +0000

Same pattern, third time in six months.

DAG runs green. Airflow shows all tasks succeeded. I wake up to a Slack message saying the revenue dashboard is broken. I dig through logs, trace it back to an extract job from four hours ago — a field that was always numeric now has strings mixed in. 48,000 rows loaded. Every downstream model is wrong.

The fix was obvious in hindsight: the quality check needs to happen before the INSERT, not after it. dbt tests, Great Expectations, warehouse constraints — they're all good tools, but they validate data that's already been written. By the time they flag an issue, the damage is done.

So I built a quality gate task that sits between extract and load. Here's exactly what I did.

The pattern

Before:

extract → load → warehouse ✓
                 [ bad rows sitting in production ]
                 [ dashboard broken at 9am ]

After:

extract → screen → PASS → load → warehouse ✓
                 → WARN → load + flag for review
                 → BLOCK → dead-letter queue, pipeline stops

One task. No custom validation logic. No rules to write and maintain.

Setup

Install the SDK:

pip install datascreeniq

Add your API key to Airflow Variables or environment (free tier at datascreeniq.com — 500K rows/month, no credit card):

export DATASCREENIQ_API_KEY="dsiq_live_..."

The DAG

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
import datascreeniq as dsiq
from datascreeniq.exceptions import DataQualityError

@dag(schedule_interval="@hourly", start_date=days_ago(1), catchup=False)
def orders_pipeline():

    @task
    def extract():
        return fetch_from_source()  # your existing extract logic

    @task
    def quality_gate(rows):
        client = dsiq.Client()  # reads DATASCREENIQ_API_KEY from env

        try:
            client.screen(rows, source="orders").raise_on_block()
        except DataQualityError as e:
            # Raises AirflowException — task fails, DAG stops
            raise ValueError(f"Quality gate failed: {e.report.summary()}")

        return rows  # only passes through on PASS or WARN

    @task
    def load(rows):
        insert_to_warehouse(rows)  # only runs if quality_gate passed

    rows = extract()
    clean = quality_gate(rows)
    load(clean)

dag = orders_pipeline()

That's the whole thing. Three tasks, one new one in the middle.

What it actually checks

The screen call runs 18 checks in a single pass on a deterministic sample of your rows:

Null rates — columns with too many missing values
Type mismatches — a field that's always been numeric now has strings
Schema drift — new fields appeared, known fields disappeared, a type changed from the baseline
Empty string rates — fields full of "" instead of null
Duplicate rates — cardinality collapse
Outliers — numeric values beyond 1.5× IQR
Timestamp staleness — most recent timestamp older than expected

The response comes back in under 10ms (runs on Cloudflare Workers at the edge, no cold starts). Your pipeline barely notices it.

The response looks like this:

{
  "status": "BLOCK",
  "health_score": 0.34,
  "decision": {
    "action": "BLOCK",
    "reason": "Type mismatch in 'amount'; High null rate in 'email' (50%)"
  },
  "issues": {
    "type_mismatches": {
      "amount": {
        "expected": "number",
        "found": ["string"],
        "sample_value": "broken",
        "severity": "critical"
      }
    },
    "null_rates": {
      "email": { "actual": 0.50, "threshold": 0.30, "severity": "critical" }
    }
  },
  "drift": [
    {
      "field": "amount",
      "kind": "type_changed",
      "severity": "block",
      "detail": "Field 'amount' changed from number to mixed"
    }
  ],
  "latency_ms": 7
}

The decision.reason field is what ends up in your Airflow task failure message — so when the pipeline stops, you know immediately what went wrong and in which field.

Handling WARN vs BLOCK

By default:

PASS — pipeline continues, nothing logged
WARN — pipeline continues, issue is in the report
BLOCK — raise_on_block() raises DataQualityError, task fails, DAG stops

If you want WARN to also stop the pipeline (useful for financial data pipelines where you want zero tolerance):

@task
def quality_gate(rows):
    client = dsiq.Client()
    report = client.screen(rows, source="payments")

    if report.is_warn:
        raise ValueError(f"Quality warning on payments: {report.summary()}")

    if report.is_blocked:
        raise ValueError(f"Quality blocked on payments: {report.summary()}")

    return rows

Per-source thresholds

The default thresholds are reasonable for most sources — WARN at 30% null, BLOCK at 70%. But a payments pipeline and an events pipeline have completely different tolerances.

You can set per-source thresholds inline:

# Financial data — very tight
client.screen(
    rows,
    source="payments",
    options={
        "thresholds": {
            "null_rate_warn":      0.01,  # WARN if > 1% nulls
            "null_rate_block":     0.02,  # BLOCK if > 2% nulls
            "type_mismatch_warn":  0.0,   # WARN on any type mismatch at all
            "type_mismatch_block": 0.01,
        }
    }
)

# Event tracking — high null rate is normal
client.screen(
    rows,
    source="events",
    options={
        "thresholds": {
            "null_rate_warn":  0.50,  # user_agent nulls are expected
            "null_rate_block": 0.80,
        }
    }
)

Or set them once in the dashboard and they apply automatically to every future call for that source.

The part that surprised me

The obvious failures — the ones you'd catch anyway — aren't what makes this worth it.

What makes it worth it is the slow drift. A null rate creeping up 2% per week. A field that was 99% numeric and is now 94% numeric. Nobody notices either of those until it's been wrong for a month and your ML model has been training on garbage.

Because the baseline is set on the first run and every subsequent batch is compared against it, even small shifts get caught. That's the thing dbt tests don't do — they test what you wrote rules for. This catches what you didn't know to write rules for.

Resources

Free tier is 500K rows/month. Enough for most pipelines to run for months before needing to upgrade.

Curious how others are handling this. Are you doing quality checks pre-load, post-load, or both? And if pre-load — what are you using?

How I built a MilliSeconds data quality firewall on Cloudflare Workers

Vignesh — Wed, 01 Apr 2026 22:34:59 +0000

I got paged at 6am twice in three months because of the same class of problem.

Upstream API changed a field type. My pipeline ingested the bad data without complaint. Six hours later, a dashboard was showing NaN totals and an ML model was producing garbage predictions. The root cause was at ingestion — but the symptom showed up at reporting.

The fix was always fast. The detection was always late.

I went looking for something that would sit between my data sources and my database, reject bad payloads at the gate, and tell me exactly what was wrong — in real time, before anything touched storage. I couldn't find something that was simple enough to drop into an existing pipeline with a single function call.

So I built it. This is what I learned.

The architecture problem with batch validation

The standard data quality approach looks like this:

Extract → Load → Transform → Test (Great Expectations / dbt) → Alert

The test runs after the INSERT. By the time it fires, bad data has been sitting in your warehouse for however long your pipeline takes to run. Your recovery options are: delete and re-ingest (if the source data hasn't changed), patch bad rows manually, mark a time range as unreliable, or silently move on. None of these are good.

What I wanted was this:

Source → [Quality Gate] → PASS  → Database
                        → WARN  → Quarantine
                        → BLOCK → Dead-letter queue

The check before the INSERT, not after. Every batch screened. Every problem caught with enough information to act on immediately.

Why Cloudflare Workers

The constraints were:

Sub-10ms — can't add meaningful latency to a pipeline
Stateless — no infrastructure to manage
In-memory only — can't write raw payload data anywhere
Global — needs to run close to the source

Cloudflare Workers fit all four. The critical privacy property is architectural, not policy: Workers have no filesystem access. There is no fs.writeFile. Raw payload values physically cannot be persisted at the edge layer — the runtime makes it impossible. What I write to persistent storage (KV, D1, Durable Objects) is exclusively post-aggregation: schema hashes, null rates, type distributions, quality scores.

The full Cloudflare stack I ended up using:

# wrangler.toml
[[kv_namespaces]]
binding = "KV"            # schema hash cache + API keys

[[d1_databases]]
binding = "DB"            # job history + QA summaries (control plane only)

[durable_objects]
bindings = [{ name = "TENANT_DO", class_name = "TenantDurableObject" }]
           # authoritative schema state + billing counters, one DO per tenant

[[queues.producers]]
queue = "datascreeniq-overflow"  # large payload overflow

The hot path: what happens in under 10ms

Every request to POST /v1/screen runs this sequence:

1. Parse payload (JSON or CSV)
2. Deterministic hash-based sampling
3. Single-pass column analysis
4. Build schema fingerprint (SHA-256)
5. KV cache lookup for previous schema hash
6. Drift detection vs Durable Object baseline
7. Health score calculation
8. Return PASS / WARN / BLOCK + full report
9. (fire-and-forget) Update DO state, save to D1, fire webhooks

Steps 1–8 are synchronous and block the response. Step 9 runs in waitUntil() after the response is already sent. That's how you hit sub-10ms while still persisting state and delivering webhook alerts.

Deterministic sampling — the part that matters most

If you're running statistical checks at the edge, you can't scan every row on large batches. You need to sample. But naive random sampling has a problem: it's not reproducible. If a customer asks "why didn't you catch that null explosion last Tuesday?", you can't answer.

The sampling is deterministic: seeded from a hash of the source name + field names + row count.

export async function payloadHash(
  source: string,
  rows: Record<string, unknown>[]
): Promise<string> {
  const firstRowKeys = rows.length > 0
    ? Object.keys(rows[0]).sort().join(",")
    : "";
  const input = `${source}:${firstRowKeys}:${rows.length}`;
  return sha256(input);
}

export function deterministicSample<T>(
  rows: T[],
  sampleSize: number,
  datasetHash: string
): { sampled: T[]; ratio: number } {
  const seed = parseInt(datasetHash.substring(0, 8), 16);
  const ratio = sampleSize / rows.length;
  const sampled: T[] = [];

  for (let i = 0; i < rows.length; i++) {
    const h = mulberry32(seed ^ (i * 2654435761));
    if (h < ratio) {
      sampled.push(rows[i]);
      if (sampled.length >= sampleSize) break;
    }
  }
  return { sampled, ratio: sampled.length / rows.length };
}

The sampling version ("v2") and ratio are returned in every API response. Any quality decision is permanently auditable — you can always show exactly what was checked and under which strategy.

Single-pass column analysis

The analyzer runs one loop over all sampled rows and builds per-column statistics simultaneously. No second passes, no intermediate data structures beyond what's needed per column.

For each column it computes:

Null rate and empty string rate (separate — they're different problems)
Dominant type + type mismatch rate (what fraction of values aren't the expected type)
Min/max (lexicographic for strings, numeric for numbers)
p25/p50/p75/p95 via sort + linear interpolation (numeric only)
Outlier count via IQR method (values outside [Q1 - 1.5×IQR, Q3 + 1.5×IQR])
Approximate distinct count via HyperLogLog
Duplicate rate derived from distinct count vs total
Enum tracking for low-cardinality string fields (≤20 unique values)
Timestamp detection — ISO 8601 strings auto-detected, recency checked

The HyperLogLog implementation uses p=14 (16,384 registers), giving ~0.8% standard error on distinct count estimates. It runs MurmurHash3 internally — no external dependency, pure TypeScript.

const P = 14;
const M = 1 << P;  // 16,384 registers
const ALPHA = 0.7213 / (1 + 1.079 / M);

// add() hashes the value, uses top P bits as register index,
// stores the position of the leftmost 1-bit in the remaining bits.
// estimate() uses harmonic mean of register values with bias correction.

Drift detection: 8 signal types

After the column analysis, the current batch is compared against the baseline stored in the tenant's Durable Object. The DO stores: schema fingerprints per source, null rate baselines, known enum value sets, and a rolling 20-batch row count history.

The drift signals, in order of severity:

Signal	Trigger	Severity
`type_changed`	Field type changed from baseline	BLOCK
`row_count_anomaly`	Batch size >10× from average	BLOCK
`null_spike`	Null rate increased >20% from baseline	WARN or BLOCK
`timestamp_stale`	Most recent timestamp >72h old	BLOCK
`field_added`	New field not in previous schema	WARN
`field_removed`	Known field missing	WARN
`empty_string_spike`	Empty string rate >30%	WARN
`new_enum_value`	New value in tracked enum field	WARN
`timestamp_stale`	Most recent timestamp >24h old	WARN

Type changes are always BLOCK because they silently corrupt downstream aggregations. A field that was a number and is now a string will cause every SUM and AVG query on it to fail or return wrong results — often without an error, just a wrong answer.

The two-layer state architecture

The tenant state has two components with different consistency guarantees:

KV (eventually consistent, ~1ms reads): Schema hash cache. Used for the fast path — if the hash matches the cached value, we skip the DO round-trip. TTL of 24 hours.

Durable Objects (strongly consistent, ~5ms): Source of truth for schema fingerprints, null rate baselines, billing counters, enum sets, row count history. One DO per tenant, keyed as do:tenant:{id}. This was a deliberate design choice after reading the Cloudflare reviewer feedback that using KV as source-of-truth for schema enforcement would be wrong — propagation delays of seconds to minutes mean you could miss drift events during the window.

// Fast path: KV hash match → skip DO
const cachedHash = await env.KV.get(`schema_hash:${tenantId}:${source}`);
if (cachedHash && cachedHash === fingerprint.hash) {
  // Schema stable — still fetch DO for null baselines, but common path is fast
  tenantState = await fetchTenantState(env, tenantId);
} else {
  // Schema changed or cache miss — full DO state comparison
  tenantState = await fetchTenantState(env, tenantId);
}

Health score and verdict

The health score is a penalty-based system starting at 1.0:

// Column-level penalties
if (col.nullRate > thresholds.null_rate_warn)          score *= (1 - col.nullRate * 0.3);
if (col.typeMismatchRate > thresholds.type_mismatch_warn) score *= (1 - col.typeMismatchRate * 0.5);
if (col.emptyStringRate > 0.2)                         score *= (1 - col.emptyStringRate * 0.15);
if (col.duplicateRate > 0.1)                           score *= (1 - col.duplicateRate * 0.1);

// Drift event penalties
// BLOCK severity: −20% | WARN: −8% | INFO: −2%

Penalties are multiplicative (diminishing returns) so a single catastrophic column can't push the score below 0. The verdict logic:

Any BLOCK drift event          → BLOCK
Health < 0.5                   → BLOCK
Health < 0.8 OR any WARN drift → WARN
Otherwise                      → PASS

Default thresholds are tuned for general use but overridable per-request via options.thresholds.

The Python SDK

The backend is TypeScript on Cloudflare Workers. The client-facing SDK is Python, published to PyPI.

pip install datascreeniq

import datascreeniq as dsiq

client = dsiq.Client("dsiq_live_...")

rows = [
    {"order_id": "ORD-001", "amount": 99.50,    "email": "alice@corp.com"},
    {"order_id": "ORD-002", "amount": "broken", "email": None},
    {"order_id": "ORD-003", "amount": 75.00,    "email": None},
]

report = client.screen(rows, source="orders")

print(report.status)     # BLOCK
print(report.summary())
# 🚨 BLOCK | Health: 34.0% | Type mismatches: amount | Null rate: email=67% | (7ms)

The raise_on_block() pattern integrates cleanly into any orchestrator:

# Airflow
@task
def quality_gate(rows, source):
    report = dsiq.Client().screen(rows, source=source)
    if report.is_blocked:
        raise ValueError(report.summary())
    return report.to_dict()

# Prefect
@task
def screen_data(rows, source):
    dsiq.Client().screen(rows, source=source).raise_on_block()

What I got wrong (and had to fix)

Cold start problem. The first batch from a new source has no baseline to compare against. The first call always returns PASS regardless of content — you need at least 2–3 batches before drift detection has meaning. I underestimated how often this confuses new users who test with a single call and assume PASS means "healthy data."

Sampling under-coverage edge cases. The hash-based sampler can undersample in rare cases when the hash distribution is unlucky for a specific seed + row count combination. I added a fallback: if the sampler returns fewer than sampleSize rows, it fills up using a fixed stride through the remaining rows. The fallback kicks in rarely but the determinism guarantee holds.

KV as source-of-truth was wrong. My original design used KV for schema enforcement. I got feedback that KV's eventual consistency model (propagation delays up to 60 seconds at the edge) meant you could miss drift events during the propagation window. Moved authoritative schema state to Durable Objects — KV is now read cache only.

Null vs empty string. These are different problems. A null means the value was absent. An empty string means someone sent "" instead of null — often a serialisation bug, sometimes intentional. Tracking them separately turned out to be important for downstream consumers who handle them differently.

Where it is now

Live at datascreeniq.com. Python SDK on PyPI (pip install datascreeniq). Free tier is 500K rows/month, no credit card.

GitHub: github.com/AppDevIQ/datascreeniq-python

If you're running pipelines that ingest from upstream APIs and have been burned by schema drift, I'd genuinely appreciate feedback — especially on the cold start handling, and on what thresholds make sense for different data domains (financial data vs user events vs IoT streams all have very different null rate norms).

Questions or issues → app
@datascreeniq.com

Add to the squad at https://dly.to/G92nQDoTm8N

How I built a data quality API that runs at the edge in milliseconds

Vignesh — Mon, 30 Mar 2026 09:10:01 +0000

Bad data is quiet. That's the problem.

Your pipeline doesn't crash. Your tests pass. Three weeks later someone notices the revenue dashboard is wrong, you trace it back, and find that one column started arriving as strings six weeks ago. The ETL swallowed it. The warehouse stored it. Everything looked fine.
I've debugged enough of these to know the pattern. The fix is always obvious in retrospect — validate the data before it enters the system. The hard part is actually doing it without adding another tool, another config file, another YAML-driven framework to maintain.
So I built DataScreenIQ — a data quality API. You POST rows, you get a verdict back. No setup. No infrastructure. One call.

curl -X POST https://api.datascreeniq.com/v1/screen \
  -H "X-API-Key: dsiq_live_..." \
  -H "Content-Type: application/json" \
  -d '{
    "source": "orders",
    "rows": [
      {"order_id": "ORD-001", "amount": 99.50, "email": "alice@corp.com"},
      {"order_id": "ORD-002", "amount": "broken", "email": null},
      {"order_id": "ORD-003", "amount": 75.00,   "email": null}
    ]
  }'

Response

{
  "status": "BLOCK",
  "health_score": 0.34,
  "issues": {
    "type_mismatches": ["amount"],
    "null_rates": {"email": 0.67}
  },
  "drift": [
    {
      "kind": "type_changed",
      "field": "amount",
      "detail": "Field changed type from number to mixed"
    }
  ],
  "latency_ms": 38
}

Three verdicts: PASS, WARN, or BLOCK. Your pipeline decides what to do with it.

The 18 checks it runs

Every payload gets:

Schema fingerprinting (SHA-256 hash of field structure)
Null rate per column
Type stability (what % of values match the expected type)
Empty string rate
Duplicate detection
IQR outlier detection on numeric columns
HyperLogLog approximate distinct counts
Enum cardinality tracking (new values appearing)
Row count anomaly detection
Schema drift against the established baseline

The drift detection is the interesting bit. On first run with a new source, it builds a baseline — field types, null rates, schema fingerprint. On every subsequent run it compares the incoming data against that baseline. If your amount field was numeric for six weeks and suddenly 40% of values are strings, that fires a BLOCK.

The Python SDK

pip install datascreeniq

import datascreeniq as dsiq
from datascreeniq.exceptions import DataQualityError

client = dsiq.Client("dsiq_live_...")

# Screen a list of dicts
report = client.screen(rows, source="orders")
print(report.status)       # PASS / WARN / BLOCK
print(report.health_pct)   # 34.0%

# Raise on block — useful as a pipeline gate
try:
    client.screen(rows, source="orders").raise_on_block()
    load_to_warehouse(rows)
except DataQualityError as e:
    print(f"Blocked: {e.report.issues}")
    send_to_dead_letter_queue(rows)

Airflow Integration

from airflow.decorators import task
import datascreeniq as dsiq

@task
def quality_gate(rows: list, source: str) -> dict:
    client = dsiq.Client()  # reads DATASCREENIQ_API_KEY from env
    report = client.screen(rows, source=source)
    if report.is_blocked:
        raise ValueError(f"Data blocked: {report.summary()}")
    return report.to_dict()

@dag
def my_pipeline():
    raw = extract()
    quality_gate(raw, source="orders")
    load(raw)

Screening CSV files directly
The API accepts raw CSV — no conversion needed:

curl -X POST https://api.datascreeniq.com/v1/screen \
  -H "Content-Type: text/csv" \
  -H "X-API-Key: dsiq_live_..." \
  -H "X-Source: orders" \
  --data-binary @orders.csv

The SDK handles CSV, Excel, JSON and XML files too:

report = client.screen_file("orders.csv", source="orders")
report = client.screen_file("orders.xlsx", source="orders")

# pandas DataFrame
import pandas as pd
df = pd.read_csv("orders.csv")
report = client.screen_dataframe(df, source="orders")

Large files get chunked automatically — 10K rows per request, merged into one report.

Why the edge

The API runs on Cloudflare Workers — JavaScript V8 isolates deployed globally. A few things fall out of this naturally:
Raw payload data never touches persistent storage. The Worker reads the rows, computes the statistics, returns the response, and the runtime ends. No database write of your actual data, just the aggregated metrics (null rates, type distributions, schema hashes).
Latency is 30-50ms end to end. The compute itself is under 10ms — the rest is network round trip.
It scales to zero and to high throughput without any configuration on your end.

Resetting baselines

If you've fixed your pipeline and want to start fresh:

client.reset_baseline("orders")

Or via curl:

curl -X DELETE https://api.datascreeniq.com/v1/schema/orders \
  -H "X-API-Key: dsiq_live_..."

Next screen call builds a new baseline from scratch.

Slack alerts

You can configure a Slack incoming webhook in the dashboard — any BLOCK or WARN verdict fires an alert to your channel automatically. Useful if you're running scheduled pipelines and want to know when something breaks without polling the API.

What it doesn't replace
Great Expectations if you need declarative test suites with dozens of custom expectations. Monte Carlo if you need table-level monitoring across your entire warehouse with lineage tracking. dbt tests if your checks live in the transformation layer.
It's a different thing — a lightweight gate you drop in front of a data ingestion point. One call, synchronous, returns immediately. Your pipeline blocks or passes based on the verdict.

Try it
Free tier: 500K rows/month, no credit card.

# Get an API key
# datascreeniq.com

pip install datascreeniq

python3 -c "
import datascreeniq as dsiq
client = dsiq.Client('your_key')
print(client.health())
"

GitHub: