1---
2name: gdpr-data-handling
3description: Implement GDPR-compliant data handling with consent management, data subject rights, and privacy by design. Use when building systems that process EU personal data, implementing privacy controls, or conducting GDPR compliance reviews.
4---
5 
6# GDPR Data Handling
7 
8Practical implementation guide for GDPR-compliant data processing, consent management, and privacy controls.
9 
10## When to Use This Skill
11 
12- Building systems that process EU personal data
13- Implementing consent management
14- Handling data subject requests (DSRs)
15- Conducting GDPR compliance reviews
16- Designing privacy-first architectures
17- Creating data processing agreements
18 
19## Core Concepts
20 
21### 1. Personal Data Categories
22 
23| Category               | Examples                    | Protection Level   |
24| ---------------------- | --------------------------- | ------------------ |
25| **Basic**              | Name, email, phone          | Standard           |
26| **Sensitive (Art. 9)** | Health, religion, ethnicity | Explicit consent   |
27| **Criminal (Art. 10)** | Convictions, offenses       | Official authority |
28| **Children's**         | Under 16 data               | Parental consent   |
29 
30### 2. Legal Bases for Processing
31 
32```
33Article 6 - Lawful Bases:
34├── Consent: Freely given, specific, informed
35├── Contract: Necessary for contract performance
36├── Legal Obligation: Required by law
37├── Vital Interests: Protecting someone's life
38├── Public Interest: Official functions
39└── Legitimate Interest: Balanced against rights
40```
41 
42### 3. Data Subject Rights
43 
44```
45Right to Access (Art. 15)      ─┐
46Right to Rectification (Art. 16) │
47Right to Erasure (Art. 17)       │ Must respond
48Right to Restrict (Art. 18)      │ within 1 month
49Right to Portability (Art. 20)   │
50Right to Object (Art. 21)       ─┘
51```
52 
53## Implementation Patterns
54 
55### Pattern 1: Consent Management
56 
57```javascript
58// Consent data model
59const consentSchema = {
60  userId: String,
61  consents: [
62    {
63      purpose: String, // 'marketing', 'analytics', etc.
64      granted: Boolean,
65      timestamp: Date,
66      source: String, // 'web_form', 'api', etc.
67      version: String, // Privacy policy version
68      ipAddress: String, // For proof
69      userAgent: String, // For proof
70    },
71  ],
72  auditLog: [
73    {
74      action: String, // 'granted', 'withdrawn', 'updated'
75      purpose: String,
76      timestamp: Date,
77      source: String,
78    },
79  ],
80};
81 
82// Consent service
83class ConsentManager {
84  async recordConsent(userId, purpose, granted, metadata) {
85    const consent = {
86      purpose,
87      granted,
88      timestamp: new Date(),
89      source: metadata.source,
90      version: await this.getCurrentPolicyVersion(),
91      ipAddress: metadata.ipAddress,
92      userAgent: metadata.userAgent,
93    };
94 
95    // Store consent
96    await this.db.consents.updateOne(
97      { userId },
98      {
99        $push: {
100          consents: consent,
101          auditLog: {
102            action: granted ? "granted" : "withdrawn",
103            purpose,
104            timestamp: consent.timestamp,
105            source: metadata.source,
106          },
107        },
108      },
109      { upsert: true },
110    );
111 
112    // Emit event for downstream systems
113    await this.eventBus.emit("consent.changed", {
114      userId,
115      purpose,
116      granted,
117      timestamp: consent.timestamp,
118    });
119  }
120 
121  async hasConsent(userId, purpose) {
122    const record = await this.db.consents.findOne({ userId });
123    if (!record) return false;
124 
125    const latestConsent = record.consents
126      .filter((c) => c.purpose === purpose)
127      .sort((a, b) => b.timestamp - a.timestamp)[0];
128 
129    return latestConsent?.granted === true;
130  }
131 
132  async getConsentHistory(userId) {
133    const record = await this.db.consents.findOne({ userId });
134    return record?.auditLog || [];
135  }
136}
137```
138 
139```html
140<!-- GDPR-compliant consent UI -->
141<div class="consent-banner" role="dialog" aria-labelledby="consent-title">
142  <h2 id="consent-title">Cookie Preferences</h2>
143 
144  <p>
145    We use cookies to improve your experience. Select your preferences below.
146  </p>
147 
148  <form id="consent-form">
149    <!-- Necessary - always on, no consent needed -->
150    <div class="consent-category">
151      <input type="checkbox" id="necessary" checked disabled />
152      <label for="necessary">
153        <strong>Necessary</strong>
154        <span>Required for the website to function. Cannot be disabled.</span>
155      </label>
156    </div>
157 
158    <!-- Analytics - requires consent -->
159    <div class="consent-category">
160      <input type="checkbox" id="analytics" name="analytics" />
161      <label for="analytics">
162        <strong>Analytics</strong>
163        <span>Help us understand how you use our site.</span>
164      </label>
165    </div>
166 
167    <!-- Marketing - requires consent -->
168    <div class="consent-category">
169      <input type="checkbox" id="marketing" name="marketing" />
170      <label for="marketing">
171        <strong>Marketing</strong>
172        <span>Personalized ads based on your interests.</span>
173      </label>
174    </div>
175 
176    <div class="consent-actions">
177      <button type="button" id="accept-all">Accept All</button>
178      <button type="button" id="reject-all">Reject All</button>
179      <button type="submit">Save Preferences</button>
180    </div>
181 
182    <p class="consent-links">
183      <a href="/privacy-policy">Privacy Policy</a> |
184      <a href="/cookie-policy">Cookie Policy</a>
185    </p>
186  </form>
187</div>
188```
189 
190### Pattern 2: Data Subject Access Request (DSAR)
191 
192```python
193from datetime import datetime, timedelta
194from typing import Dict, List, Optional
195import json
196 
197class DSARHandler:
198    """Handle Data Subject Access Requests."""
199 
200    RESPONSE_DEADLINE_DAYS = 30
201    EXTENSION_ALLOWED_DAYS = 60  # For complex requests
202 
203    def __init__(self, data_sources: List['DataSource']):
204        self.data_sources = data_sources
205 
206    async def submit_request(
207        self,
208        request_type: str,  # 'access', 'erasure', 'rectification', 'portability'
209        user_id: str,
210        verified: bool,
211        details: Optional[Dict] = None
212    ) -> str:
213        """Submit a new DSAR."""
214        request = {
215            'id': self.generate_request_id(),
216            'type': request_type,
217            'user_id': user_id,
218            'status': 'pending_verification' if not verified else 'processing',
219            'submitted_at': datetime.utcnow(),
220            'deadline': datetime.utcnow() + timedelta(days=self.RESPONSE_DEADLINE_DAYS),
221            'details': details or {},
222            'audit_log': [{
223                'action': 'submitted',
224                'timestamp': datetime.utcnow(),
225                'details': 'Request received'
226            }]
227        }
228 
229        await self.db.dsar_requests.insert_one(request)
230        await self.notify_dpo(request)
231 
232        return request['id']
233 
234    async def process_access_request(self, request_id: str) -> Dict:
235        """Process a data access request."""
236        request = await self.get_request(request_id)
237 
238        if request['type'] != 'access':
239            raise ValueError("Not an access request")
240 
241        # Collect data from all sources
242        user_data = {}
243        for source in self.data_sources:
244            try:
245                data = await source.get_user_data(request['user_id'])
246                user_data[source.name] = data
247            except Exception as e:
248                user_data[source.name] = {'error': str(e)}
249 
250        # Format response
251        response = {
252            'request_id': request_id,
253            'generated_at': datetime.utcnow().isoformat(),
254            'data_categories': list(user_data.keys()),
255            'data': user_data,
256            'retention_info': await self.get_retention_info(),
257            'processing_purposes': await self.get_processing_purposes(),
258            'third_party_recipients': await self.get_recipients()
259        }
260 
261        # Update request status
262        await self.update_request(request_id, 'completed', response)
263 
264        return response
265 
266    async def process_erasure_request(self, request_id: str) -> Dict:
267        """Process a right to erasure request."""
268        request = await self.get_request(request_id)
269 
270        if request['type'] != 'erasure':
271            raise ValueError("Not an erasure request")
272 
273        results = {}
274        exceptions = []
275 
276        for source in self.data_sources:
277            try:
278                # Check for legal exceptions
279                can_delete, reason = await source.can_delete(request['user_id'])
280 
281                if can_delete:
282                    await source.delete_user_data(request['user_id'])
283                    results[source.name] = 'deleted'
284                else:
285                    exceptions.append({
286                        'source': source.name,
287                        'reason': reason  # e.g., 'legal retention requirement'
288                    })
289                    results[source.name] = f'retained: {reason}'
290            except Exception as e:
291                results[source.name] = f'error: {str(e)}'
292 
293        response = {
294            'request_id': request_id,
295            'completed_at': datetime.utcnow().isoformat(),
296            'results': results,
297            'exceptions': exceptions
298        }
299 
300        await self.update_request(request_id, 'completed', response)
301 
302        return response
303 
304    async def process_portability_request(self, request_id: str) -> bytes:
305        """Generate portable data export."""
306        request = await self.get_request(request_id)
307        user_data = await self.process_access_request(request_id)
308 
309        # Convert to machine-readable format (JSON)
310        portable_data = {
311            'export_date': datetime.utcnow().isoformat(),
312            'format_version': '1.0',
313            'data': user_data['data']
314        }
315 
316        return json.dumps(portable_data, indent=2, default=str).encode()
317```
318 
319### Pattern 3: Data Retention
320 
321```python
322from datetime import datetime, timedelta
323from enum import Enum
324 
325class RetentionBasis(Enum):
326    CONSENT = "consent"
327    CONTRACT = "contract"
328    LEGAL_OBLIGATION = "legal_obligation"
329    LEGITIMATE_INTEREST = "legitimate_interest"
330 
331class DataRetentionPolicy:
332    """Define and enforce data retention policies."""
333 
334    POLICIES = {
335        'user_account': {
336            'retention_period_days': 365 * 3,  # 3 years after last activity
337            'basis': RetentionBasis.CONTRACT,
338            'trigger': 'last_activity_date',
339            'archive_before_delete': True
340        },
341        'transaction_records': {
342            'retention_period_days': 365 * 7,  # 7 years for tax
343            'basis': RetentionBasis.LEGAL_OBLIGATION,
344            'trigger': 'transaction_date',
345            'archive_before_delete': True,
346            'legal_reference': 'Tax regulations require 7 year retention'
347        },
348        'marketing_consent': {
349            'retention_period_days': 365 * 2,  # 2 years
350            'basis': RetentionBasis.CONSENT,
351            'trigger': 'consent_date',
352            'archive_before_delete': False
353        },
354        'support_tickets': {
355            'retention_period_days': 365 * 2,
356            'basis': RetentionBasis.LEGITIMATE_INTEREST,
357            'trigger': 'ticket_closed_date',
358            'archive_before_delete': True
359        },
360        'analytics_data': {
361            'retention_period_days': 365,  # 1 year
362            'basis': RetentionBasis.CONSENT,
363            'trigger': 'collection_date',
364            'archive_before_delete': False,
365            'anonymize_instead': True
366        }
367    }
368 
369    async def apply_retention_policies(self):
370        """Run retention policy enforcement."""
371        for data_type, policy in self.POLICIES.items():
372            cutoff_date = datetime.utcnow() - timedelta(
373                days=policy['retention_period_days']
374            )
375 
376            if policy.get('anonymize_instead'):
377                await self.anonymize_old_data(data_type, cutoff_date)
378            else:
379                if policy.get('archive_before_delete'):
380                    await self.archive_data(data_type, cutoff_date)
381                await self.delete_old_data(data_type, cutoff_date)
382 
383            await self.log_retention_action(data_type, cutoff_date)
384 
385    async def anonymize_old_data(self, data_type: str, before_date: datetime):
386        """Anonymize data instead of deleting."""
387        # Example: Replace identifying fields with hashes
388        if data_type == 'analytics_data':
389            await self.db.analytics.update_many(
390                {'collection_date': {'$lt': before_date}},
391                {'$set': {
392                    'user_id': None,
393                    'ip_address': None,
394                    'device_id': None,
395                    'anonymized': True,
396                    'anonymized_date': datetime.utcnow()
397                }}
398            )
399```
400 
401### Pattern 4: Privacy by Design
402 
403```python
404class PrivacyFirstDataModel:
405    """Example of privacy-by-design data model."""
406 
407    # Separate PII from behavioral data
408    user_profile_schema = {
409        'user_id': str,  # UUID, not sequential
410        'email_hash': str,  # Hashed for lookups
411        'created_at': datetime,
412        # Minimal data collection
413        'preferences': {
414            'language': str,
415            'timezone': str
416        }
417    }
418 
419    # Encrypted at rest
420    user_pii_schema = {
421        'user_id': str,
422        'email': str,  # Encrypted
423        'name': str,   # Encrypted
424        'phone': str,  # Encrypted (optional)
425        'address': dict,  # Encrypted (optional)
426        'encryption_key_id': str
427    }
428 
429    # Pseudonymized behavioral data
430    analytics_schema = {
431        'session_id': str,  # Not linked to user_id
432        'pseudonym_id': str,  # Rotating pseudonym
433        'events': list,
434        'device_category': str,  # Generalized, not specific
435        'country': str,  # Not city-level
436    }
437 
438class DataMinimization:
439    """Implement data minimization principles."""
440 
441    @staticmethod
442    def collect_only_needed(form_data: dict, purpose: str) -> dict:
443        """Filter form data to only fields needed for purpose."""
444        REQUIRED_FIELDS = {
445            'account_creation': ['email', 'password'],
446            'newsletter': ['email'],
447            'purchase': ['email', 'name', 'address', 'payment'],
448            'support': ['email', 'message']
449        }
450 
451        allowed = REQUIRED_FIELDS.get(purpose, [])
452        return {k: v for k, v in form_data.items() if k in allowed}
453 
454    @staticmethod
455    def generalize_location(ip_address: str) -> str:
456        """Generalize IP to country level only."""
457        import geoip2.database
458        reader = geoip2.database.Reader('GeoLite2-Country.mmdb')
459        try:
460            response = reader.country(ip_address)
461            return response.country.iso_code
462        except:
463            return 'UNKNOWN'
464```
465 
466### Pattern 5: Breach Notification
467 
468```python
469from datetime import datetime
470from enum import Enum
471 
472class BreachSeverity(Enum):
473    LOW = "low"
474    MEDIUM = "medium"
475    HIGH = "high"
476    CRITICAL = "critical"
477 
478class BreachNotificationHandler:
479    """Handle GDPR breach notification requirements."""
480 
481    AUTHORITY_NOTIFICATION_HOURS = 72
482    AFFECTED_NOTIFICATION_REQUIRED_SEVERITY = BreachSeverity.HIGH
483 
484    async def report_breach(
485        self,
486        description: str,
487        data_types: List[str],
488        affected_count: int,
489        severity: BreachSeverity
490    ) -> dict:
491        """Report and handle a data breach."""
492        breach = {
493            'id': self.generate_breach_id(),
494            'reported_at': datetime.utcnow(),
495            'description': description,
496            'data_types_affected': data_types,
497            'affected_individuals_count': affected_count,
498            'severity': severity.value,
499            'status': 'investigating',
500            'timeline': [{
501                'event': 'breach_reported',
502                'timestamp': datetime.utcnow(),
503                'details': description
504            }]
505        }
506 
507        await self.db.breaches.insert_one(breach)
508 
509        # Immediate notifications
510        await self.notify_dpo(breach)
511        await self.notify_security_team(breach)
512 
513        # Authority notification required within 72 hours
514        if self.requires_authority_notification(severity, data_types):
515            breach['authority_notification_deadline'] = (
516                datetime.utcnow() + timedelta(hours=self.AUTHORITY_NOTIFICATION_HOURS)
517            )
518            await self.schedule_authority_notification(breach)
519 
520        # Affected individuals notification
521        if severity.value in [BreachSeverity.HIGH.value, BreachSeverity.CRITICAL.value]:
522            await self.schedule_individual_notifications(breach)
523 
524        return breach
525 
526    def requires_authority_notification(
527        self,
528        severity: BreachSeverity,
529        data_types: List[str]
530    ) -> bool:
531        """Determine if supervisory authority must be notified."""
532        # Always notify for sensitive data
533        sensitive_types = ['health', 'financial', 'credentials', 'biometric']
534        if any(t in sensitive_types for t in data_types):
535            return True
536 
537        # Notify for medium+ severity
538        return severity in [BreachSeverity.MEDIUM, BreachSeverity.HIGH, BreachSeverity.CRITICAL]
539 
540    async def generate_authority_report(self, breach_id: str) -> dict:
541        """Generate report for supervisory authority."""
542        breach = await self.get_breach(breach_id)
543 
544        return {
545            'organization': {
546                'name': self.config.org_name,
547                'contact': self.config.dpo_contact,
548                'registration': self.config.registration_number
549            },
550            'breach': {
551                'nature': breach['description'],
552                'categories_affected': breach['data_types_affected'],
553                'approximate_number_affected': breach['affected_individuals_count'],
554                'likely_consequences': self.assess_consequences(breach),
555                'measures_taken': await self.get_remediation_measures(breach_id),
556                'measures_proposed': await self.get_proposed_measures(breach_id)
557            },
558            'timeline': breach['timeline'],
559            'submitted_at': datetime.utcnow().isoformat()
560        }
561```
562 
563## Compliance Checklist
564 
565```markdown
566## GDPR Implementation Checklist
567 
568### Legal Basis
569 
570- [ ] Documented legal basis for each processing activity
571- [ ] Consent mechanisms meet GDPR requirements
572- [ ] Legitimate interest assessments completed
573 
574### Transparency
575 
576- [ ] Privacy policy is clear and accessible
577- [ ] Processing purposes clearly stated
578- [ ] Data retention periods documented
579 
580### Data Subject Rights
581 
582- [ ] Access request process implemented
583- [ ] Erasure request process implemented
584- [ ] Portability export available
585- [ ] Rectification process available
586- [ ] Response within 30-day deadline
587 
588### Security
589 
590- [ ] Encryption at rest implemented
591- [ ] Encryption in transit (TLS)
592- [ ] Access controls in place
593- [ ] Audit logging enabled
594 
595### Breach Response
596 
597- [ ] Breach detection mechanisms
598- [ ] 72-hour notification process
599- [ ] Breach documentation system
600 
601### Documentation
602 
603- [ ] Records of processing activities (Art. 30)
604- [ ] Data protection impact assessments
605- [ ] Data processing agreements with vendors
606```
607 
608## Best Practices
609 
610### Do's
611 
612- **Minimize data collection** - Only collect what's needed
613- **Document everything** - Processing activities, legal bases
614- **Encrypt PII** - At rest and in transit
615- **Implement access controls** - Need-to-know basis
616- **Regular audits** - Verify compliance continuously
617 
618### Don'ts
619 
620- **Don't pre-check consent boxes** - Must be opt-in
621- **Don't bundle consent** - Separate purposes separately
622- **Don't retain indefinitely** - Define and enforce retention
623- **Don't ignore DSARs** - 30-day response required
624- **Don't transfer without safeguards** - SCCs or adequacy decisions
625