Protegrity Anonymization

• 1: Introduction
• 2: Understanding the Architecture and Components
• 3: Prerequisites for Deployment
• 4: Server Installation
• 5: Upgrading Protegrity Anonymization
• 6: User Creation in PPC
• 7: Using Anon
• 7.1: Python SDK
• 7.2: Python SDK Reference
• 8: Uninstallation and Cleanup
• 9: Troubleshooting

1 - Introduction

Organizations today collect vast amounts of personal data, providing valuable insights into individuals’ habits, purchasing trends, health, and preferences. This information helps businesses refine their strategies, develop products, and drive success. However, much of this data is highly sensitive and private. Organizations must implement robust protection measures aligned with compliance requirements and business needs.

To safeguard personal data, pseudonymization replaces direct identifiers with encrypted or tokenized values. This approach allows data processing while minimizing exposure to sensitive attributes. Because pseudonymized data can be re-identified with access to the decryption mechanism, it enables controlled usage while maintaining privacy. However, as more fields, particularly quasi-identifiers, are pseudonymized to prevent re-identification, the overall utility of the data may decrease.

For higher privacy protection, Anonymization adds security by generalizing both personally identifiable information (PII) and quasi-identifiers. Generalization prevents re-identification even when multiple data points are combined. Anonymization techniques include removing key attributes and generalizing data to broader categories. For example, you can replace an address with only a city or state. Organizations retain analytical value while minimizing re-identification risk. This ensures compliance with privacy regulations and ethical data practices.

2 - Understanding the Architecture and Components

Protegrity Anonymization processes datasets through generalization to ensure the risk of re-identification stays within tolerable thresholds. The anonymization process affects data utility. However, Protegrity Anonymization optimizes the privacy-utility trade-off to maximize data quality within defined privacy goals.

Protegrity Anonymization uses Kubernetes for data anonymization at scale and provides deployment support for AWS EKS.

Component Diagram

The following diagram shows the deployment architecture and communication between components.

graph TD
    User([User / Client Application]) -->|HTTPS :443| GW[PPC Gateway]
    User --> SDK[Anonymization Python SDK]
    SDK -->|HTTPS :443| GW

    GW -->|:8000  /pty/anonymization/v3| App[anon-app\nREST Server]
    GW -->|:8786  /pty/dask| Sched[anon-scheduler\nDask Scheduler]

    App -->|:5432| DB[(anon-db\nPostgreSQL)]
    App --> Sched
    Sched -->|random ports| W1[anon-worker]
    Sched -->|random ports| W2[anon-worker]
    W1 <-->|read / write| S3[(S3 Bucket\nAnon-Storage)]
    W2 <-->|read / write| S3

    subgraph EKS [EKS Cluster  ·  anon-ns]
        GW
        App
        DB
        Sched
        W1
        W2
    end

    subgraph AWS
        S3
        IAM[IAM Role\nPod Identity]
    end

    App -.->|assumes| IAM
    IAM -.->|grants access| S3

Communication Ports

The following table lists the ports used for communication between the components.

Components

The deployment consists of the following components:

• Anonymization REST Server or anon-app: Exposes the REST API at /pty/anonymization/v3. It receives anonymization requests, submits tasks to the Dask scheduler, and stores job metadata in PostgreSQL. The system queues and executes tasks in first-in first-out (FIFO) order.

  Note: Only one Anonymization task runs at a time.

• Dask Scheduler or anon-scheduler: Receives tasks from the REST server and distributes them to available Dask workers. Communicates with workers over random ports.

• Dask Workers or anon-worker: Executes the anonymization algorithm. Workers read input datasets from and write output datasets to the S3 bucket. Workers scale horizontally using replicaCount.

• Anon-DB or anon-db: Stores job metadata, including status, configuration, and result references, in a PostgreSQL database.

• S3 Storage or Anon-Storage: Stores input and output datasets in an S3 bucket. An IAM role using EKS Pod Identity grants workers access.

• Python SDK: Interacts programmatically with the REST server using the Python client library. For more information about the Python SDK, refer to the Python SDK Installation section.

3 - Prerequisites for Deployment

Ensure the following prerequisites are met:

Tools

Install and configure the following tools on the jumpbox:

• opentofu 1.10 or later. For more information about installation steps, refer to OpenTofu Installation.
• helm 3.12+, kubectl 1.27+, and aws CLI v2 with access to the Protegrity Provisioned Cluster (PPC).

AWS Setup

Ensure that the following AWS requirements are met:

• A Protegrity Provisioned Cluster (PPC) is available. For more information about PPC, refer to Protegrity Provisioned Cluster.

• AWS CLI credentials are configured using aws configure.

• The AWS credentials have permission to provision IAM roles, S3 buckets, EKS Pod Identity associations, and Karpenter node resources. If you encounter AccessDenied errors, refer to IAM Permissions for Installation.

• The jumpbox can connect to the required registry. If you are not already authenticated, log in to it.

  • To authenticate to the Protegrity Container Registry (PCR), use the following command with credentials from the My.Protegrity portal during account creation:

  echo "<password>" | helm registry login registry.protegrity.com --username "<username>" --password-stdin
  

  • For a local registry, use your credentials and local registry endpoint as required.

EKS Cluster Requirements

Ensure that the EKS cluster meets the following requirements:

• Kubernetes 1.27 or later is installed.
• The default StorageClass is used unless you override it. Verify the StorageClass with kubectl get storageclass.
• The EKS Pod Identity Agent add-on is enabled on the cluster.
• Karpenter 1.0 or later is installed and configured on the cluster.

IAM Permissions for Installation

The AWS credentials used during installation require the following permissions. These permissions are required only during installation and can be reduced afterward.

4 - Server Installation

Note: This guide uses version 2.0.1 throughout. Replace 2.0.1 with the version you are installing, including in the Python SDK Installation section.

This project deploys the Protegrity Anonymization service on Amazon EKS as part of the Protegrity AI Team Edition.

Deployment Steps

1. Provision Cloud Infrastructure with OpenTofu

The OpenTofu module creates the S3 bucket, IAM role, EKS Pod Identity association, and Karpenter NodePool. It also writes anonymization-values.yaml with cloud-specific Helm overrides.

Option A - You already have a root module

Add the following module block to your existing root module:

module "anonymization" {
  source = "oci://registry.protegrity.com/anonymization/anonymization-server/opentofu/anonymization?tag=aws-2.0.1"

  cluster_name  = "<CLUSTER_NAME>"
  bucket_name   = "<globally-unique-s3-bucket-name>"
  oci_host      = "registry.protegrity.com"
  environment   = "production"
  chart_version = "2.0.1"
}

Note: This module requires the hashicorp/aws (~> 5.0) and hashicorp/kubernetes (~> 2.35) providers.

If your root module does not already declare them, add the following to your terraform {} block:

required_providers {
  aws = {
    source  = "registry.opentofu.org/hashicorp/aws"
    version = "~> 5.0"
  }
  kubernetes = {
    source  = "registry.opentofu.org/hashicorp/kubernetes"
    version = "~> 2.35"
  }
}

Configure the kubernetes provider to connect to the target EKS cluster:

data "aws_eks_cluster" "this" {
  name = "<your-cluster-name>"
}

data "aws_eks_cluster_auth" "this" {
  name = "<your-cluster-name>"
}

provider "kubernetes" {
  host                   = data.aws_eks_cluster.this.endpoint
  cluster_ca_certificate = base64decode(data.aws_eks_cluster.this.certificate_authority[0].data)
  token                  = data.aws_eks_cluster_auth.this.token
}

Then run the following commands:

tofu init
tofu plan
tofu apply

Option B - You do not have a root module

Create a new directory with a single main.tf:

terraform {
  required_version = "~> 1.10"

  required_providers {
    aws = {
      source  = "registry.opentofu.org/hashicorp/aws"
      version = "~> 5.0"
    }
    kubernetes = {
      source  = "registry.opentofu.org/hashicorp/kubernetes"
      version = "~> 2.35"
    }
  }
}

variable "cluster_name" {
  type        = string
  description = "EKS cluster name."
}

variable "environment" {
  type    = string
  default = "production"
}

data "aws_eks_cluster" "this" {
  name = var.cluster_name
}

data "aws_eks_cluster_auth" "this" {
  name = var.cluster_name
}

provider "kubernetes" {
  host                   = data.aws_eks_cluster.this.endpoint
  cluster_ca_certificate = base64decode(data.aws_eks_cluster.this.certificate_authority[0].data)
  token                  = data.aws_eks_cluster_auth.this.token
}

module "anonymization" {
  source = "oci://registry.protegrity.com/anonymization/anonymization-server/opentofu/anonymization?tag=aws-2.0.1"

  cluster_name  = var.cluster_name
  bucket_name   = "<globally-unique-s3-bucket-name>"
  oci_host      = "registry.protegrity.com"
  environment   = var.environment
  chart_version = "2.0.1"
}

output "helm_install" {
  value = module.anonymization.helm_install
}

Then run the following commands:

tofu init
tofu plan -var="cluster_name=<cluster-name>" -var="bucket_name=<globally-unique-bucket>"
tofu apply -var="cluster_name=<cluster-name>" -var="bucket_name=<globally-unique-bucket>"

Note: tofu output -raw helm_install prints a ready-to-run Helm command with the correct version and namespace.

2. Deploy with Helm

The OpenTofu module writes an anonymization-values.yaml file in your working directory. Passwords are required only for the first installation. They are preserved automatically for later upgrades.

export MLOPS_PASSWORD=<MLOPS_PASSWORD>
export JOBSTATE_PASSWORD=<JOBSTATE_PASSWORD>

helm upgrade --install anonymization \
  oci://registry.protegrity.com/anonymization/anonymization-server/helm/anonymization \
  --version 2.0.1 \
  --namespace anon-ns --create-namespace \
  --values anonymization-values.yaml \
  --set database.auth.mlopsPassword=$MLOPS_PASSWORD \
  --set database.auth.jobstatePassword=$JOBSTATE_PASSWORD

Note: Ensure that “<JOBSTATE_PASSWORD>” contains only alphanumeric characters.

For upgrades, omit the --set flags because the passwords are already stored in the cluster secret:

helm upgrade anonymization \
  oci://registry.protegrity.com/anonymization/anonymization-server/helm/anonymization \
  --version 2.0.1 \
  --namespace anon-ns \
  --values anonymization-values.yaml

3. Monitor

1. Monitor the deployment process using the following command.

   kubectl get pods -n anon-ns
   

   Verify that all pods are in the Running state. The following is a sample output.

   NAME                                  READY   STATUS    RESTARTS   AGE
   anon-app-depl-f5c4d4cd6-42wgn         1/1     Running   0          3m20s
   anon-db-depl-0                        1/1     Running   0          3m20s
   anon-scheduler-depl-7b87fcb74-l5q6v   1/1     Running   0          3m20s
   anon-worker-depl-7c4d95496f-djw7f     1/1     Running   0          3m20s
   anon-worker-depl-7c4d95496f-gnnvp     1/1     Running   0          3m20s
   

2. Verify that the Anonymization service is deployed.

   kubectl get svc -n anon-ns
   

   The following is the sample output.

   NAME            TYPE        CLUSTER-IP       EXTERNAL-IP   PORT(S)    AGE
   anon-app-svc    ClusterIP   172.20.151.139   <none>        8000/TCP   61s
   anon-dask-svc   ClusterIP   172.20.224.133   <none>        8786/TCP   61s
   

   For more information about the Python SDK, refer to Python SDK Reference.

5 - Upgrading Protegrity Anonymization

This section describes how to upgrade Protegrity Anonymization to the latest version.

Note: If you are upgrading across major versions, review the release notes for any breaking changes before proceeding.

1. Run helm upgrade with the new --version value.

   Note: Passwords are preserved during upgrades. Omit the --set database.auth.* flags.

   helm upgrade anonymization \
     oci://registry.protegrity.com/anonymization/anonymization-server/helm/anonymization \
     --version <NEW_VERSION> \
     --namespace anon-ns \
     --values <VALUES_FILE>
   

2. After upgrading, verify that all pods are in the Running state.

   kubectl get pods -n anon-ns
   

6 - User Creation in PPC

Warning: Use the -k flag only in development or trusted internal environments. Do not use it in production, where valid certificates must be configured.

Role Update and User Creation

After deployment, update the default anonymization_administrator role to include the can_create_token permission and then create a user with this role.

1. Update anonymization_administrator role permission

export GATEWAY_URL="https://$(kubectl get configmap/nfa-config -n default -o jsonpath='{.data.FQDN}')"

# 1. Obtain an Authentication Token
TOKEN=$(curl -sk -X POST "${GATEWAY_URL}/api/v1/auth/login/token" \
  -H 'Content-Type: application/x-www-form-urlencoded' \
  -d 'loginname=admin&password=Admin123!' \
  -D - -o /dev/null | grep -i 'pty_access_jwt_token' | awk '{print $2}' | tr -d '\r\n')

curl -sk -X PUT \
  "${GATEWAY_URL}/pty/v1/auth/roles" \
  -H 'accept: application/json' \
  -H "Authorization: Bearer ${TOKEN}" \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "anonymization_administrator",
    "description": "Administrator role",
    "permissions": [
      "can_create_token",
      "anonymization_operations_admin"
    ]
  }'

2. Create a user with the anonymization_administrator role

Use the following request payload when creating the user:

{
  "username": "anonymization_admin",
  "email": "anonadmin@example.com",
  "firstName": "Anon",
  "lastName": "<USERNAME>",
  "password": "<PASSWORD>",
  "roles": [
    "anonymization_administrator"
  ]
}

Create the user with the following API call:

curl -sk -X POST \
  "${GATEWAY_URL}/pty/v1/auth/users" \
  -H 'accept: application/json' \
  -H "Authorization: Bearer ${TOKEN}" \
  -H 'Content-Type: application/json' \
  -d '{
    "username": "anonymization_admin",
    "email": "anonadmin@example.com",
    "firstName": "Anon",
    "lastName": "<USERNAME>",
    "password": "<PASSWORD>",
    "roles": [
      "anonymization_administrator"
    ]
  }'

Note: In production environments, replace the default credentials and URLs. For example, for <USERNAME>, enter User and for <PASSWORD>, enter StrongPassword123!.

7 - Using Anon

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.

7.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:

• pip is 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>, enter anonymization_admin and for <PASSWORD>, enter StrongPassword123!.

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.

7.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, default 30) - Request timeout in seconds.
• max_retries (int, default 3) - Retry attempts on connection failure.
• verify_ssl (bool, default True) - Set to False for self-signed certificates.
• username (str | None, default None) - Username for token-based authentication.
• password (str | None, default None) - Password for token-based authentication.
• refresh_buffer (int, default 30) - Seconds before token expiry to proactively refresh.
• poll_interval (float, default 2.0) - Seconds between async job status polls.
• job_timeout (float, default 600) - 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, default 0.8) - Stop adding QIs when cumulative importance reaches this threshold (0.0–1.0).
• max_quasi_identifiers (int, default 10) - Maximum number of QIs to return.
• uniqueness_threshold (float, default 0.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, default 0.2) - Records above this threshold are considered “at risk”.
• suppress_value (str, default "*") - Value used to mark suppressed records.
• include_prosecutor (bool, default True) - Include prosecutor risk model.
• include_journalist (bool, default True) - Include journalist risk model.
• include_marketer (bool, default True) - 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, default 5) - 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 has name, type ("quasi_identifier", "sensitive", "identifier", "insensitive"), and optional hierarchy.
• max_suppression (float, default 0.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, default 5) - 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 in AnonymizeResult.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, default 5) - 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, default 5) - 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 by submit_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, default 2.0) - Seconds between status polls.
• timeout (float, default 600.0) - Maximum seconds to wait.
• callback (callable, optional) - Called with the current JobStatusResponse after each poll.

Returns: JobStatusResponse - Final status at the terminal state.

Raises:

• APIError - If the job ends in a failed state.
• TimeoutError - If the job does not complete within timeout.

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, default 100) - Page size (1–1000).
• offset (int, default 0) - 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, default 50) - Priority level 1–1000; lower = checked first.
• value_patterns (list[str], optional) - Regex patterns for value-level validation.
• min_match_ratio (float, default 0.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, default 1.0) - Privacy parameter ε (must be > 0).
• delta (float, default 0.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, default False) - If True, 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, default 0.0) - Total delta budget.
• composition (str, default "basic") - Composition mode: "basic" or "rdp". RDP requires delta_budget > 0 and 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, default 0.0) - Total delta budget (required for RDP comparison).
• delta_per_query (float, default 0.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, default 50) - Max entries to return (1–500).
• offset (int, default 0) - 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.

8 - Uninstallation and Cleanup

To remove the Anonymization service and all associated Kubernetes and AWS resources:

1. Uninstall the Helm release

# Removes all chart resources. PVC data is RETAINED by default (retentionPolicy: whenDeleted=Retain).
helm uninstall anonymization -n anon-ns

# To also delete the database PVC:
kubectl delete pvc -n anon-ns -l app.kubernetes.io/name=anonymization

Optionally, delete the namespace after the Helm release and PVCs are removed using the following command:

kubectl delete namespace anon-ns

2. Destroy OpenTofu infrastructure

Warning: tofu destroy deletes the S3 bucket, including all stored input and output datasets. Back up any data you need before proceeding.

# Destroys the Karpenter nodes, IAM role, Pod Identity association, and S3 bucket.
tofu destroy -var="cluster_name=<CLUSTER_NAME>"

Note: By default, tofu destroy prompts for confirmation before deleting resources. Add -auto-approve to skip the prompt.

9 - Troubleshooting

Diagnostic Flowchart

flowchart TD
    Start[Problem?] --> ServerRunning{Server pods Running?}
    ServerRunning -->|No| PodIssue[Check pod logs<br/>kubectl logs -n anon-ns]
    ServerRunning -->|Yes| APIWorks{Does API respond?}
    APIWorks -->|No| NetworkIssue[Check Service and Gateway<br/>kubectl get svc -n anon-ns]
    APIWorks -->|Yes| ValidationFails{Request fails?}
    ValidationFails -->|Yes| ConfigIssue[Review request config<br/>Check privacy model params]
    ValidationFails -->|No| OtherIssue[Check logs or open a support ticket]
    style Start fill:#e1f5ff

Common Issues

Pods not reaching the Running state

kubectl describe pod -n anon-ns <POD_NAME>
kubectl logs -n anon-ns <POD_NAME>

Check that the Karpenter NodePool provisioned the required nodes. Ensure that OpenTofu created the S3 bucket and IAM role.

API returns 401 Unauthorized

Ensure that the login token is valid and has not expired. Re-authenticate using the login endpoint and update the Authorization header.

API returns 403 Forbidden

Verify that the user’s role includes the can_create_token permission. Follow the User Creation in PPC steps to confirm role configuration.

OpenTofu errors on tofu apply

• AccessDenied: The AWS credentials do not have the required IAM permissions. For more information about permissions, refer to IAM Permissions for Installation section.
• BucketAlreadyExists: An S3 bucket with the same name already exists. Choose a unique bucket name or import the existing bucket into the OpenTofu state.

Helm upgrade fails with an immutable field error

Delete the existing release, then reinstall:

helm uninstall anonymization -n anon-ns
# Re-run the install command

For more troubleshooting guidance, refer to Protegrity Support.