⚡ The 5-Second Pitch

Web scrapers and upstream API integrations break at 2:00 AM when an engineer renames price_usd to productPrice, wraps fields inside data.listing.item, or migrates to a new GraphQL schema.

Calling an LLM on every scraped payload is unusable—it adds 800ms of latency, costs thousands of dollars per month, and leaks user PII to external endpoints.

payload-heal provides a drop-in replacement for pydantic.BaseModel called HealableModel with a 3-tier hybrid self-healing engine:

• 0ms overhead, $0 cost on healthy payloads.
• <1ms, $0 cost heuristic auto-healing for 90% of schema drifts (casing, leaf promotion, prefix/suffix changes).
• Runs LLMs ONCE per schema drift on a lightweight, PII-sanitized skeleton, commits the healed path into a local lockfile cache, and never calls the LLM again.

🧠 How the 3-Tier Hybrid Engine Works

Incoming JSON / GraphQL Payload
              │
              ▼
    [ All Required Fields Present? ]
         ├─── YES ───► [ Fast Path: 0ms Overhead, $0 Cost ] ───► Validated Model
         │
         └─── NO  (Schema Drift Detected!)
              │
              ├──► Tier 0: Lockfile Cache (.payload_heal_lock.json)
              │         Lookup previously healed dot paths. (0ms, $0)
              │         *Re-Drift Guard: Auto-invalidates if upstream drifts a second time!
              │
              ├──► Tier 1: Deterministic Heuristic Matcher
              │         Zero-LLM instant recovery in <1ms. ($0)
              │         - camelCase / snake_case / kebab-case normalization
              │         - Nested leaf promotion (e.g., user_id -> data.user.id)
              │         - Token overlap + SequenceMatcher type compatibility
              │         - Ambiguity protection (refuses to guess on tied confidence)
              │
              └──► Tier 2: AI Skeleton Healer
                        Invoked ONLY ONCE when heuristics cannot resolve drift.
                        - Generates a 97% compressed, PII-masked JSON skeleton
                        - Queries OpenAI / Groq / Ollama / Gemini for path mapping
                        - Strict Type-Safety verification prevents cache poisoning
                        - Caches result in Tier 0 lockfile and alerts team via Webhook

🚀 30-Second Quickstart

Installation

pip install payload-heal

Drop-In Usage

Simply replace pydantic.BaseModel with HealableModel:

from pydantic import Field
from payload_heal import HealableModel

class Product(HealableModel):
    title: str
    price: float = Field(description="Retail price in USD")
    rating: float
    in_stock: bool

# Upstream scraped API drifted: keys renamed and wrapped inside nested objects!
broken_payload = {
    "status": "ok",
    "data": {
        "itemDetails": {
            "itemTitle": "Sony WH-1000XM5 Wireless Headphones",
            "priceUsd": "399.99",   # string coerced to float
            "userRating": 4.8,
            "inStock": "true",       # string coerced to bool
        }
    }
}

# Automatically recovers!
product = Product.model_validate(broken_payload)

print(product.title)     # "Sony WH-1000XM5 Wireless Headphones"
print(product.price)     # 399.99 (float)
print(product.rating)    # 4.8 (float)
print(product.in_stock)  # True (bool)

# Next time this payload is encountered: Tier 0 Lockfile Cache hit in 0ms!

🔌 Supported AI Providers (Tier 2)

Configure providers globally or per-model via HealConfig:

from typing import ClassVar
from payload_heal import HealableModel, HealConfig

# 1. OpenAI (Default: gpt-4o-mini)
class OpenAIModel(HealableModel):
    heal_config: ClassVar[HealConfig] = HealConfig(
        llm_provider="openai",
        api_key="sk-..."  # or set OPENAI_API_KEY env var
    )

# 2. Groq (Ultra-fast Llama-3.3-70B)
class GroqModel(HealableModel):
    heal_config: ClassVar[HealConfig] = HealConfig(
        llm_provider="groq",
        model="llama-3.3-70b-versatile",
        api_key="gsk_..."  # or set GROQ_API_KEY env var
    )

# 3. Local Ollama (100% Free & Private on localhost)
class OllamaModel(HealableModel):
    heal_config: ClassVar[HealConfig] = HealConfig(
        llm_provider="ollama",
        model="llama3",
        api_base="http://localhost:11434/v1"
    )

# 4. Google Gemini
class GeminiModel(HealableModel):
    heal_config: ClassVar[HealConfig] = HealConfig(
        llm_provider="gemini",
        api_key="..."  # or set GEMINI_API_KEY env var
    )

🛠️ Standalone Decorator (@heal_json)

Wrap existing third-party models or function handlers without modifying class definitions:

from pydantic import BaseModel
from payload_heal import heal_json

class ThirdPartySchema(BaseModel):
    user_id: int
    email: str

@heal_json(ThirdPartySchema)
def ingest_webhook(payload: dict):
    # payload is healed before function execution
    user = ThirdPartySchema.model_validate(payload)
    print(f"Processed user #{user.user_id}: {user.email}")

🛡️ Privacy-First PII Masking

Before any skeleton is sent to an external LLM endpoint, payload-heal automatically redacts sensitive data from preview samples:

• Email Addresses: user@domain.com -> [MASKED_EMAIL]
• Bearer & JWT Tokens: Bearer eyJhbGci... -> Bearer [MASKED_TOKEN]
• API Keys: sk-proj-1234... -> [MASKED_TOKEN]
• Credit Cards: 4111-2222-3333-4444 (and Luhn-validated sequences) -> [MASKED_CREDIT_CARD]

🔔 Slack / Discord Webhook Drift Alerts

Get notified in Slack or Discord the moment an upstream API drifts:

class MonitoredModel(HealableModel):
    heal_config: ClassVar[HealConfig] = HealConfig(
        webhook_url="https://hooks.slack.com/services/T00/B00/X00"  # or Discord webhook
    )
    price: float

When schema drift is resolved, payload-heal fires a non-blocking notification:

⚠️ Payload Schema Drift Healed
Model: MonitoredModel
Healed Fields (1):
• price → data.pricing.priceUsd (healed by heuristic)
Lockfile updated automatically. No immediate action required.

💻 Developer CLI

Inspect skeletons and manage lockfiles from your terminal:

# View all healed models and field mappings
payload-heal status

# Inspect a raw JSON file and view skeleton compression ratio
payload-heal inspect scraper_dump.json

# Reset cached mappings for a specific model (or all models)
payload-heal reset ProductModel
payload-heal reset

🏃 Try the Live Demo

Run the self-contained scraper drift simulation:

python examples/demo_scraper_drift.py

Output:

=====================================================================================
  PAYLOAD-HEAL LIVE DEMO: Resilient Web Scraper Ingestion
=====================================================================================
Iteration    API State                      Tier Used              Latency    Status
-------------------------------------------------------------------------------------
Run 1        v1 (Normal Schema)             Fast Path (0ms, $0)    0.074 ms   [OK] Ingested
Run 2        v2 (camelCase + Nesting)       Tier-1 Heuristic ($0)  24.551 ms  [OK] Healed
Run 3        v3 (Complete Rewrite)          Tier-2 AI Healer       46.463 ms  [OK] Healed & Cached
Run 4        v3 Repeat Payload              Tier-0 Lockfile (0ms)  2.693 ms   [OK] 0ms Cache Hit
-------------------------------------------------------------------------------------

🧪 Test Suite

Run unit and integration tests across all modules:

pytest -v

All 54 tests pass in ~4 seconds.

📄 License

MIT License. See LICENSE for details. Built for resilient engineering teams.