quba-sensitive-data-protection 0.0.8

A client library for accessing Sensitive Data Protection API

quba-sensitive-data-protection

A client library for accessing Sensitive Data Protection API — a Python client for detecting and de-identifying sensitive data (PII) in text: scan for entities (emails, names, phone numbers, …) and anonymize them with configurable rules (redact, mask, replace, hash, encrypt).

Installation

pip install quba-sensitive-data-protection
# or
uv add quba-sensitive-data-protection

Quickstart

from quba_sdp import Client, models
from quba_sdp.api import scan_text, anonymize_text

client = Client()  # defaults to the production API; pass base_url=... to override

# --- scan: locate sensitive entities ---
resp = scan_text.sync(client=client, body=models.ScanRequestBody(text="Email me at [e-mail]"))
if isinstance(resp, models.ScanResponseBody):
    for r in resp.results:
        print(r.entity_type, r.start, r.end, r.score)

# --- anonymize: transform matches ---
body = models.AnonymizeRequestBody(
    text="Email me at [e-mail]",
    rules=[
        models.MaskRule(entities=[models.ModelEntity(value="email")], masking_char="*"),
    ],
)
result = anonymize_text.sync(client=client, body=body)
if isinstance(result, models.AnonymizeResponseBody):
    print(result.text)  # 'Email me at ***********'

Base URL

Client() targets the production API by default. Override it for local or staging:

client = Client(base_url="http://localhost:8000")

Authentication

The production API requires a token — use AuthenticatedClient (it shares the same default base URL):

from quba_sdp import AuthenticatedClient

client = AuthenticatedClient(token="SuperSecretToken")

Calling endpoints

Every operation lives in a module under quba_sdp.api and exposes four functions:

function	blocking	returns
sync	✅	parsed body, or None for an undocumented/error status
sync_detailed	✅	Response[...] with .status_code, .headers, .parsed
asyncio	—	like sync, awaitable
asyncio_detailed	—	like sync_detailed, awaitable

sync returns a union of the success body and any documented error (e.g. HTTPValidationError), so narrow it before use:

r = scan_text.sync(client=client, body=...)
if isinstance(r, models.ScanResponseBody):
    ...  # success
else:
    ...  # HTTPValidationError or None

Async variant:

async with Client() as client:
    resp = await scan_text.asyncio(client=client, body=models.ScanRequestBody(text="..."))

Advanced configuration

Client / AuthenticatedClient accept timeout=, verify_ssl=, headers=, cookies=, and httpx_args= for full control of the underlying httpx client (see the Client docstring). Disabling TLS verification is a security risk:

client = Client(verify_ssl=False)  # not recommended