PII Firewall Developer Guide

# Recommended
pip install "pii-firewall[presidio,langdetect]"

# Download spaCy models for the languages you use
python -m spacy download en_core_web_sm
python -m spacy download es_core_news_sm
python -m spacy download fr_core_news_sm

from privacy_firewall import create_firewall

firewall = create_firewall("healthcare")

result = firewall.process(
    text="Ana García, 43 años, hipertensión. Prescripción: enalapril 10mg.",
    context={
        "tenant_id": "hospital-001",
        "case_id":   "patient-123",
        "thread_id": "consultation-1",
        "actor_id":  "doctor-456",
    },
)

print(result.sanitized_text)
# → "[PERSON_001], 40-49, hipertensión. enalapril 10mg."
#   Name pseudonymized. Age generalized. Medical terms KEPT.

print(result.final_text)
# → LLM response with real names restored.

firewall = create_firewall("healthcare")        # or "finance", "legal", "generic"
firewall = create_firewall("healthcare", detector_backend="presidio")
firewall = create_firewall("healthcare", detector_backend="presidio", language="es")

from privacy_firewall import (
    create_custom_profile,
    EntityDisposition,
    DispositionAction,
    create_firewall,
)

profile = create_custom_profile("my_domain")

profile.add_disposition(EntityDisposition(
    entity_type="EMPLOYEE_ID",
    action=DispositionAction.PSEUDONYMIZE,
    confidence_threshold=0.9,
))
profile.add_disposition(EntityDisposition(
    entity_type="CASE_NUMBER",
    action=DispositionAction.KEEP,
    confidence_threshold=0.8,
))

firewall = create_firewall("generic", profile=profile)

firewall = create_firewall("healthcare", detector_backend="presidio")   # recommended
firewall = create_firewall("healthcare", detector_backend="regex")      # zero deps
firewall = create_firewall("healthcare", detector_backend="gliner")      # zero-shot NER
firewall = create_firewall("healthcare", detector_backend="transformers", transformer_model_id="d4data/biomedical-ner-all")
firewall = create_firewall("healthcare", detector_backend="opf")          # token classifier
firewall = create_firewall("healthcare", detector_backend="hybrid")        # max coverage

from openai import OpenAI
from privacy_firewall import create_firewall

client   = OpenAI()
firewall = create_firewall("healthcare", detector_backend="presidio")
ctx      = {"tenant_id": "t1", "case_id": "c1", "thread_id": "th1", "actor_id": "u1"}

def llm(prompt: str) -> str:
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    ).choices[0].message.content

result = firewall.secure_call(text=user_input, context=ctx, llm_client=llm)
print(result.final_text)   # real names restored

import anthropic
from privacy_firewall import create_firewall

ac       = anthropic.Anthropic()
firewall = create_firewall("healthcare", detector_backend="presidio")

def llm(prompt: str) -> str:
    return ac.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    ).content[0].text

result = firewall.secure_call(text=user_input, context=ctx, llm_client=llm)

from langchain_openai import ChatOpenAI
from privacy_firewall import create_firewall

llm_chain = ChatOpenAI(model="gpt-4o")
firewall   = create_firewall("generic", detector_backend="presidio")

result = firewall.secure_call(
    text=user_input,
    context=ctx,
    llm_client=lambda prompt: llm_chain.invoke(prompt).content,
)

# secure_call_stream yields tokens with real names already restored
for token in firewall.secure_call_stream(
    text=user_input,
    context=ctx,
    llm_client=your_streaming_llm,
):
    yield token   # send to SSE / WebSocket immediately

# main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from privacy_firewall import PrivacyFirewallSDK

SDK = None

@asynccontextmanager
async def lifespan(app):
    global SDK
    SDK = PrivacyFirewallSDK.create(
        domain=os.getenv("PII_DOMAIN", "healthcare"),
        language=os.getenv("PII_LANGUAGE") or None,
        detector_backend=os.getenv("PII_BACKEND", "presidio"),
    )
    yield

app = FastAPI(lifespan=lifespan)

class Req(BaseModel):
    text: str
    context: dict

@app.post("/sanitize")
async def sanitize(req: Req):
    if not SDK: raise HTTPException(503)
    return {"sanitized_text": SDK.anonymize_text(req.text, req.context).sanitized_text}

@app.post("/rehydrate")
async def rehydrate(req: Req):
    if not SDK: raise HTTPException(503)
    return {"final_text": SDK.rehydrate_text(req.text, req.context)}

@app.get("/health")
async def health(): return {"ok": SDK is not None}

const BASE = process.env.PII_URL ?? "http://localhost:8000";

export async function sanitize(text: string, ctx: Record<string, string>) {
  const r = await fetch(`${BASE}/sanitize`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ text, context: ctx }),
  });
  return (await r.json()).sanitized_text as string;
}

export async function rehydrate(text: string, ctx: Record<string, string>) {
  const r = await fetch(`${BASE}/rehydrate`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ text, context: ctx }),
  });
  return (await r.json()).final_text as string;
}

# In-memory (default) — wiped on restart
firewall = create_firewall("healthcare")

# SQLite — survives restarts
from privacy_firewall import SQLiteMappingVault
vault    = SQLiteMappingVault("privacy_vault.db")
firewall = create_firewall("healthcare", vault=vault)

# Step 1 — Anonymize before sending to LLM
anon       = firewall.anonymize(text=raw_text, context=ctx)
clean_text = anon.sanitized_text

# Step 2 — Call LLM with sanitized text
llm_out = my_llm(clean_text)

# Step 3 — Rehydrate on the way back
final = firewall.rehydrate(text=llm_out, context=ctx)
print(final)   # original names restored

deleted = firewall.forget(
    tenant_id="hospital-001",
    case_id="patient-123",
    thread_id="consultation-1",
)
print(f"Deleted {deleted} mappings")

# One-liner
firewall.add_custom_regex(
    entity_type="EMPLOYEE_ID",
    regex=r"\bEMP-\d{6}\b",
    locales=["GLOBAL"],          # or ["US"], ["ES"]...
    confidence=0.95,
    context_words=["employee", "staff"],
    disposition_action="pseudonymize",   # keep / pseudonymize / generalize / mask / redact
)

# Full EntityPattern for more control
import re
from privacy_firewall.patterns.catalog import EntityPattern

firewall.add_custom_pattern(EntityPattern(
    entity_type="CASE_NUMBER",
    locale="ES",
    pattern=re.compile(r"\bEXP-\d{4}/\d{6}\b"),
    confidence=0.98,
    context_words=("expediente", "exp"),
    description="Spanish legal case number",
))

from privacy_firewall import create_firewall
from privacy_firewall.presidio_integration import create_custom_recognizer

recognizer = create_custom_recognizer(
    entity_type="EMPLOYEE_ID",
    patterns=[r"\bEMP\d{6}\b"],
    context_words=["employee", "badge"],
    score=0.9,
)

firewall = create_firewall(
    "generic",
    detector_backend="presidio",
    custom_recognizers=[recognizer],
)

# ── Full ML-based recognizer ──────────────────────────────────
from presidio_analyzer import EntityRecognizer, RecognizerResult

class MyRecognizer(EntityRecognizer):
    def load(self): ...
    def analyze(self, text, entities, nlp_artifacts):
        return [
            RecognizerResult("CUSTOM", s.start, s.end, s.score)
            for s in my_model.predict(text)
        ]

firewall = create_firewall(
    "generic",
    detector_backend="presidio",
    custom_recognizers=[MyRecognizer(supported_entities=["CUSTOM"])],
)

pip install "pii-firewall[transformers]"

from privacy_firewall import create_firewall

# Any HuggingFace NER model ID
firewall = create_firewall(
    "healthcare",
    detector_backend="transformers",
    transformer_model_id="dslim/bert-base-NER",
)

# GPU (0 = first card, -1 = CPU default)
firewall = create_firewall(
    "healthcare",
    detector_backend="transformers",
    transformer_model_id="d4data/biomedical-ner-all",
    transformer_device=0,
)

# Use the built-in curated catalog
from privacy_firewall.transformers_ner.models import get_model_for_domain
config   = get_model_for_domain("medical", "en")
firewall = create_firewall("healthcare", detector_backend="transformers", transformer_model_id=config.model_id)

from presidio_analyzer import EntityRecognizer, RecognizerResult
from transformers import pipeline

class HFRecognizer(EntityRecognizer):
    def __init__(self, model_id: str):
        super().__init__(supported_entities=["PERSON", "ORG", "LOC"])
        self._pipe = pipeline("ner", model=model_id, aggregation_strategy="simple")
    def load(self): ...
    def analyze(self, text, entities, nlp_artifacts):
        return [
            RecognizerResult(s["entity_group"], s["start"], s["end"], s["score"])
            for s in self._pipe(text)
        ]

firewall = create_firewall(
    "healthcare",
    detector_backend="presidio",
    custom_recognizers=[HFRecognizer("dslim/bert-base-NER")],
)

# Force a single language (skips auto-detect)
firewall = create_firewall("healthcare", language="es")

# Pre-warm spaCy models at startup — avoids lazy-load latency
firewall.preload_languages(["es", "en", "fr"])