dpdpguard-python-sdk

DPDP Guard Server SDK (Python) — typed API client, consent gate, token broker helper, audit-hash verifier, and webhook signature verifier over DPDP Guard's public /api/v1.

PyPI distribution: dpdpguard-sdk · Import as: dpdpguard

This SDK mirrors @dpdpguard/server (the Node/TypeScript reference implementation) method for method, so behavior — auth modes, idempotency headers, error codes, audit-hash canonicalization — stays identical across ecosystems.

Install

pip install dpdpguard-sdk

Or with uv:

uv add dpdpguard-sdk

Quickstart

import os
from dpdpguard import DpdpGuardClient, has_consent, verify_webhook_signature

client = DpdpGuardClient(
    "https://<your-deployment>.example.com",
    api_key=os.environ["DPDP_SERVICE_API_KEY"],  # required for broker_token() only
)

# Mint a brokered principal access token for a known user.
client.broker_token(external_id)

# Now authenticated calls use that token automatically.
result = client.list_dsr_requests()
client.create_dsr_request(organization_id=org_id, type="erasure")

# Public reads need no auth at all.
org = client.get_organization("acme")
notices = client.get_notices(org["orgId"])

# Verify an inbound webhook.
ok = verify_webhook_signature(webhook_secret, raw_body, request.headers.get("x-dpdp-signature"))

Use it as a context manager to close the underlying httpx.Client automatically:

with DpdpGuardClient(base_url) as client:
    org = client.get_organization("acme")

DpdpGuardClient

DpdpGuardClient(
    base_url: str,
    *,
    api_key: str | None = None,
    access_token: str | None = None,
    http_client: httpx.Client | None = None,
)

• base_url — your DPDP Guard deployment's base URL.
• api_key — service API key; required for broker_token().
• access_token — a brokered principal access token, if you already minted one.
• http_client — override for testing/transport customization; defaults to a new httpx.Client. The client owns and closes it unless one was passed in.

Every non-2xx response raises a DpdpGuardApiError with a code from the error catalog (err.code, e.g. "NOT_FOUND") and the HTTP status. Every request also sends X-DPDP-SDK: python/<version> for SDK-version identification.

Public reads (no auth)

Auth

DSR (Data Subject Requests)

Grievances

Nomination

Consent gate

from dpdpguard import has_consent

if not has_consent(consents, "Marketing"):
    return  # don't send the campaign

has_consent(consents, purpose) is a pure helper — it doesn't call the network. consents is any sequence of objects matching the ConsentRecord protocol (a purpose: str and a withdrawn_at: int | None); combine it with list_dsr_requests()-style reads or your own cached consent state.

Audit-hash verification

A holder of the platform's audit-hash HMAC secret can independently verify a consent audit trail row's auditHash:

from dpdpguard import AuditHashInput, compute_audit_hash

input = AuditHashInput(
    organization_id="org_abc123",
    notice_id="notice_v1",
    notice_version=1,
    purpose="Newsletter",
    data_types=["email"],
    given_at=1700000000000,
    source="direct",
)
assert compute_audit_hash(input, secret) == stored_audit_hash

canonicalize_audit_event(input) is also exported if you need the canonical string form (e.g. for debugging a mismatch) without hashing it.

Webhook signature verification

from dpdpguard import verify_webhook_signature

ok = verify_webhook_signature(secret, raw_body, request.headers.get("x-dpdp-signature"))

raw_body must be the exact bytes/string received on the wire, before any json.loads — HMACs are sensitive to whitespace/key-order, so re-serializing a parsed object and hashing that will not match. Comparison is constant-time.

Error handling

from dpdpguard import ApiErrorCode, DpdpGuardApiError, ERROR_CATALOG

try:
    client.get_notice("missing")
except DpdpGuardApiError as err:
    print(err.code, err.status)  # e.g. "NOT_FOUND", 404

ERROR_CATALOG is a list of ErrorCatalogEntry ({"code": ..., "description": ...}) describing every code in ApiErrorCode.

Typed models

dpdpguard.models exposes TypedDicts for every wire shape the client returns — OrgSummary, Notice, NoticeItemized, DsrRequest, Grievance, Nomination, BannerConfig, BrokerTokenResult, LinkAnonymousConsentResult, NeedsReconsentEntry, NoticesResponse, DsrRequestsResponse, GrievancesResponse, and the DsrType alias. Field names are the exact JSON keys (camelCase) the API sends/expects — no snake_case remapping — so what you see is what's on the wire.

Development

pip install -e ".[dev]"
ruff check .
mypy src
pytest --cov=src --cov-report=term-missing