Build a Low‑Cost Automated GDPR Stack in 30 Days

# rpa_service.py
import uuid
from datetime import datetime
from typing import List

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite:///./rpa.db"
engine = sa.create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

app = FastAPI(title="GDPR Real‑Time RPA API", version="1.0.0")

class Activity(Base):
    __tablename__ = "activities"
    id = sa.Column(sa.String, primary_key=True, default=lambda: str(uuid.uuid4()))
    purpose = sa.Column(sa.String, nullable=False)
    data_category = sa.Column(sa.String, nullable=False)
    legal_basis = sa.Column(sa.String, nullable=False)
    retention_days = sa.Column(sa.Integer, nullable=False)
    created_at = sa.Column(sa.DateTime, default=datetime.utcnow)

Base.metadata.create_all(bind=engine)

# Pydantic schema for API responses
class ActivityOut(BaseModel):
    id: str
    purpose: str
    data_category: str
    legal_basis: str
    retention_days: int
    created_at: datetime

    class Config:
        orm_mode = True

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# ---- Endpoints -------------------------------------------------

@app.post("/activities", response_model=ActivityOut)
def add_activity(activity: ActivityOut, db: sa.orm.Session = Depends(get_db)):
    db_activity = Activity(**activity.dict())
    db.add(db_activity)
    db.commit()
    db.refresh(db_activity)
    return db_activity

@app.get("/rpa", response_model=List[ActivityOut])
def read_rpa(db: sa.orm.Session = Depends(get_db)):
    """Return the full, real‑time Record of Processing Activities."""
    return db.query(Activity).order_by(Activity.created_at.desc()).all()

# 1️⃣ Install dependencies
pip install fastapi uvicorn sqlalchemy pydantic

# 2️⃣ Start the service
uvicorn rpa_service:app --host 0.0.0.0 --port 8000

# Add a sample activity
curl -X POST http://localhost:8000/activities \
  -H "Content-Type: application/json" \
  -d '{"purpose":"email marketing","data_category":"email address","legal_basis":"consent","retention_days":365}'

# Retrieve the RPA (what auditors will see)
curl http://localhost:8000/rpa

import csv
from fastapi.responses import StreamingResponse
from io import StringIO

@app.get("/dsar/{subject_id}", response_class=StreamingResponse)
def dsar(subject_id: str, db: sa.orm.Session = Depends(get_db)):
    """Export all records linked to a specific data subject."""
    query = db.query(Activity).filter(Activity.id == subject_id)  # adapt to your schema
    if not query.first():
        raise HTTPException(status_code=404, detail="Subject not found")

    def iter_csv():
        output = StringIO()
        writer = csv.writer(output)
        writer.writerow(["id","purpose","data_category","legal_basis","retention_days","created_at"])
        for act in query:
            writer.writerow([act.id, act.purpose, act.data_category,
                             act.legal_basis, act.retention_days,
                             act.created_at.isoformat()])
            yield output.getvalue()
            output.seek(0)
            output.truncate(0)

    return StreamingResponse(iter_csv(),
                             media_type="text/csv",
                             headers={"Content-Disposition": f"attachment; filename={subject_id}_dsar.csv"})

markdown
# Privacy Policy

**Effective Date:** {{DATE}}

## 1. Who