from __future__ import annotations
"""Privacy resource for the CorePlexML SDK."""
from coreplexml.exceptions import CorePlexMLError
from coreplexml._http import HTTPClient
[docs]
class PrivacyResource:
"""Privacy Suite -- PII detection and data transformation.
Supports 72+ PII types across HIPAA, GDPR, PCI-DSS, and CCPA
compliance profiles. Create policies, scan datasets for PII,
and apply transformations (masking, hashing, redaction, etc.).
"""
def __init__(self, http: HTTPClient):
self._http = http
@staticmethod
def _normalize_policy(payload: dict) -> dict:
if not isinstance(payload, dict):
return {}
out = dict(payload)
if "id" not in out and out.get("policy_id"):
out["id"] = out["policy_id"]
if isinstance(out.get("policy"), dict):
return out["policy"]
return out
@staticmethod
def _normalize_session(payload: dict) -> dict:
if not isinstance(payload, dict):
return {}
out = dict(payload)
if "id" not in out and out.get("session_id"):
out["id"] = out["session_id"]
if isinstance(out.get("session"), dict):
return out["session"]
return out
[docs]
def list_policies(self, project_id: str | None = None, limit: int = 50, offset: int = 0) -> dict:
"""List privacy policies.
Args:
project_id: Filter by project UUID (optional).
limit: Maximum results (default 50).
offset: Pagination offset.
Returns:
Dictionary with ``items`` list and ``total`` count.
"""
params: dict = {"limit": limit, "offset": offset}
if project_id:
params["project_id"] = project_id
return self._http.get("/api/privacy/policies", params=params)
[docs]
def create_policy(self, project_id: str, name: str, profile: str | None = None, description: str = "") -> dict:
"""Create a new privacy policy.
Args:
project_id: UUID of the project.
name: Policy name.
profile: Compliance profile -- ``hipaa``, ``gdpr``, ``pci_dss``, or ``ccpa`` (optional).
description: Optional description.
Returns:
Created policy dictionary.
"""
body: dict = {"project_id": project_id, "name": name, "description": description}
if profile:
body["profile"] = profile
data = self._http.post("/api/privacy/policies", json=body)
return self._normalize_policy(data)
[docs]
def get_policy(self, policy_id: str) -> dict:
"""Get privacy policy details.
Args:
policy_id: UUID of the policy.
Returns:
Policy dictionary.
"""
data = self._http.get(f"/api/privacy/policies/{policy_id}")
return self._normalize_policy(data)
[docs]
def delete_policy(self, policy_id: str) -> dict:
"""Delete a privacy policy.
Args:
policy_id: UUID of the policy.
Returns:
Empty dictionary on success.
"""
return self._http.delete(f"/api/privacy/policies/{policy_id}")
[docs]
def create_session(self, policy_id: str, dataset_id: str) -> dict:
"""Create a privacy processing session.
Args:
policy_id: UUID of the privacy policy.
dataset_id: UUID of the dataset to scan.
Returns:
Created session dictionary.
"""
data = self._http.post("/api/privacy/sessions", json={"policy_id": policy_id, "dataset_id": dataset_id})
return self._normalize_session(data)
[docs]
def detect(
self,
session_id: str,
wait: bool = True,
interval: float = 2.0,
timeout: float = 300.0,
) -> dict:
"""Run PII detection on a session.
Args:
session_id: UUID of the privacy session.
wait: Wait for the async job to finish and return findings.
interval: Polling interval in seconds when ``wait=True``.
timeout: Maximum wait time in seconds when ``wait=True``.
Returns:
Detection results with PII findings.
"""
queued = self._http.post(f"/api/privacy/sessions/{session_id}/detect")
if not wait:
return queued
job_id = queued.get("job_id")
if job_id:
job = self._http.poll_job(job_id, interval=interval, timeout=timeout)
if job.get("status") in ("failed", "error"):
raise CorePlexMLError(
f"Privacy detection failed for session {session_id}: {job.get('error', 'unknown')}"
)
findings = self.results(session_id).get("results", [])
return {
"session_id": session_id,
"job_id": job_id,
"status": "completed",
"findings": findings,
}
[docs]
def anonymize(
self,
session_id: str,
wait: bool = True,
interval: float = 2.0,
timeout: float = 300.0,
) -> dict:
"""Apply anonymization transformations to a session.
Alias for :meth:`transform` — provided for API consistency.
"""
return self.transform(session_id, wait=wait, interval=interval, timeout=timeout)
[docs]
def results(self, session_id: str) -> dict:
"""Get full results for a privacy session.
Args:
session_id: UUID of the privacy session.
Returns:
Complete session results including detections and transformations.
"""
return self._http.get(f"/api/privacy/sessions/{session_id}/results")