Not legal advice
dpdpkit is software that helps you implement obligations under India's Digital Personal Data Protection Act, 2023 and the DPDP Rules, 2025. It does not provide legal advice and does not guarantee compliance.
FastAPI adapter¶
DPDPKit — the object a FastAPI developer creates.
DPDPKit(*, principal_resolver: PrincipalResolver, admin_guard: AdminGuard | None = None, db_url: str | None = None, engine: Any = None, repository: Repository | None = None, config: str | Path | Mapping[str, Any] | None = None, policy: str | Policy | None = None, overlays: Iterable[str] = (), registry: Registry | None = None, notifier: Notifier | None = None, contact: Mapping[str, str] | None = None, contact_resolver: ContactResolver | None = None, identity_verifier: IdentityVerifier | None = None, signing_key: bytes | str | None = None, tenant_id: str = 'default', webhooks: Iterable[WebhookEndpoint] | None = None, migrate_on_startup: bool = False, create_tables: bool = False, **kit_kwargs: Any)
Wire dpdpkit into a FastAPI app.
webhooks are delivered through a database outbox (see dpdpkit.fastapi.webhooks); when
omitted they are read from the webhooks: list in the config, with secrets from secret_env.
principal_resolver(request) returns a stable, non-personal reference for the signed-in user
(for example the user's primary key), or None. admin_guard(request) returns a role
(viewer/handler/admin), True for admin, or None/False for no access.
instance-attribute
¶
kit = kit_from_config(self.config, repository, registry=registry, contact=contact, **common)
async
¶
require_role(request: Request, needed: Role) -> str
Check the caller's admin role; returns an actor label for the audit trail.
staticmethod
¶
request_metadata(request: Request) -> dict[str, Any]
Evidence stored with consent events. No IP address or personal data by default.
async
¶
run(fn: Callable[P, T], *args: args, **kwargs: kwargs) -> T
Call a kit method from async def code without blocking the event loop::
allowed = await dpdp.run(dpdp.kit.consent.check, user_id, "marketing")
bind(session: Session) -> AbstractContextManager[Session]
Run kit calls inside the app's sync Session; see SqlAlchemyRepository.bind.
async
¶
run_in_session(session: AsyncSession, fn: Callable[P, T], *args: args, **kwargs: kwargs) -> T
Call a kit method inside the app's AsyncSession, so its writes commit or roll back with the
app's own work::
async with AsyncSession(engine) as s, s.begin():
s.add(Order(...))
await dpdp.run_in_session(s, dpdp.kit.consent.grant, user_id, "marketing")
dpdpkit never commits the session. The call runs through AsyncSession.run_sync on the event
loop, so keep it to database work; event listeners see the event before the app commits.
install(app: FastAPI, *, prefix: str = '/dpdp', admin_path: str | None = None, track_activity: bool = True) -> None
Include the router, error handlers, activity middleware and the admin dashboard.
mount_admin(app: FastAPI, path: str = '/dpdp/admin/ui') -> None
Serve the prebuilt admin dashboard bundle. Until it ships (v0.4) a placeholder page is served.
start_scheduler(*, retention_minutes: int = 15, verify_hour: int = 2, root_export_hour: int | None = 3, root_sinks: Iterable[RootSink] = (), webhook_seconds: int = 30) -> Any
Run retention every retention_minutes, ledger verification daily at verify_hour and
a root-hash export daily at root_export_hour UTC (None disables it) (APScheduler).
The export writes to each of root_sinks and publishes ledger.root_exported; subscribe a
webhook to that event to hand a copy to a third party. Due webhook deliveries are sent every
webhook_seconds when any endpoint is configured.
Export the ledger root hash. A failing sink is logged and the job reports None.
ActivityMiddleware(app: Any, dpdpkit: DPDPKit, min_interval_seconds: float = 300)
Bases: BaseHTTPMiddleware
Records principal activity (which restarts inactivity clocks and cancels warned erasures). Throttled per principal so busy users do not write on every request.
require_consent(purpose: str) -> Callable[[Request], Awaitable[None]]
FastAPI dependency: 403 consent_required unless the caller has consented to purpose.
SqlAlchemyRepository: dpdpkit-core's Repository on SQLAlchemy 2.0 (Postgres, MySQL, SQLite).
SqlAlchemyRepository(engine: Engine | str | Any)
engine is a URL, an Engine, or the app's AsyncEngine (dpdpkit then opens its own
small sync pool to the same database).
create_all() -> None
Create tables directly (tests and demos). Production installs use dpdpkit migrate.
Run kit calls inside the app's own Session, so they commit or roll back with its work::
with Session(engine) as s, s.begin(), dpdp.repository.bind(s):
s.add(Order(...))
dpdp.kit.consent.grant(user_id, "marketing")
dpdpkit never commits a bound session; each write runs in a savepoint. For an AsyncSession use
await dpdp.run_in_session(session, fn, ...).
latest_consent_event(tenant_id: str, principal: str, purpose: str) -> ConsentEvent | None
list_consent_events(tenant_id: str, *, principal: str | None = None, after_seq: int = 0, limit: int | None = None) -> list[ConsentEvent]
list_audit(tenant_id: str, *, subject_id: str | None = None, after_seq: int = 0, limit: int | None = None) -> list[AuditEntry]
list_requests(tenant_id: str, *, principal: str | None = None, statuses: Collection[RequestStatus] | None = None) -> list[RightsRequest]
list_schedules(tenant_id: str, *, principal: str | None = None, statuses: Collection[ScheduleStatus] | None = None) -> list[ErasureSchedule]
make_engine(url: str | URL, **kwargs: Any) -> Engine
Create an engine with settings dpdpkit relies on (SQLite savepoints, shared in-memory DB).
Async URLs (postgresql+asyncpg://, sqlite+aiosqlite:// ...) are accepted and connect with the
matching sync driver, which must be installed (dpdpkit-fastapi[postgres] or [mysql]).
Durable outbound webhooks: a database outbox with retries, backoff and a dead-letter state.
Publishing an event only inserts one row per matching endpoint, so request handlers never wait on
the network. WebhookOutbox.deliver_due (run by the scheduler) sends due rows, reschedules
failures with exponential backoff, and marks a row dead after max_attempts. Dead rows stay
until someone redelivers them; nothing is dropped silently.
Receivers verify X-DPDPKit-Signature with dpdpkit.events.verify_signature and should treat
the event id as an idempotency key: a delivery can arrive more than once.
WebhookOutbox(repository: SqlAlchemyRepository, endpoints: Iterable[WebhookEndpoint], *, clock: Callable[[], datetime], sender: Sender = urllib_sender, max_attempts: int = 10, base_delay: timedelta = timedelta(seconds=30), max_delay: timedelta = timedelta(hours=2), lease: timedelta = timedelta(minutes=5), batch_size: int = 100)
max_attempts sends spaced base_delay * 2**n apart (capped at max_delay): with the
defaults, 10 attempts over roughly 4 hours. lease stops two workers sending the same row.
enqueue(event: Event) -> int
Store one pending delivery per endpoint that wants this event. Returns how many.
backoff(attempts: int) -> timedelta
Delay before the next try, after attempts failed sends.
deliver_due(now: datetime | None = None) -> DeliveryRun
Send every due delivery once. Safe to run from several processes at the same time.
redeliver(delivery_id: str | None = None) -> int
Move one dead delivery (or all of them) back to pending with a fresh attempt budget.
Bases: BaseModel
dataclass
¶
DeliveryRun(delivered: int = 0, retrying: int = 0, dead: int = 0)
endpoints_from_config(items: Iterable[Mapping[str, Any]]) -> list[WebhookEndpoint]
Build endpoints from the webhooks: list in dpdpkit.yaml. Secrets come from the environment
(secret_env) so they never sit in the config file.
ConsentRequiredHTTP(purpose: str)
Bases: HTTPException
Raised by require_consent. Rendered as the contract's Error body once the kit is installed
on the app; otherwise FastAPI renders it as {"detail": {...}} with the same code.