selfheal-sdk

AI self-healing decorator for any Python data pipeline.

When your pipeline raises an exception, @healed classifies the failure, applies a fix automatically, and escalates to humans only when genuinely needed.

Built by Sahil Deol

---

Install

pip install selfheal-sdk

With AI backends (recommended):

pip install selfheal-sdk[typesafe]   # TypeSafe JEV — best classification
pip install selfheal-sdk[anthropic]  # Claude Haiku fallback
pip install selfheal-sdk[nvidia]     # Llama-3.1 fallback
pip install selfheal-sdk[all]        # everything

No extras? Zero-dep keyword fallback always works out of the box.

---

Quickstart

from selfheal import healed

@healed(pipeline="daily_sales_etl")
def run():
    records = extract()
    validate(records)        # raises? selfheal catches and heals it
    cleaned = transform(records)
    load(cleaned)

if __name__ == "__main__":
    run()

That's the only change needed.

---

How it works

When run() raises, @healed does this:

1. Classify — sends the error + traceback to TypeSafe JEV (or fallback LLM). Returns failure_type, action, confidence, severity.
2. Threshold check — auto-heals only when confidence ≥ 0.65 AND severity ≤ 3.
3. Act:

Failure type	Action	Auto?	Hook needed
BAD_DATA	QUARANTINE	✅	run_quarantine()
TRANSIENT	RETRY	✅	none
SCHEMA_DRIFT (new col)	ADD_COLUMN	✅	add_columns_to_dest()
PERMISSION / CODE_DEFECT	ESCALATE	❌	human approval
High severity (≥4)	ESCALATE	❌	human approval

4. Report — best-effort POST to your dashboard if configured. Never blocks the pipeline.

---

API keys

The healer tries backends in order: TypeSafe JEV → Anthropic → NVIDIA → keyword fallback.

export TYPESAFE_API_KEY=ts-...       # TypeSafe JEV (recommended)
export ANTHROPIC_API_KEY=sk-ant-...  # Claude Haiku fallback
export NVIDIA_API_KEY=nvapi-...      # Llama-3.1 fallback

Keyword fallback requires no API key and always works.

---

Decorator options

@healed(
    pipeline="my_etl",                    # name in logs
    server_url="http://localhost:8000",   # optional dashboard URL
    on_escalate=None,                     # callback(result) on ESCALATE
    max_retries=1,                        # retry attempts before giving up
    silent=False,                         # suppress INFO logs
)
def run():
    ...

---

Recovery hooks

Define these in the same module as your decorated function. selfheal calls them automatically.

BAD_DATA → run_quarantine()

@healed(pipeline="daily_sales_etl")
def run():
    records = extract()
    validate(records)
    load(transform(records))

def run_quarantine():
    records = extract()
    clean = [r for r in records if r.get("customer_id") and r.get("amount", 0) > 0]
    bad   = [r for r in records if r not in clean]
    log_bad_rows(bad)
    load(transform(clean))

SCHEMA_DRIFT → add_columns_to_dest(cols, schema)

def add_columns_to_dest(extra_cols: list[str], actual_schema: dict) -> None:
    with connect() as conn:
        for col in extra_cols:
            conn.execute(f"ALTER TABLE dest ADD COLUMN IF NOT EXISTS {col} TEXT")
        conn.commit()

selfheal calls this before retrying, so the retry succeeds with the new column in place.

ESCALATE → on_escalate callback

def notify_slack(result: dict):
    print(f"[ALERT] {result['root_cause']}")

@healed(pipeline="orders_etl", on_escalate=notify_slack, max_retries=3)
def run():
    ...

---

Environment variables

Variable	Default	Description
TYPESAFE_API_KEY	—	TypeSafe JEV API key
ANTHROPIC_API_KEY	—	Anthropic Claude fallback
NVIDIA_API_KEY	—	NVIDIA NIM fallback
SH_MIN_CONFIDENCE	0.65	Minimum confidence to auto-heal
SH_MAX_SEVERITY	3	Max severity (1–5) to auto-heal

---

Version history

Version	Changes
0.3.1	PyPI release; README updated
0.3.0	TypeSafe JEV primary backend; ADD_COLUMN auto-heal; max_retries; runbook ingestion; confidence + severity thresholds
0.2.0	Anthropic + NVIDIA LLM backends
0.1.0	Initial release — keyword fallback only

---

selfheal-sdk — by Sahil Deol