How to Build a Multi-Tenant SaaS Data Validation Architecture with DataForge API

# schemas/tenant_schema_loader.py
import json
import redis
from typing import Optional
 
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
 
def get_tenant_schema(tenant_id: str) -> Optional[dict]:
    """Load tenant-specific validation schema from Redis."""
    raw = redis_client.get(f"schema:{tenant_id}")
    if not raw:
        return None
    return json.loads(raw)
 
def set_tenant_schema(tenant_id: str, schema: dict) -> None:
    """Register or update a tenant's validation schema."""
    redis_client.set(f"schema:{tenant_id}", json.dumps(schema))
 
# Example schema for Tenant A (FinTech company — strict)
tenant_a_schema = {
    "fields": {
        "customer_id": {"type": "uuid", "required": True},
        "amount": {"type": "decimal", "min": 0.01, "max": 1_000_000},
        "currency": {"type": "enum", "values": ["USD", "EUR", "GBP"]},
        "email": {"type": "email", "required": True},
        "phone": {"type": "e164_phone", "required": True}
    },
    "strict_mode": True,
    "reject_extra_fields": True
}
 
# Example schema for Tenant B (E-commerce — flexible)
tenant_b_schema = {
    "fields": {
        "customer_id": {"type": "integer", "required": True},
        "amount": {"type": "decimal", "min": 0.01},
        "currency": {"type": "string", "max_length": 3},
        "email": {"type": "email", "required": False},
        "phone": {"type": "string", "required": False}
    },
    "strict_mode": False,
    "reject_extra_fields": False
}
 
set_tenant_schema("tenant_a", tenant_a_schema)
set_tenant_schema("tenant_b", tenant_b_schema)

# validation/tenant_validator.py
import httpx
import asyncio
from typing import Any
from schemas.tenant_schema_loader import get_tenant_schema
 
DATAFORGE_API_URL = "https://apivult.com/api/dataforge/v1/validate"
DATAFORGE_API_KEY = "YOUR_API_KEY"
 
class TenantValidationError(Exception):
    """Raised when validation fails for a tenant's data batch."""
    def __init__(self, tenant_id: str, errors: list):
        self.tenant_id = tenant_id
        self.errors = errors
        super().__init__(f"Validation failed for tenant {tenant_id}: {len(errors)} errors")
 
async def validate_tenant_batch(
    tenant_id: str,
    records: list[dict[str, Any]],
    raise_on_error: bool = False
) -> dict:
    """
    Validate a batch of records against the tenant's registered schema.
    
    Returns a quality report scoped to this tenant's rules.
    """
    schema = get_tenant_schema(tenant_id)
    if not schema:
        raise ValueError(f"No schema registered for tenant: {tenant_id}")
 
    payload = {
        "tenant_id": tenant_id,           # Scopes the result to this tenant
        "schema": schema,
        "records": records,
        "options": {
            "include_cleaned_records": True,
            "include_field_stats": True,
            "error_limit": 100             # Cap error messages per batch
        }
    }
 
    headers = {
        "X-RapidAPI-Key": DATAFORGE_API_KEY,
        "Content-Type": "application/json"
    }
 
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(DATAFORGE_API_URL, json=payload, headers=headers)
        response.raise_for_status()
        result = response.json()
 
    if raise_on_error and result.get("error_count", 0) > 0:
        raise TenantValidationError(tenant_id, result["errors"])
 
    return {
        "tenant_id": tenant_id,
        "total_records": result["total_records"],
        "valid_records": result["valid_records"],
        "error_count": result["error_count"],
        "quality_score": result["quality_score"],
        "cleaned_records": result.get("cleaned_records", []),
        "field_stats": result.get("field_stats", {}),
        "errors": result.get("errors", [])
    }

# pipeline/multi_tenant_pipeline.py
import asyncio
from dataclasses import dataclass
from validation.tenant_validator import validate_tenant_batch, TenantValidationError
 
@dataclass
class TenantBatch:
    tenant_id: str
    records: list[dict]
 
async def process_tenant_batches(batches: list[TenantBatch]) -> dict[str, dict]:
    """
    Process multiple tenant batches in parallel.
    
    Critical: results are keyed by tenant_id, never mixed.
    """
    tasks = {
        batch.tenant_id: validate_tenant_batch(
            tenant_id=batch.tenant_id,
            records=batch.records
        )
        for batch in batches
    }
 
    results = {}
    for tenant_id, task in tasks.items():
        try:
            results[tenant_id] = await task
        except TenantValidationError as e:
            results[tenant_id] = {
                "tenant_id": tenant_id,
                "status": "error",
                "error_count": len(e.errors),
                "errors": e.errors
            }
        except Exception as e:
            # Isolate failures: one tenant's error must not affect others
            results[tenant_id] = {
                "tenant_id": tenant_id,
                "status": "failed",
                "error": str(e)
            }
 
    return results
 
# Example usage
async def main():
    batches = [
        TenantBatch(
            tenant_id="tenant_a",
            records=[
                {"customer_id": "550e8400-e29b-41d4-a716-446655440000",
                 "amount": "1500.00", "currency": "USD",
                 "email": "[email protected]", "phone": "+14155552671"},
                {"customer_id": "invalid-id", "amount": "-50",
                 "currency": "XYZ", "email": "bad-email"}  # Multiple errors
            ]
        ),
        TenantBatch(
            tenant_id="tenant_b",
            records=[
                {"customer_id": 12345, "amount": "299.99", "currency": "USD"},
                {"customer_id": 67890, "amount": "49.99"}
            ]
        )
    ]
 
    results = await process_tenant_batches(batches)
 
    for tenant_id, report in results.items():
        print(f"\n=== {tenant_id} ===")
        print(f"Quality Score: {report.get('quality_score', 'N/A')}")
        print(f"Valid: {report.get('valid_records', 0)}/{report.get('total_records', 0)}")
        if report.get("errors"):
            print("Errors:")
            for err in report["errors"][:3]:
                print(f"  Row {err['row']}: {err['field']} — {err['message']}")
 
asyncio.run(main())

# reporting/quality_store.py
import json
import redis
from datetime import datetime
 
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
 
def store_quality_report(tenant_id: str, report: dict) -> str:
    """Store a tenant's validation report with timestamp."""
    timestamp = datetime.utcnow().isoformat()
    key = f"quality:{tenant_id}:{timestamp}"
    
    # Store for 90 days (GDPR-compliant retention)
    redis_client.setex(key, 7_776_000, json.dumps({
        **report,
        "timestamp": timestamp
    }))
    
    # Update rolling quality score in a sorted set
    redis_client.zadd(
        f"quality_scores:{tenant_id}",
        {timestamp: report.get("quality_score", 0)}
    )
    # Keep last 90 data points
    redis_client.zremrangebyrank(f"quality_scores:{tenant_id}", 0, -91)
    
    return key
 
def get_tenant_quality_trend(tenant_id: str, last_n: int = 30) -> list[float]:
    """Get quality score trend for a tenant over recent batches."""
    scores = redis_client.zrange(
        f"quality_scores:{tenant_id}",
        -last_n, -1,
        withscores=True
    )
    return [score for _, score in scores]

# main.py
from fastapi import FastAPI, Depends, HTTPException, Header
from pydantic import BaseModel
from typing import Optional
import asyncio
 
from validation.tenant_validator import validate_tenant_batch
from reporting.quality_store import store_quality_report
 
app = FastAPI(title="Multi-Tenant Data Validation Service")
 
class ValidationRequest(BaseModel):
    records: list[dict]
    raise_on_error: bool = False
 
def extract_tenant_id(x_tenant_id: str = Header(...)) -> str:
    """Extract tenant ID from request header (replace with JWT in production)."""
    if not x_tenant_id or len(x_tenant_id) > 64:
        raise HTTPException(status_code=400, detail="Invalid tenant ID")
    return x_tenant_id
 
@app.post("/validate")
async def validate_data(
    request: ValidationRequest,
    tenant_id: str = Depends(extract_tenant_id)
):
    """Validate a batch of records against the tenant's registered schema."""
    report = await validate_tenant_batch(
        tenant_id=tenant_id,
        records=request.records,
        raise_on_error=request.raise_on_error
    )
    
    # Persist for audit trail
    report_key = store_quality_report(tenant_id, report)
    
    return {
        **report,
        "report_key": report_key
    }
 
@app.get("/quality/trend")
async def quality_trend(tenant_id: str = Depends(extract_tenant_id)):
    """Return quality score trend for this tenant."""
    from reporting.quality_store import get_tenant_quality_trend
    return {
        "tenant_id": tenant_id,
        "trend": get_tenant_quality_trend(tenant_id)
    }