spanforge.sdk.pii — PII Service Client

Module: spanforge.sdk.pii
Added in: 2.0.3 (Phase 3: PII Service Hardening)

spanforge.sdk.pii provides the Phase 3 PII service client with full-text scanning, payload anonymisation, pipeline action enforcement, GDPR/HIPAA/CCPA/ DPDP/PIPL compliance helpers, and training-data auditing.

The pre-built sf_pii singleton is available at the top level:

from spanforge.sdk import sf_pii

Quick example

from spanforge.sdk import sf_pii

# Scan raw text
result = sf_pii.scan_text("Contact alice@example.com or call +1 555-867-5309")
print(result.detected)           # True
for entity in result.entities:
    print(entity.type, entity.start, entity.end, entity.score)

# Anonymise a payload dict
anon = sf_pii.anonymise({
    "user": "alice@example.com",
    "note": "SSN 078-05-1120",
})
print(anon.clean_payload)        # {"user": "<EMAIL_ADDRESS>", "note": "<US_SSN>"}

SFPIIClient

class SFPIIClient(SFServiceClient)

All methods are thread-safe. The class can be used standalone or via the sf_pii singleton exported from spanforge.sdk.

scan_text()

def scan_text(
    self,
    text: str,
    *,
    language: str = "en",
) -> PIITextScanResult

Scan raw text for PII entities. Uses Presidio when installed; falls back to the regex-based redact.scan_payload() engine automatically.

Returns: PIITextScanResult

Example:

result = sf_pii.scan_text("SSN: 078-05-1120", language="en")
assert result.detected
assert result.entities[0].type == "US_SSN"

anonymise()

def anonymise(
    self,
    payload: dict,
    *,
    language: str = "en",
    max_depth: int = 10,
) -> PIIAnonymisedResult

Recursively walk payload, scan every string value, and replace PII hits with <TYPE> placeholders. Returns the cleaned payload alongside a full redaction_manifest recording each replacement.

Returns: PIIAnonymisedResult

Example:

anon = sf_pii.anonymise({"email": "alice@example.com", "meta": {"ip": "203.0.113.4"}})
# anon.clean_payload == {"email": "<EMAIL_ADDRESS>", "meta": {"ip": "<IP_ADDRESS>"}}
# anon.redaction_manifest[0].field_path == "email"
# anon.redaction_manifest[0].original_hash  (SHA-256 of original — never raw PII)

scan_async()

Added in: 2.0.14

async def scan_async(
    self,
    text: str,
    *,
    language: str = "en",
    score_threshold: float = 0.5,
) -> PIITextScanResult

Non-blocking async variant of scan_text(). Delegates to the synchronous method via asyncio.get_event_loop().run_in_executor() so it never blocks the event loop.

Returns: PIITextScanResult

Example:

import asyncio
from spanforge.sdk import sf_pii

result = asyncio.run(sf_pii.scan_async("alice@example.com"))
assert result.detected
assert result.entities[0].type == "EMAIL_ADDRESS"

scan_batch()

async def scan_batch(
    self,
    texts: list[str],
    *,
    language: str = "en",
) -> list[PIITextScanResult]

Scan multiple texts in parallel via asyncio.gather. Falls back to sequential execution when no running event loop is present.

Example:

import asyncio

results = asyncio.run(sf_pii.scan_batch(["alice@example.com", "hello world"]))
assert results[0].detected
assert not results[1].detected

apply_pipeline_action()

def apply_pipeline_action(
    self,
    scan_result: PIITextScanResult,
    action: str = "flag",
    *,
    threshold: float = 0.85,
) -> PIIPipelineResult

Enforce a pipeline action based on a previous scan_text() result.

The threshold parameter filters entities: only those with score >= threshold are considered when deciding whether to fire the action (default 0.85). Sub-threshold entities are still included in scan_result for audit purposes.

Returns: PIIPipelineResult

Raises: SFPIIBlockedError (action "block" only)

Example:

from spanforge.sdk._exceptions import SFPIIBlockedError

scan = sf_pii.scan_text("My SSN is 078-05-1120")
try:
    pipeline = sf_pii.apply_pipeline_action(scan, action="block", threshold=0.85)
except SFPIIBlockedError as exc:
    print("Blocked:", exc.entity_types)   # ["US_SSN"]

get_status()

def get_status(self) -> PIIServiceStatus

Return the current sf-pii service status.

Returns: PIIServiceStatus

Example:

status = sf_pii.get_status()
print(status.presidio_available)      # True / False
print(status.entity_types_loaded)     # ["EMAIL_ADDRESS", "PHONE_NUMBER", ...]
print(status.last_scan_at)            # "2026-04-17T12:00:00Z" or None

erase_subject()

def erase_subject(
    self,
    subject_id: str,
    project_id: str,
) -> ErasureReceipt

GDPR Article 17 — Right to Erasure. Locate all audit events associated with subject_id in project_id and issue erasure instructions to the downstream store.

The subject_id is SHA-256 hashed in the returned ErasureReceipt — the raw identifier is never persisted.

Returns: ErasureReceipt

Example:

receipt = sf_pii.erase_subject("user-12345", "proj-abc")
print(receipt.records_erased)         # 42
print(receipt.erased_at)              # "2026-04-17T12:00:00Z"
print(receipt.subject_id_hash)        # SHA-256 of "user-12345"

export_subject_data()

def export_subject_data(
    self,
    subject_id: str,
    project_id: str,
) -> DSARExport

CCPA / GDPR — Data Subject Access Request (DSAR). Aggregate all events referencing subject_id from the audit store for project_id.

Returns: DSARExport

Example:

export = sf_pii.export_subject_data("user-12345", "proj-abc")
print(export.total_records)
for record in export.records:
    print(record["event_type"], record["created_at"])

safe_harbor_deidentify()

def safe_harbor_deidentify(self, text: str) -> SafeHarborResult

HIPAA Safe Harbor De-identification per 45 CFR §164.514(b)(2). Removes or generalises all 18 PHI identifier categories from text:

Returns: SafeHarborResult

Example:

result = sf_pii.safe_harbor_deidentify(
    "Patient John Doe (DOB: 04/17/1932, MRN 0000-4321) lives at 902 Oak Lane, 90210."
)
print(result.deidentified_text)
# "Patient [NAME] (DOB: 1932, MRN [REMOVED]) lives at [ADDRESS], 902XX."
print(result.identifiers_removed)    # 5

audit_training_data()

def audit_training_data(
    self,
    dataset_path: str,
    *,
    max_records: int = 10_000,
    language: str = "en",
) -> TrainingDataPIIReport

EU AI Act Article 10 — Training Data Governance. Scan each line of a JSONL dataset file for PII and produce a prevalence report. Lines that are not valid JSON are counted as malformed_lines and skipped.

Returns: TrainingDataPIIReport

Example:

report = sf_pii.audit_training_data("dataset/train.jsonl", max_records=5000)
print(f"{report.pii_record_count} / {report.total_records} records contain PII")
for entry in report.entity_breakdown:
    print(f"  {entry.entity_type}: {entry.count}")

get_pii_stats()

def get_pii_stats(self, project_id: str) -> list[PIIHeatMapEntry]

Aggregate PII detection statistics per entity type for project_id. Powers the SpanForge dashboard PII heat-map.

Returns: list[PIIHeatMapEntry]

Example:

for entry in sf_pii.get_pii_stats("proj-abc"):
    print(entry.entity_type, entry.count, entry.last_seen_at)

Return types

PIIEntityResult

@dataclass(frozen=True)
class PIIEntityResult:
    type: str
    start: int
    end: int
    score: float

A single detected PII entity.

PIITextScanResult

@dataclass
class PIITextScanResult:
    detected: bool
    entities: list[PIIEntityResult]
    redacted_text: str

PIIRedactionManifestEntry

@dataclass(frozen=True)
class PIIRedactionManifestEntry:
    field_path: str
    entity_type: str
    original_hash: str
    replacement: str

One entry in an anonymise() redaction manifest.

PIIAnonymisedResult

@dataclass
class PIIAnonymisedResult:
    clean_payload: dict
    redaction_manifest: list[PIIRedactionManifestEntry]

PIIPipelineResult

@dataclass
class PIIPipelineResult:
    action: str
    detected: bool
    entity_types: list[str]
    redacted_text: str | None

PIIServiceStatus

@dataclass
class PIIServiceStatus:
    status: str
    presidio_available: bool
    entity_types_loaded: list[str]
    last_scan_at: str | None

ErasureReceipt

@dataclass
class ErasureReceipt:
    subject_id_hash: str
    project_id: str
    records_erased: int
    erased_at: str

DSARExport

@dataclass
class DSARExport:
    subject_id_hash: str
    project_id: str
    total_records: int
    records: list[dict]
    exported_at: str

SafeHarborResult

@dataclass
class SafeHarborResult:
    deidentified_text: str
    identifiers_removed: int

TrainingDataPIIReport

@dataclass
class TrainingDataPIIReport:
    dataset_path: str
    total_records: int
    pii_record_count: int
    malformed_lines: int
    entity_breakdown: list[PIIHeatMapEntry]

PIIHeatMapEntry

@dataclass
class PIIHeatMapEntry:
    entity_type: str
    count: int
    last_seen_at: str | None

Exceptions

See exceptions reference for full details.

PIPL entity types

China PIPL-specific entity types registered in presidio_backend.py:

These types are flagged as pipl_sensitive for cross-border transfer controls.

Environment variables

See also

• Configuration reference — PII service settings
• User Guide — PII Redaction
• User Guide — Compliance
• Runbook — PII Scanning
• spanforge.redact — low-level PII detection and field-level redaction
• spanforge.sdk.secrets — secrets scanning
• Exceptions reference