Building a Bulletproof SaaS Data Quality Pipeline with DataForge API

pip install requests python-dotenv sqlalchemy redis

export RAPIDAPI_KEY="YOUR_API_KEY"

import requests
import os
from typing import Dict, Any, List, Optional
 
RAPIDAPI_KEY = os.environ["RAPIDAPI_KEY"]
DATAFORGE_HOST = "dataforge-data-tools.p.rapidapi.com"
 
HEADERS = {
    "X-RapidAPI-Key": RAPIDAPI_KEY,
    "X-RapidAPI-Host": DATAFORGE_HOST,
    "Content-Type": "application/json"
}
 
# Define your schema — this can be loaded from config/database
CUSTOMER_SCHEMA = {
    "fields": {
        "email": {
            "type": "email",
            "required": True,
            "constraints": {"max_length": 254}
        },
        "phone": {
            "type": "phone",
            "required": False,
            "normalization": "e164"  # Normalize to international format
        },
        "company_name": {
            "type": "string",
            "required": True,
            "constraints": {"min_length": 1, "max_length": 200}
        },
        "billing_address": {
            "type": "address",
            "required": False,
            "normalization": "usps_standard"
        },
        "mrr": {
            "type": "currency",
            "required": False,
            "constraints": {"min": 0, "max": 1000000}
        },
        "signup_date": {
            "type": "date",
            "required": True,
            "formats": ["ISO8601", "MM/DD/YYYY", "YYYY-MM-DD"]
        },
        "plan": {
            "type": "enum",
            "required": True,
            "allowed_values": ["starter", "growth", "professional", "enterprise"]
        }
    }
}
 
 
def validate_record(record: Dict[str, Any], schema: Dict) -> Dict[str, Any]:
    """Validate a single record against a schema definition."""
    response = requests.post(
        f"https://{DATAFORGE_HOST}/validate",
        headers=HEADERS,
        json={
            "record": record,
            "schema": schema,
            "options": {
                "strict_mode": False,  # Warn on extra fields, don't reject
                "coerce_types": True   # Try to convert types before failing
            }
        }
    )
    response.raise_for_status()
    return response.json()
 
 
# Test with a messy record
test_record = {
    "email": "[email protected]",         # Needs lowercasing
    "phone": "(555) 123-4567",             # Needs normalization
    "company_name": "Acme Corp",
    "billing_address": "123 main st new york ny 10001",  # Needs standardization
    "mrr": "1499",                          # String, should be number
    "signup_date": "03/15/2026",           # Non-ISO format
    "plan": "growth"
}
 
result = validate_record(test_record, CUSTOMER_SCHEMA)
print(f"Valid: {result['valid']}")
print(f"Errors: {result['errors']}")
print(f"Warnings: {result['warnings']}")
print(f"Normalized record: {result['normalized_record']}")

def normalize_record(record: Dict[str, Any], schema: Dict) -> Dict[str, Any]:
    """
    Normalize a record: standardize formats, fill in derivable fields.
    Returns the normalized record with a changelog.
    """
    response = requests.post(
        f"https://{DATAFORGE_HOST}/normalize",
        headers=HEADERS,
        json={
            "record": record,
            "schema": schema,
            "enrichment": {
                "enabled": True,
                "fill_from_postal_code": ["city", "state", "country"],
                "derive_domain_from_email": True
            }
        }
    )
    response.raise_for_status()
    return response.json()
 
 
def process_incoming_record(raw_record: Dict[str, Any]) -> Dict[str, Any]:
    """
    Full ingestion flow for a single record:
    1. Validate
    2. Normalize if valid (or partially valid)
    3. Return with status
    """
    # Step 1: Validate
    validation = validate_record(raw_record, CUSTOMER_SCHEMA)
 
    if not validation["valid"] and validation.get("fatal_errors"):
        # Fatal errors = record cannot be processed (e.g., missing required field)
        return {
            "status": "rejected",
            "reason": "fatal_validation_errors",
            "errors": validation["errors"],
            "original": raw_record
        }
 
    # Step 2: Normalize (using the partially-validated record)
    base_record = validation.get("normalized_record", raw_record)
    normalization = normalize_record(base_record, CUSTOMER_SCHEMA)
 
    return {
        "status": "valid" if validation["valid"] else "valid_with_warnings",
        "record": normalization["normalized_record"],
        "warnings": validation.get("warnings", []),
        "changes_made": normalization.get("changes", []),
        "original": raw_record
    }

def check_for_duplicates(
    new_record: Dict[str, Any],
    existing_records: List[Dict[str, Any]],
    threshold: float = 0.85
) -> Dict[str, Any]:
    """
    Check if new_record is a duplicate of any existing record.
    Uses fuzzy matching on key fields: email, company_name, phone.
    Returns the best match if above threshold.
    """
    response = requests.post(
        f"https://{DATAFORGE_HOST}/deduplicate/check",
        headers=HEADERS,
        json={
            "candidate": new_record,
            "existing_records": existing_records,
            "match_fields": ["email", "company_name", "phone"],
            "similarity_threshold": threshold,
            "strategy": "fuzzy"  # "exact" | "fuzzy" | "ml"
        }
    )
    response.raise_for_status()
    return response.json()
 
 
def deduplicate_batch(records: List[Dict[str, Any]]) -> Dict[str, Any]:
    """
    Find and group duplicates within a batch of records.
    Returns groups of duplicates and a recommended merge strategy.
    """
    response = requests.post(
        f"https://{DATAFORGE_HOST}/deduplicate/batch",
        headers=HEADERS,
        json={
            "records": records,
            "match_fields": ["email", "company_name", "phone"],
            "similarity_threshold": 0.85,
            "strategy": "fuzzy",
            "merge_strategy": "newest_wins"  # "newest_wins" | "most_complete" | "manual"
        }
    )
    response.raise_for_status()
    return response.json()

def score_record_quality(record: Dict[str, Any], schema: Dict) -> Dict[str, Any]:
    """
    Score a record's data quality on a 0-100 scale.
    Returns dimension scores: completeness, accuracy, consistency, freshness.
    """
    response = requests.post(
        f"https://{DATAFORGE_HOST}/quality/score",
        headers=HEADERS,
        json={
            "record": record,
            "schema": schema,
            "dimensions": ["completeness", "accuracy", "consistency"]
        }
    )
    response.raise_for_status()
    return response.json()
 
 
def route_by_quality(record: Dict[str, Any], score: Dict[str, Any]) -> str:
    """Determine record routing based on quality score."""
    overall = score["overall_score"]
 
    if overall >= 80:
        return "production"      # Write directly to production DB
    elif overall >= 60:
        return "review_queue"    # Flag for human review before production
    elif overall >= 40:
        return "enrichment"      # Send to enrichment queue, retry later
    else:
        return "quarantine"      # Hold for manual investigation
 
 
# Full pipeline function
def ingest_record(raw_record: Dict[str, Any]) -> Dict[str, Any]:
    """End-to-end record ingestion with quality pipeline."""
 
    # Stage 1: Validate and normalize
    processed = process_incoming_record(raw_record)
    if processed["status"] == "rejected":
        return {**processed, "route": "quarantine"}
 
    clean_record = processed["record"]
 
    # Stage 2: Score quality
    score = score_record_quality(clean_record, CUSTOMER_SCHEMA)
 
    # Stage 3: Route
    route = route_by_quality(clean_record, score)
 
    return {
        "status": processed["status"],
        "record": clean_record,
        "quality_score": score["overall_score"],
        "quality_dimensions": score["dimensions"],
        "route": route,
        "warnings": processed.get("warnings", []),
        "changes": processed.get("changes_made", [])
    }

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
 
def process_import_batch(records: List[Dict], max_workers: int = 10) -> Dict[str, Any]:
    """
    Process a batch of imported records concurrently.
    Returns routing summary and detailed results.
    """
    results = []
    routes = {"production": 0, "review_queue": 0, "enrichment": 0, "quarantine": 0}
 
    start = time.time()
 
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(ingest_record, r): i for i, r in enumerate(records)}
 
        for future in as_completed(futures):
            result = future.result()
            results.append(result)
            routes[result["route"]] = routes.get(result["route"], 0) + 1
 
    elapsed = time.time() - start
 
    return {
        "processed": len(records),
        "elapsed_seconds": round(elapsed, 2),
        "throughput": round(len(records) / elapsed, 1),
        "routing_summary": routes,
        "average_quality_score": sum(
            r.get("quality_score", 0) for r in results
        ) / len(results),
        "results": results
    }
 
 
# Example: process a CSV import
import csv
 
def process_csv_import(filepath: str) -> Dict:
    records = []
    with open(filepath, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        records = list(reader)
 
    print(f"Processing {len(records)} records from {filepath}...")
    result = process_import_batch(records)
 
    print(f"Completed in {result['elapsed_seconds']}s "
          f"({result['throughput']} records/sec)")
    print(f"Quality score: {result['average_quality_score']:.1f}/100")
    print(f"Routing: {result['routing_summary']}")
 
    return result

from collections import defaultdict
from datetime import datetime
 
 
class PipelineMonitor:
    """Track data quality metrics over time for dashboards and alerts."""
 
    def __init__(self):
        self.daily_stats = defaultdict(lambda: {
            "processed": 0, "rejected": 0,
            "quality_scores": [], "routes": defaultdict(int)
        })
 
    def record_result(self, result: Dict[str, Any]):
        date = datetime.utcnow().date().isoformat()
        stats = self.daily_stats[date]
        stats["processed"] += 1
 
        if result.get("route") == "quarantine" and result.get("status") == "rejected":
            stats["rejected"] += 1
 
        if "quality_score" in result:
            stats["quality_scores"].append(result["quality_score"])
 
        stats["routes"][result.get("route", "unknown")] += 1
 
    def get_daily_report(self, date: str = None) -> Dict:
        date = date or datetime.utcnow().date().isoformat()
        stats = self.daily_stats[date]
        scores = stats["quality_scores"]
 
        return {
            "date": date,
            "processed": stats["processed"],
            "rejection_rate": f"{stats['rejected'] / max(stats['processed'], 1) * 100:.1f}%",
            "avg_quality_score": round(sum(scores) / len(scores), 1) if scores else 0,
            "routing": dict(stats["routes"])
        }