DPDPStack logodpdpstack

FastAPI / SQLAlchemy integration

The SQLAlchemy adapter (pip install "dpdpstack-python-sdk[sqlalchemy]") runs the engine against your own mapped models, with a database-backed hash-chained audit store — still zero-egress, since the mutation happens on your instance in your database. It mirrors the Django adapter; the @pii declaration is shared between the two.

Map the audit store

You own the Base/registry, so map the audit entry once from the provided mixin:

from sqlalchemy.orm import DeclarativeBase
from dpdpstack.contrib.sqlalchemy.models import DpdpAuditEntryMixin

class Base(DeclarativeBase):
    ...

class DpdpAuditEntry(Base, DpdpAuditEntryMixin):
    __tablename__ = "dpdp_audit_entries"

# Base.metadata.create_all(engine)  # or an Alembic migration

Declare PII once with @pii

The @pii(...) decorator records each model's PII fields and how to anonymize them, so you don't pass pii_fields= on every call.

from dpdpstack import null, redact
from dpdpstack.contrib.sqlalchemy.service import pii

@pii(name=null, email=null, phone=redact(keep_last=4))
class User(Base):
    __tablename__ = "users"
    ...

Erase per a retention policy

erase_instance runs the engine against your Session. You commit the transaction.

from dpdpstack import RetentionPolicy, Action, rbi_kyc
from dpdpstack.contrib.sqlalchemy.service import erase_instance

# Hard delete + audit
erase_instance(session, user, audit_model=DpdpAuditEntry, subject=user.external_ref,
               policy=RetentionPolicy(purpose="marketing", action=Action.DELETE))

# Anonymize PII, keep the (regulated) row — uses the @pii declaration above
erase_instance(session, user, audit_model=DpdpAuditEntry, subject=user.external_ref,
               policy=RetentionPolicy(purpose="profile", action=Action.ANONYMIZE))

# KYC withdrawal → deferred under RBI hold, nothing deleted, basis recorded
erase_instance(session, user, audit_model=DpdpAuditEntry, subject=user.external_ref,
               policy=rbi_kyc("kyc"))

session.commit()

FastAPI sketch

@app.delete("/users/{user_id}")
def forget_user(user_id: int, session: Session = Depends(get_session)):
    user = session.get(User, user_id)
    res = erase_instance(
        session, user, audit_model=DpdpAuditEntry, subject=user.external_ref,
        policy=RetentionPolicy(purpose="profile", action=Action.ANONYMIZE),
    )
    session.commit()
    return {"status": res.status, "action": res.action}

The audit chain verifies with audit_log(session, DpdpAuditEntry).verify(), and you can issue a Certificate of Erasure from it exactly as in the framework-agnostic quickstart.