Python SDK

The official PrivaiShield SDK for Python. Features both synchronous and async clients, type hints, and pandas integration.

View on PyPI GitHub

Installation

Install the SDK using pip or your preferred package manager.

pipbash

pip install privaishield

poetrybash

poetry add privaishield

With async supportbash

pip install privaishield[async]

Requirements

Python 3.8+
httpx (included)
pydantic (included)

Quick Start

Initialize the client with your API key and make your first request.

Basic Usagepython

import os
from privaishield import PrivaiShield

# Initialize the client
client = PrivaiShield(
    api_key=os.environ["PRIVAISHIELD_API_KEY"]
)

# Redact PII from text
result = client.redact(
    text="Contact John Smith at john@acme.com or 555-123-4567",
    mode="enhanced"
)

print(result.redacted)
# "Contact [PERSON_1] at [EMAIL_1] or [PHONE_1]"

print(result.entities)
# [Entity(type='PERSON', token='[PERSON_1]', confidence=0.97), ...]

Environment Variablebash

export PRIVAISHIELD_API_KEY=sk_live_your_key_here

Sync & Async

The SDK provides both synchronous and asynchronous clients. Choose based on your application architecture.

Synchronous Clientpython

from privaishield import PrivaiShield

client = PrivaiShield(api_key=api_key)

# Blocking call
result = client.redact(text="Email: john@example.com")
print(result.redacted)

# Context manager for connection pooling
with PrivaiShield(api_key=api_key) as client:
    result = client.redact(text="Phone: 555-123-4567")
    print(result.redacted)

Asynchronous Clientpython

import asyncio
from privaishield import AsyncPrivaiShield

async def main():
    client = AsyncPrivaiShield(api_key=api_key)

    # Async call
    result = await client.redact(text="Email: john@example.com")
    print(result.redacted)

    # Context manager
    async with AsyncPrivaiShield(api_key=api_key) as client:
        result = await client.redact(text="SSN: 123-45-6789")
        print(result.redacted)

asyncio.run(main())

Concurrent Requestspython

import asyncio
from privaishield import AsyncPrivaiShield

async def process_texts(texts: list[str]):
    async with AsyncPrivaiShield(api_key=api_key) as client:
        # Process all texts concurrently
        tasks = [client.redact(text=t) for t in texts]
        results = await asyncio.gather(*tasks)
        return results

texts = [
    "Email: john@example.com",
    "Phone: 555-123-4567",
    "SSN: 123-45-6789",
]

results = asyncio.run(process_texts(texts))
for r in results:
    print(r.redacted)

Batch Processing

Process multiple texts efficiently in a single API call.

Basic Batchpython

from privaishield import PrivaiShield

client = PrivaiShield(api_key=api_key)

items = [
    {"id": "1", "text": "Email: john@example.com"},
    {"id": "2", "text": "Phone: 555-123-4567"},
    {"id": "3", "text": "SSN: 123-45-6789"},
]

result = client.redact_batch(items=items, mode="enhanced")

for item in result.results:
    print(f"{item.id}: {item.redacted}")
# 1: Email: [EMAIL_1]
# 2: Phone: [PHONE_1]
# 3: SSN: [SSN_1]

With Pandas DataFramepython

import pandas as pd
from privaishield import PrivaiShield

client = PrivaiShield(api_key=api_key)

# Create DataFrame with text to redact
df = pd.DataFrame({
    "id": [1, 2, 3],
    "content": [
        "Contact john@example.com",
        "Call 555-123-4567",
        "SSN: 123-45-6789"
    ]
})

# Redact entire column
df["redacted"] = client.redact_series(
    df["content"],
    mode="enhanced"
)

print(df)
#    id                   content              redacted
# 0   1  Contact john@example.com    Contact [EMAIL_1]
# 1   2         Call 555-123-4567        Call [PHONE_1]
# 2   3        SSN: 123-45-6789         SSN: [SSN_1]

Streaming for Large Batchespython

from privaishield import PrivaiShield

client = PrivaiShield(api_key=api_key)

# Stream results for progress tracking
for event in client.redact_batch_stream(items=items):
    if event.type == "item":
        print(f"Processed: {event.data.id}")
    elif event.type == "progress":
        print(f"Progress: {event.data.processed}/{event.data.total}")
    elif event.type == "complete":
        print(f"Done! Total entities: {event.data.total_entities}")

API Reference

Complete reference for all SDK methods and classes.

PrivaiShield(api_key, **options)

Create a new PrivaiShield client instance.

api_keystrYour API key (required)

base_urlstrCustom API base URL

timeoutfloatRequest timeout in seconds (default: 30.0)

max_retriesintMax retry attempts (default: 3)

client.redact(**options)

Redact PII from a text string.

def redact(
    self,
    text: str,
    mode: Literal["standard", "enhanced"] = "enhanced",
    entities: Optional[List[str]] = None,
    confidence: Optional[float] = None,
    return_original: bool = False,
) -> RedactResult:
    ...

@dataclass
class RedactResult:
    redacted: str
    entities: List[Entity]
    mode: str
    processing_time: str

@dataclass
class Entity:
    type: str
    token: str
    start: int
    end: int
    confidence: float
    original: Optional[str] = None

client.redact_document(**options)

Process and redact a document file.

def redact_document(
    self,
    file: Union[str, Path, BinaryIO],
    mode: Literal["standard", "enhanced"] = "enhanced",
    output_format: Literal["pdf", "text", "json"] = "pdf",
) -> DocumentResult:
    ...

@dataclass
class DocumentResult:
    document_id: str
    pages: int
    total_entities: int
    download_url: str
    expires_at: str

client.redact_batch(**options)

Process multiple texts in a single request.

def redact_batch(
    self,
    items: List[Dict[str, str]],  # [{"id": "1", "text": "..."}]
    mode: Literal["standard", "enhanced"] = "enhanced",
) -> BatchResult:
    ...

@dataclass
class BatchResult:
    results: List[BatchItem]
    total_items: int
    total_entities: int
    processing_time: str

Error Handling

from privaishield import PrivaiShield
from privaishield.exceptions import (
    PrivaiShieldError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
)

try:
    result = client.redact(text="...")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except PrivaiShieldError as e:
    print(f"API error: {e.code} - {e.message}")

REST API

Full API reference

Entity Types

Custom patterns & config