This section describes how to use the Python SDK and its APIs. It covers client configuration and core anonymization workflows such as health checks, quasi-identifier detection, and risk analysis.
This is the multi-page printable view of this section. Click here to print.
Using Anon
1 - Python SDK
Use this section to install, configure, and validate the Python SDK for Protegrity Anonymization.
Prerequisites
Ensure one of the following is met:
pipis available in your active Python virtual environment.- A conda environment is activated.
Overview
Use the Python SDK to access the Anonymization service programmatically.
1. Install the Anonymization Python SDK
Install the SDK using pip or conda using the version that matches the deployed Anonymization service.
Using pip:
pip install protegrity-anonymization-sdk==2.0.1
Using conda:
conda install -c protegrity protegrity-anonymization-sdk==2.0.1
2. Configure the SDK
Create a ClientConfig pointing to your PPC gateway. Obtain the gateway URL using the following command:
export GATEWAY_URL="https://$(kubectl get configmap/nfa-config -n default -o jsonpath='{.data.FQDN}')"
Then configure the client in Python:
from anonymization_sdk.config import ClientConfig
config = ClientConfig(
base_url="<GATEWAY_URL>/pty/anonymization/v3",
username="<USERNAME>",
password="<PASSWORD>",
verify_ssl=True,
)
Alternatively, configure using environment variables:
export ANON_API_URL="<GATEWAY_URL>/pty/anonymization/v3"
export ANON_USERNAME="anonymization_admin"
export ANON_PASSWORD="StrongPassword123!"
from anonymization_sdk.config import ClientConfig
config = ClientConfig.from_env()
Note: Replace default credentials and URLs for production environments. For example, for
<USERNAME>, enteranonymization_adminand for<PASSWORD>, enterStrongPassword123!.
3. Test the Anonymization Python SDK
from anonymization_sdk import AnonymizationClient
from anonymization_sdk.config import ClientConfig
config = ClientConfig(
base_url="<GATEWAY_URL>/pty/anonymization/v3",
username="anonymization_admin",
password="StrongPassword123!",
verify_ssl=True,
)
with AnonymizationClient(config=config) as client:
print("Connection established successfully.")
The connection is established successfully. If the connection fails, an error message is displayed.
2 - Python SDK Reference
This section describes the classes and methods available in the Anonymization Python SDK, along with usage examples. For installation steps, refer to the Python SDK Installation section.
ClientConfig
Configuration dataclass for AnonymizationClient.
from anonymization_sdk.config import ClientConfig
config = ClientConfig(
base_url="<GATEWAY_URL>/pty/anonymization/v3",
username="anonymization_admin",
password="StrongPassword123!",
verify_ssl=True,
)
Attributes:
- base_url (
str, default"http://localhost:8000") - Base URL of the Anonymization API. - timeout (
float, default30) - Request timeout in seconds. - max_retries (
int, default3) - Retry attempts on connection failure. - verify_ssl (
bool, defaultTrue) - Set toFalsefor self-signed certificates. - username (
str | None, defaultNone) - Username for token-based authentication. - password (
str | None, defaultNone) - Password for token-based authentication. - refresh_buffer (
int, default30) - Seconds before token expiry to proactively refresh. - poll_interval (
float, default2.0) - Seconds between async job status polls. - job_timeout (
float, default600) - Maximum seconds to wait for a job.
Alternatively, configure from environment variables using ClientConfig.from_env().
AnonymizationClient
Synchronous client for the Anonymization API. An AsyncAnonymizationClient is also available with the same interface.
from anonymization_sdk import AnonymizationClient
with AnonymizationClient(config=config) as client:
...
is_healthy
def is_healthy() -> bool
Check whether the API is reachable and healthy.
Returns: bool - True if the API responds with a healthy status, False otherwise.
get_health
def get_health() -> dict[str, Any]
Get detailed health information from the API.
Returns: dict - Health status including version and component states.
Raises:
APIError- If the API returns an error status.
detect_qi
def detect_qi(
data: DataInputType,
*,
mode: DetectionMode | str = DetectionMode.AUTO,
sampling_method: SamplingMethod | str = SamplingMethod.FAST,
cumulative_importance_threshold: float = 0.8,
max_quasi_identifiers: int = 10,
uniqueness_threshold: float = 0.95,
known_identifiers: list[str] | None = None,
known_sensitive: list[str] | None = None,
ignore_columns: list[str] | None = None,
) -> DetectionResult
Detect quasi-identifiers in a dataset.
Parameters:
- data (
DataInputType) - Inline records (List[Dict]), local file path, or cloud URI (s3://,gs://,azure://). - mode (
str, default"auto") - Detection algorithm. One of"auto","ml","heuristic"."ml"mode requires TEAMS (ML-based QI detection). - sampling_method (
str, default"fast") - Sampling strategy. One of"fast","full","adaptive". - cumulative_importance_threshold (
float, default0.8) - Stop adding QIs when cumulative importance reaches this threshold (0.0–1.0). - max_quasi_identifiers (
int, default10) - Maximum number of QIs to return. - uniqueness_threshold (
float, default0.95) - Columns above this uniqueness ratio are flagged as direct identifiers (0.0–1.0). - known_identifiers (
list[str], optional) - Columns you already know are direct identifiers. - known_sensitive (
list[str], optional) - Columns you already know are sensitive. - ignore_columns (
list[str], optional) - Columns to skip during detection.
Returns: DetectionResult - Contains quasi_identifiers, direct_identifiers, sensitive_attributes, attributes, and optional model_metrics.
Raises:
APIError- If the API returns an error.ValidationError- If the request is invalid.
calculate_risk
def calculate_risk(
data: DataInputType,
quasi_identifiers: list[str] | None = None,
*,
risk_threshold: float = 0.2,
suppress_value: str = "*",
include_prosecutor: bool = True,
include_journalist: bool = True,
include_marketer: bool = True,
mlops_config: dict[str, Any] | None = None,
) -> RiskResult
Calculate re-identification risk metrics for a dataset.
Parameters:
- data (
DataInputType) - Inline records, local file path, or cloud URI. - quasi_identifiers (
list[str], optional) - QI column names to include in the risk calculation. - risk_threshold (
float, default0.2) - Records above this threshold are considered “at risk”. - suppress_value (
str, default"*") - Value used to mark suppressed records. - include_prosecutor (
bool, defaultTrue) - Include prosecutor risk model. - include_journalist (
bool, defaultTrue) - Include journalist risk model. - include_marketer (
bool, defaultTrue) - Include marketer risk model. - mlops_config (
dict, optional) - MLOps tracking configuration.
Returns: RiskResult - Contains prosecutor, journalist, marketer risk models, k_anonymity, highest_risk_level, and equivalence class statistics.
anonymize
def anonymize(
data: DataInputType,
*,
privacy_model: PrivacyModel | str = PrivacyModel.K_ANONYMITY,
k: int = 5,
l: int | None = None,
t: float | None = None,
attributes: list[dict[str, Any]] | None = None,
max_suppression: float = 0.0,
output_uri: str | None = None,
output_format: str | None = None,
mlops_config: dict[str, Any] | None = None,
) -> AnonymizeResult
Anonymize data synchronously using the specified privacy model.
Parameters:
- data (
DataInputType) - Inline records, local file path, or cloud URI. - privacy_model (
str, default"k-anonymity") - Privacy model to apply. One of"k-anonymity","l-diversity","t-closeness". - k (
int, default5) - K value for k-anonymity. - l (
int, optional) - L value for l-diversity. - t (
float, optional) - T threshold for t-closeness (0.0–1.0). - attributes (
list[dict], optional) - Attribute configurations. Each dict hasname,type("quasi_identifier","sensitive","identifier","insensitive"), and optionalhierarchy. - max_suppression (
float, default0.0) - Maximum fraction of records allowed to be suppressed (0.0–1.0). - output_uri (
str, optional) - Cloud URI to write results to (e.g."s3://bucket/output.csv").s3://,azure://,gcs://require TEAMS.minio://is available in all tiers. - output_format (
str, optional) - Cloud output format:"csv","parquet","json". Inferred from extension when omitted. - mlops_config (
dict, optional) - MLOps tracking configuration.
Returns: AnonymizeResult - Contains data (inline) or result_path (cloud), row_count, suppressed_count, and metrics.
auto_anonymize
def auto_anonymize(
data: DataInputType,
*,
privacy_model: PrivacyModel | str = PrivacyModel.K_ANONYMITY,
k: int = 5,
l: int | None = None,
t: float | None = None,
mode: DetectionMode | str = DetectionMode.AUTO,
output_uri: str | None = None,
output_format: str | None = None,
mlops_config: dict[str, Any] | None = None,
) -> AutoAnonymizeResult
Automatically detect quasi-identifiers and anonymize in a single call.
Parameters:
- data (
DataInputType) - Inline records, local file path, or cloud URI. - privacy_model (
str, default"k-anonymity") - Privacy model to apply. - k (
int, default5) - K value for k-anonymity. - l (
int, optional) - L value for l-diversity. - t (
float, optional) - T threshold for t-closeness. - mode (
str, default"auto") - QI detection algorithm. One of"auto","ml","heuristic". - output_uri (
str, optional) - Cloud URI for output. - output_format (
str, optional) - Output format:"csv","parquet","json".
Returns: AutoAnonymizeResult - Contains detection results and anonymized data.
apply_anon
def apply_anon(
job_id: str,
data: DataInputType,
*,
mlops_config: dict[str, Any] | None = None,
) -> ApplyResult
Re-apply a previously computed anonymization solution to new data without recomputing the generalization lattice.
Parameters:
- job_id (
str) - Solution identifier returned inAnonymizeResult.job_id. - data (
DataInputType) - New data to apply the solution to. - mlops_config (
dict, optional) - MLOps tracking configuration.
Returns: ApplyResult - Contains anonymized data, row_count, suppressed_count, source_job_id, and privacy_model.
generate_config
def generate_config(
data: DataInputType,
*,
privacy_model: PrivacyModel | str = PrivacyModel.K_ANONYMITY,
k: int = 5,
l: int | None = None,
t: float | None = None,
mode: DetectionMode | str = DetectionMode.AUTO,
**kwargs,
) -> AutoConfigResult
Auto-generate an anonymization configuration from a dataset.
Parameters:
- data (
DataInputType) - Inline records, local file path, or cloud URI. - privacy_model (
str, default"k-anonymity") - Target privacy model. - k (
int, default5) - K value. - l (
int, optional) - L value for l-diversity. - t (
float, optional) - T threshold for t-closeness. - mode (
str, default"auto") - QI detection algorithm. - kwargs - Additional options:
max_suppression,diversity_type,distance_metric,sampling_method.
Returns: AutoConfigResult - Contains detection results and a ready-to-use anonymize_request configuration dict.
validate
def validate(
data: DataInputType,
quasi_identifiers: list[str] | None = None,
*,
privacy_model: PrivacyModel | str = PrivacyModel.K_ANONYMITY,
k: int = 5,
l: int | None = None,
t: float | None = None,
sensitive_attributes: list[str] | None = None,
) -> ValidationResult
Validate that a dataset meets the specified privacy requirements.
Parameters:
- data (
DataInputType) - Dataset to validate. - quasi_identifiers (
list[str], optional) - QI column names to check. - privacy_model (
str, default"k-anonymity") - Privacy model to validate against. - k (
int, default5) - Required k value. - l (
int, optional) - Required l value for l-diversity. - t (
float, optional) - Required t threshold for t-closeness. - sensitive_attributes (
list[str], optional) - Sensitive columns (required for l-diversity and t-closeness).
Returns: ValidationResult - Contains is_valid, model_type, violations, and statistics.
measure
def measure(
original_data: DataInputType,
anonymized_data: DataInputType,
quasi_identifiers: list[str] | None = None,
) -> MetricsResult
Measure information loss between original and anonymized datasets.
Parameters:
- original_data (
DataInputType) - Original dataset. - anonymized_data (
DataInputType) - Anonymized dataset to compare. - quasi_identifiers (
list[str], optional) - QI columns that were generalized.
Returns: MetricsResult - Contains information_loss and detailed per-column metrics.
submit_job
def submit_job(
data: DataInputType,
*,
privacy_model: PrivacyModel | str = PrivacyModel.K_ANONYMITY,
k: int = 5,
l: int | None = None,
t: float | None = None,
attributes: list[dict[str, Any]] | None = None,
max_suppression: float = 0.0,
output_uri: str | None = None,
output_format: str | None = None,
) -> JobResponse
Submit an anonymization job for asynchronous processing.
Parameters: Same as anonymize(), except mlops_config is not available.
Returns: JobResponse - Contains job_id, status, message, and created_at. Use job_id with get_job_status() or wait_for_job().
get_job_status
def get_job_status(job_id: str) -> JobStatusResponse
Get the current status of an asynchronous job.
Parameters:
- job_id (
str) - Job identifier returned bysubmit_job().
Returns: JobStatusResponse - Contains job_id, status (pending, running, completed, failed, cancelled), progress (0–100), message, timestamps, result_path, and error.
Raises:
APIError- If the job is not found.
wait_for_job
def wait_for_job(
job_id: str,
*,
poll_interval: float = 2.0,
timeout: float = 600.0,
callback: Callable | None = None,
) -> JobStatusResponse
Block until an asynchronous job reaches a terminal state.
Parameters:
- job_id (
str) - Job identifier. - poll_interval (
float, default2.0) - Seconds between status polls. - timeout (
float, default600.0) - Maximum seconds to wait. - callback (
callable, optional) - Called with the currentJobStatusResponseafter each poll.
Returns: JobStatusResponse - Final status at the terminal state.
Raises:
APIError- If the job ends in afailedstate.TimeoutError- If the job does not complete withintimeout.
cancel_job
def cancel_job(job_id: str) -> None
Cancel a pending or running job.
Parameters:
- job_id (
str) - Job identifier.
Raises:
APIError- If the job is not found or cannot be cancelled.
list_jobs
def list_jobs(
*,
status: JobStatus | str | None = None,
limit: int = 100,
offset: int = 0,
) -> JobListResult
List jobs with optional status filter and pagination. Returns newest jobs first.
Parameters:
- status (
str, optional) - Filter by status (e.g."completed","failed"). - limit (
int, default100) - Page size (1–1000). - offset (
int, default0) - Page offset.
Returns: JobListResult - Contains jobs, total, limit, and offset.
Patterns
Custom detection patterns stored server-side, used by detect_qi to classify columns by name and value.
create_pattern
def create_pattern(
name: str,
classification: str,
column_patterns: list[str],
*,
priority: int = 50,
value_patterns: list[str] | None = None,
min_match_ratio: float = 0.8,
description: str | None = None,
) -> Pattern
Create a custom QI detection pattern.
Parameters:
- name (
str) - Unique pattern name (e.g."customer_id"). - classification (
str) - One of"DI"(Direct Identifier),"QI"(Quasi-Identifier),"SI"(Sensitive Identifier),"NSI"(Non-Sensitive Identifier). - column_patterns (
list[str]) - Column name patterns to match. Case-insensitive; use"*"as wildcard (e.g.["*_id", "user*"]). - priority (
int, default50) - Priority level 1–1000; lower = checked first. - value_patterns (
list[str], optional) - Regex patterns for value-level validation. - min_match_ratio (
float, default0.8) - Minimum fraction of values that must match (0–1). - description (
str, optional) - Human-readable description.
Returns: Pattern - The created pattern with its assigned ID and metadata.
Raises:
APIError- If creation fails (e.g. duplicate name).ValidationError- If parameters are invalid.
list_patterns
def list_patterns(
classification: str | None = None,
) -> PatternListResult
List all custom detection patterns.
Parameters:
- classification (
str, optional) - Filter by classification:"DI","QI","SI", or"NSI".
Returns: PatternListResult - Contains the list of patterns and total count.
get_pattern
def get_pattern(pattern_id: str) -> Pattern
Retrieve a pattern by ID.
Parameters:
- pattern_id (
str) - Pattern ID to retrieve.
Returns: Pattern - The requested pattern.
Raises:
APIError- If the pattern is not found.
update_pattern
def update_pattern(
pattern_id: str,
*,
name: str | None = None,
classification: str | None = None,
column_patterns: list[str] | None = None,
priority: int | None = None,
value_patterns: list[str] | None = None,
min_match_ratio: float | None = None,
description: str | None = None,
) -> Pattern
Update an existing pattern. Only provided fields are updated; others remain unchanged.
Parameters:
- pattern_id (
str) - ID of the pattern to update. - name (
str, optional) - New name. - classification (
str, optional) - New classification. - column_patterns (
list[str], optional) - New column name patterns. - priority (
int, optional) - New priority (1–1000). - value_patterns (
list[str], optional) - New value regex patterns. - min_match_ratio (
float, optional) - New minimum match ratio (0–1). - description (
str, optional) - New description.
Returns: Pattern - The updated pattern.
Raises:
APIError- If the pattern is not found or update fails.ValidationError- If parameters are invalid.
delete_pattern
def delete_pattern(pattern_id: str) -> dict[str, Any]
Delete a pattern by ID.
Parameters:
- pattern_id (
str) - ID of the pattern to delete.
Returns: dict - Confirmation message.
Raises:
APIError- If the pattern is not found.
delete_all_patterns
def delete_all_patterns() -> dict[str, Any]
Delete all custom patterns. Built-in patterns from the server configuration are not affected.
Returns: dict - Contains the count of deleted patterns.
reload_patterns
def reload_patterns() -> dict[str, Any]
Reload patterns from server storage. Use after manual file edits to sync changes.
Returns: dict - Contains the count of reloaded patterns.
Differential Privacy (TEAMS)
Requires the differential_privacy feature (TEAMS and above).
dp_compute
def dp_compute(
data: DataInputType,
*,
mechanism: DPMechanismType | str = DPMechanismType.MEAN,
column: str | None = None,
columns: list[str] | None = None,
group_by: str | None = None,
epsilon: float = 1.0,
delta: float = 0.0,
noise_type: DPNoiseType | str = DPNoiseType.LAPLACE,
bounds: tuple | None = None,
bins: int | None = None,
histogram_range: tuple | None = None,
session_id: str | None = None,
predicate: str | None = None,
candidates: list | None = None,
utility_scores: list[float] | None = None,
sensitivity: float | None = None,
epsilon_map: dict[str, float] | None = None,
min_group_size: int | None = None,
) -> DPComputeResult
Compute a differentially private statistic on a data column.
Parameters:
- data (
DataInputType) - Inline records, local file path, or cloud URI. - mechanism (
str, default"mean") - DP mechanism:"mean","sum","variance","histogram","count","exponential". - column (
str, optional) - Column name for single-column queries. - columns (
list[str], optional) - Column names for multi-column queries. - group_by (
str, optional) - Categorical column to group by. - epsilon (
float, default1.0) - Privacy parameter ε (must be > 0). - delta (
float, default0.0) - Privacy parameter δ (0 ≤ δ < 1). - noise_type (
str, default"laplace") - Noise distribution:"laplace"or"gaussian". - bounds (
tuple, optional) -(lower, upper)clipping bounds. Required for mean/sum/variance. - bins (
int, optional) - Number of histogram bins (histogram mechanism only). - histogram_range (
tuple, optional) -(min, max)range for histogram bins. - session_id (
str, optional) - Budget session ID for cumulative epsilon tracking. - predicate (
str, optional) - Filter expression (e.g."> 50","<= 100"). - candidates (
list, optional) - Candidate outputs (exponential mechanism only). - utility_scores (
list[float], optional) - Utility scores for candidates (exponential only). - sensitivity (
float, optional) - Utility function sensitivity (exponential only). - epsilon_map (
dict[str, float], optional) - Per-column or per-group epsilon overrides. - min_group_size (
int, optional) - Minimum rows per group (default 5).
Returns: DPComputeResult - Contains private_value (single query) or results dict (multi-column or grouped queries).
dp_stream_update
def dp_stream_update(
session_id: str | None = None,
data: DataInputType | None = None,
*,
column: str | None = None,
columns: list[str] | None = None,
group_by: str | None = None,
mechanism: DPStreamMechanismType | str | None = None,
epsilon: float | None = None,
delta: float | None = None,
noise_type: DPNoiseType | str | None = None,
bounds: tuple | None = None,
get_result: bool = False,
window_size: int | None = None,
epsilon_map: dict[str, float] | None = None,
min_group_size: int | None = None,
budget_session_id: str | None = None,
) -> DPStreamResult
Feed a batch of data into a streaming DP session. On the first call for a session_id, provide mechanism, epsilon, and bounds. Subsequent calls only need session_id, data, and column.
Parameters:
- session_id (
str, optional) - Unique session identifier. - data (
DataInputType, optional) - Batch of records to process. - column (
str, optional) - Column for single-column streaming. - columns (
list[str], optional) - Columns for multi-column streaming. - group_by (
str, optional) - Categorical column to group by. - mechanism (
str, optional) - Streaming mechanism. Required on first call. - epsilon (
float, optional) - Privacy ε. Required on first call. - delta (
float, optional) - Privacy δ. - noise_type (
str, optional) - Noise distribution. - bounds (
tuple, optional) - Clipping bounds. Required on first call (except count). - get_result (
bool, defaultFalse) - IfTrue, also return the current private result. - window_size (
int, optional) - Window size for sliding/tumbling window mechanisms. - epsilon_map (
dict[str, float], optional) - Per-column or per-group epsilon overrides. - min_group_size (
int, optional) - Minimum rows per group (default 5). - budget_session_id (
str, optional) - Link to a budget session for automatic epsilon deduction.
Returns: DPStreamResult - Session status and optional current private results.
dp_stream_delete
def dp_stream_delete(session_id: str) -> None
Delete a streaming DP session.
Parameters:
- session_id (
str) - Session ID to delete.
dp_stream_list_sessions
def dp_stream_list_sessions() -> list
List all active streaming DP sessions.
Returns: list - Each entry contains session_id, mechanism, column, batches_processed, and total_count.
dp_budget_create
def dp_budget_create(
session_id: str,
epsilon_budget: float,
delta_budget: float = 0.0,
composition: str = "basic",
) -> DPBudgetStatus
Create a privacy budget session for tracking cumulative epsilon spend across multiple queries.
Parameters:
- session_id (
str) - Unique session identifier. - epsilon_budget (
float) - Total epsilon budget for this session. - delta_budget (
float, default0.0) - Total delta budget. - composition (
str, default"basic") - Composition mode:"basic"or"rdp". RDP requiresdelta_budget > 0and yields tighter privacy accounting.
Returns: DPBudgetStatus - Initial budget state.
dp_budget_status
def dp_budget_status(session_id: str) -> DPBudgetStatus
Query the current epsilon spend and remaining budget for a session.
Parameters:
- session_id (
str) - Session to query.
Returns: DPBudgetStatus - Contains current spend and remaining budget.
dp_budget_delete
def dp_budget_delete(session_id: str) -> None
Delete a privacy budget session.
Parameters:
- session_id (
str) - Session ID to delete.
dp_advise_composition
def dp_advise_composition(
epsilon_budget: float,
num_queries: int,
delta_budget: float = 0.0,
delta_per_query: float = 0.0,
) -> dict
Get composition advice for a planned set of queries. Returns optimal per-query epsilon under basic and RDP composition with a recommendation.
Parameters:
- epsilon_budget (
float) - Total epsilon budget available. - num_queries (
int) - Number of planned queries. - delta_budget (
float, default0.0) - Total delta budget (required for RDP comparison). - delta_per_query (
float, default0.0) - Delta per query for Gaussian noise.0= Laplace.
Returns: dict - Contains basic and rdp analysis, recommendation, and savings_pct.
Audit
audit_list
def audit_list(
*,
operation: str | None = None,
status: str | None = None,
limit: int = 50,
offset: int = 0,
) -> list[AuditEntry]
List audit log entries.
Parameters:
- operation (
str, optional) - Filter by operation (e.g."dp_compute","anonymize_sync"). - status (
str, optional) - Filter by outcome:"success"or"error". - limit (
int, default50) - Max entries to return (1–500). - offset (
int, default0) - Pagination offset.
Returns: list[AuditEntry] - Matching audit log entries.
audit_get
def audit_get(entry_id: str) -> AuditEntry
Retrieve a single audit entry by ID.
Parameters:
- entry_id (
str) - Audit entry ID.
Returns: AuditEntry - Full entry details.
Raises:
APIError- If the entry is not found.
Enumerations (PrivacyModel, DetectionMode, SamplingMethod, JobStatus, DPMechanismType, DPNoiseType, DPStreamMechanismType) are importable from anonymization_sdk.models. Exceptions (APIError, ValidationError, AnonymizationConnectionError, TierRestrictionError) are importable from anonymization_sdk.exceptions.