Automated Micro-Tools &Python Assetsfor Developers & Founders

#!/usr/bin/env python3
"""
===============================================================================
SchemaGuard: Autonomous Data Contract & Schema Drift Prevention Engine
===============================================================================
SchemaGuard detects breaking schema mutations, quantifies downstream financial
and operational blast radius, and enforces CI/CD deployment gates for APIs,
event streams, and analytical data pipelines.

Features:
- Recursive structural AST diffing (JSON Schema / Event Payload contracts)
- Backward compatibility verification (Field drops, type shifts, strictness)
- Downstream consumer blast radius & economic triage cost estimation
- Automated Markdown and structured JSON audit generation
- Zero external dependencies (Pure Python 3.8+ standard library)
===============================================================================
"""

import argparse
import datetime
import enum
import json
import os
import sys
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List, Optional, Tuple, Union


class Severity(str, enum.Enum):
    CRITICAL = "CRITICAL"  # Definite breaking change: consumer failure guaranteed
    HIGH = "HIGH"          # Probable breakage: strict validation failure
    MEDIUM = "MEDIUM"      # Semantic/constraint shift: potential runtime bug
    LOW = "LOW"            # Non-breaking modification: optional field addition
    INFO = "INFO"          # Metadata or descriptive modification


@dataclass
class BlastRadiusAnalysis:
    total_mutations: int
    breaking_mutations: int
    risk_score: float  # Scale 0.0 - 100.0
    estimated_incident_hours: float
    estimated_financial_loss_usd: float
    decision: GateDecision


class SchemaGuardComparator:
    """Recursive comparator detecting contract breaking changes."""

    SEVERITY_WEIGHTS = {
        Severity.CRITICAL: 25.0,
        Severity.HIGH: 15.0,
        Severity.MEDIUM: 7.5,
        Severity.LOW: 1.0,
        Severity.INFO: 0.0,
    }

    def compare(self) -> List[SchemaMutation]:
        self.mutations = []
        self._diff_node(self.baseline, self.candidate, path="$")
        return self.mutations

# ... the full file ships with your download