How to Build a GDPR-Compliant Data Deletion Pipeline

A user clicks "Delete my account." You have 30 days under GDPR Article 17 to erase all their personal data — across your database, your backups, your logs, your analytics, and every third-party system you pushed that data to. Most teams discover they don't have a deletion pipeline on the day a lawyer sends the first erasure request.

This article walks through building one that actually holds up.

What GDPR Article 17 Requires (Developer Edition)

Article 17, the "right to erasure," requires you to delete personal data when:

• The user withdraws consent
• The data is no longer necessary for its original purpose
• The user objects and you have no overriding legitimate interest

What it doesn't tell you is how to do it technically. That gap is where most engineering teams struggle.

Three practical rules:

1. Deletion must be verifiable — you need an audit log that proves erasure happened, without retaining the erased data
2. Downstream systems must be notified — if you pushed PII to your analytics platform or a third-party email tool, you need to delete it there too
3. Backups are not exempt — but GDPR allows a grace period; the common approach is pseudonymization in hot data and scheduled purges for backup chains

Mapping Your Data Graph First

Before writing a single line of deletion code, build a data inventory. You need to know every table, field, and service that holds a given user's PII.

A minimal Python script to audit your PostgreSQL schema:

import psycopg2
from dataclasses import dataclass

@dataclass
class PIIField:
    table: str
    column: str
    sample: str | None

def find_pii_fields(conn_str: str, user_id: int) -> list[PIIField]:
    """
    Scan tables that reference a user_id and return rows that
    contain email, name, phone, or IP address columns.
    """
    pii_patterns = {"email", "name", "phone", "ip_address", "address", "birth"}
    results = []

    with psycopg2.connect(conn_str) as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT table_name, column_name
                FROM information_schema.columns
                WHERE table_schema = 'public'
            """)
            schema = cur.fetchall()

    # Group columns by table
    by_table: dict[str, list[str]] = {}
    for table, col in schema:
        by_table.setdefault(table, []).append(col)

    with psycopg2.connect(conn_str) as conn:
        with conn.cursor() as cur:
            for table, cols in by_table.items():
                if "user_id" not in cols:
                    continue
                pii_cols = [c for c in cols if any(p in c.lower() for p in pii_patterns)]
                if not pii_cols:
                    continue
                cur.execute(
                    f"SELECT {', '.join(pii_cols)} FROM {table} WHERE user_id = %s LIMIT 1",
                    (user_id,)
                )
                row = cur.fetchone()
                if row:
                    for col, val in zip(pii_cols, row):
                        results.append(PIIField(table=table, column=col, sample=str(val)[:30]))

    return results

Run this against a staging database and you will find columns you forgot existed. Build it into an inventory doc and keep it up to date.

Building the Deletion Pipeline

The pipeline has four stages: validate the request, delete from hot storage, notify downstream services, record the audit event.

import hashlib
import logging
from datetime import datetime, timezone
from typing import Protocol

logger = logging.getLogger(__name__)

class StorageBackend(Protocol):
    def delete_user(self, user_id: int) -> None: ...

class DownstreamService(Protocol):
    name: str
    def delete_user(self, user_id: int) -> None: ...

class AuditLog(Protocol):
    def record(self, user_id_hash: str, timestamp: str, services: list[str]) -> None: ...


def run_erasure(
    user_id: int,
    backends: list[StorageBackend],
    downstream: list[DownstreamService],
    audit: AuditLog,
) -> dict:
    errors: list[str] = []
    deleted: list[str] = []

    # Delete from primary storage
    for backend in backends:
        try:
            backend.delete_user(user_id)
            deleted.append(backend.__class__.__name__)
        except Exception as e:
            logger.error("Deletion failed in %s: %s", backend.__class__.__name__, e)
            errors.append(backend.__class__.__name__)

    # Notify downstream systems
    for svc in downstream:
        try:
            svc.delete_user(user_id)
            deleted.append(svc.name)
        except Exception as e:
            logger.error("Downstream deletion failed for %s: %s", svc.name, e)
            errors.append(svc.name)

    # Audit: store a hash of the user_id, not the ID itself
    user_id_hash = hashlib.sha256(f"gdpr-erasure:{user_id}".encode()).hexdigest()
    ts = datetime.now(timezone.utc).isoformat()
    audit.record(user_id_hash=user_id_hash, timestamp=ts, services=deleted)

    return {"deleted": deleted, "errors": errors, "timestamp": ts}

A few design decisions worth explaining:

Hash the user ID in the audit log. You need to prove deletion happened, but storing user_id=42 in your audit log means you are still holding personal data (assuming user 42 can be identified). A keyed hash breaks that link while still being verifiable if you have the original ID.

Use Protocols, not inheritance. Different storage backends (Postgres, Redis, S3, your email provider's API) each have their own delete methods. A structural Protocol keeps the pipeline decoupled from any specific implementation.

Don't swallow errors silently. A failed downstream deletion is a GDPR violation in progress. Log it, alert on it, and have a retry mechanism with dead-letter tracking.

Handling Backups, Logs, and Third-Party Data

The three hardest parts:

Backups: You can't rewrite a nightly dump retroactively at scale. The practical approach is pseudonymization — replace PII with a consistent token before it hits cold storage, and store the mapping separately. When a deletion request arrives, delete the mapping. The backup becomes unresolvable.

Logs: Application logs often contain emails in request paths, user agents, or error messages. Run a structured logging policy: log user IDs only, never raw PII. For existing logs, a scrubbing job can find and redact entries, but this is expensive at scale. Use structured log fields so scrubbing is surgical rather than regex-based.

Third-party services: Maintain a service inventory (Mailchimp, Segment, Mixpanel, Intercom, Stripe — each has a data deletion API). Your deletion pipeline should call each one. Some have webhooks you can register; others require explicit API calls. A GDPR compliance checklist can help you inventory these surfaces systematically before an erasure request arrives.

Testing the Pipeline

GDPR deletion is not a feature you test only in production. Write integration tests that:

• Create a user with known PII across all storage backends
• Trigger the erasure pipeline
• Assert that every field returns NULL or is absent
• Assert that the downstream service mocks received the deletion call
• Assert that the audit log contains a hashed reference with the correct timestamp

A passing unit test against a mock database is worth very little here. Test against the real schema.

The Takeaway

The technical challenge of GDPR erasure is not deleting a row — it is knowing every place a user's data lives and having a pipeline that reliably reaches all of them. Start with a data inventory script. Build the pipeline to be async and retry-safe. Never store the raw user ID in your audit log. Treat downstream service failures as incidents, not warnings.

The 30-day window sounds comfortable until you are debugging a half-broken third-party integration at day 28.

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.