Coverage for src/lexigram/admin/gdpr/service.py: 0%
65 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""GDPRService — orchestrates anonymization, SAR tracking, and consent."""
3from __future__ import annotations
5from datetime import UTC, datetime
6from typing import Any
8from lexigram.admin.gdpr.anonymizer import anonymize_record
9from lexigram.admin.gdpr.models import (
10 AnonymizationRule,
11 ConsentRecord,
12 SARStatus,
13 SubjectAccessRequest,
14)
17class GDPRService:
18 """Orchestrates GDPR compliance operations.
20 Manages anonymization rules, Subject Access Requests, and consent records.
21 Backed by in-process stores by default — wire a DB-backed store for
22 production via subclassing or dependency injection.
23 """
25 def __init__(self) -> None:
26 self._rules: dict[str, AnonymizationRule] = {}
27 self._sars: dict[str, SubjectAccessRequest] = {}
28 self._consents: dict[str, list[ConsentRecord]] = {}
29 self._sar_counter = 0
30 self._consent_counter = 0
32 # ------------------------------------------------------------------
33 # Anonymization rules
34 # ------------------------------------------------------------------
36 def add_rule(self, rule: AnonymizationRule) -> None:
37 """Register an anonymization rule for a resource type.
39 Args:
40 rule: :class:`AnonymizationRule` to register.
41 """
42 self._rules[rule.resource_type] = rule
44 def get_rule(self, resource_type: str) -> AnonymizationRule | None:
45 """Return the rule for *resource_type*, or ``None``.
47 Args:
48 resource_type: Resource name.
49 """
50 return self._rules.get(resource_type)
52 def anonymize(self, resource_type: str, record: dict[str, Any]) -> dict[str, Any]:
53 """Anonymize *record* using the registered rule for *resource_type*.
55 If no rule is registered, the record is returned unchanged (safe
56 default — no data destruction without an explicit rule).
58 Args:
59 resource_type: Resource name.
60 record: Original record dict.
62 Returns:
63 Anonymized copy of the record.
64 """
65 rule = self._rules.get(resource_type)
66 if rule is None:
67 return dict(record)
68 return anonymize_record(record, rule)
70 # ------------------------------------------------------------------
71 # Right to erasure
72 # ------------------------------------------------------------------
74 def erasure_plan(
75 self, resource_type: str, record: dict[str, Any]
76 ) -> dict[str, Any]:
77 """Return what the record would look like after right-to-erasure.
79 Does **not** persist anything — callers must apply the returned dict
80 to their data source.
82 Args:
83 resource_type: Resource name.
84 record: Current record dict.
86 Returns:
87 Anonymized / erased record dict.
88 """
89 return self.anonymize(resource_type, record)
91 # ------------------------------------------------------------------
92 # Subject Access Requests
93 # ------------------------------------------------------------------
95 def submit_sar(
96 self, subject_id: str, subject_email: str, *, notes: str = ""
97 ) -> SubjectAccessRequest:
98 """Create and track a new Subject Access Request.
100 Args:
101 subject_id: Identifier of the data subject.
102 subject_email: Contact email for the response.
103 notes: Optional notes.
105 Returns:
106 Newly created :class:`SubjectAccessRequest`.
107 """
108 self._sar_counter += 1
109 sar = SubjectAccessRequest(
110 sar_id=f"sar-{self._sar_counter:06d}",
111 subject_id=subject_id,
112 subject_email=subject_email,
113 notes=notes,
114 )
115 self._sars[sar.sar_id] = sar
116 return sar
118 def complete_sar(
119 self, sar_id: str, data_snapshot: dict[str, Any]
120 ) -> SubjectAccessRequest | None:
121 """Mark an SAR as completed and attach the exported data.
123 Args:
124 sar_id: SAR identifier.
125 data_snapshot: Exported data for the subject.
127 Returns:
128 Updated :class:`SubjectAccessRequest` or ``None`` if not found.
129 """
130 sar = self._sars.get(sar_id)
131 if sar is None:
132 return None
133 sar.status = SARStatus.COMPLETED
134 sar.completed_at = datetime.now(UTC)
135 sar.data_snapshot = data_snapshot
136 return sar
138 def reject_sar(self, sar_id: str, reason: str = "") -> SubjectAccessRequest | None:
139 """Reject an SAR (e.g. identity verification failed).
141 Args:
142 sar_id: SAR identifier.
143 reason: Reason for rejection (stored in notes).
145 Returns:
146 Updated :class:`SubjectAccessRequest` or ``None`` if not found.
147 """
148 sar = self._sars.get(sar_id)
149 if sar is None:
150 return None
151 sar.status = SARStatus.REJECTED
152 if reason:
153 sar.notes = f"Rejected: {reason}"
154 return sar
156 def get_sar(self, sar_id: str) -> SubjectAccessRequest | None:
157 """Return an SAR by ID.
159 Args:
160 sar_id: SAR identifier.
161 """
162 return self._sars.get(sar_id)
164 def list_sars(
165 self, *, status: SARStatus | None = None
166 ) -> list[SubjectAccessRequest]:
167 """Return all SARs, optionally filtered by status.
169 Args:
170 status: When provided, only return SARs with this status.
171 """
172 sars = list(self._sars.values())
173 if status:
174 return [s for s in sars if s.status == status]
175 return sars
177 # ------------------------------------------------------------------
178 # Consent
179 # ------------------------------------------------------------------
181 def record_consent(
182 self,
183 subject_id: str,
184 purpose: str,
185 *,
186 granted: bool,
187 metadata: dict[str, Any] | None = None,
188 ) -> ConsentRecord:
189 """Record a consent grant or withdrawal.
191 Args:
192 subject_id: Data subject identifier.
193 purpose: Consent purpose slug (e.g. ``"marketing_email"``).
194 granted: ``True`` = granted, ``False`` = withdrawn.
195 metadata: Optional context (IP, source, etc.).
197 Returns:
198 Newly created :class:`ConsentRecord`.
199 """
200 self._consent_counter += 1
201 record = ConsentRecord(
202 consent_id=f"con-{self._consent_counter:06d}",
203 subject_id=subject_id,
204 purpose=purpose,
205 granted=granted,
206 metadata=metadata or {},
207 )
208 self._consents.setdefault(subject_id, []).append(record)
209 return record
211 def has_consent(self, subject_id: str, purpose: str) -> bool:
212 """Return ``True`` if the subject's most recent consent for *purpose* is granted.
214 Args:
215 subject_id: Data subject identifier.
216 purpose: Consent purpose slug.
217 """
218 events = self._consents.get(subject_id, [])
219 for event in reversed(events):
220 if event.purpose == purpose:
221 return event.granted
222 return False
224 def consent_history(self, subject_id: str) -> list[ConsentRecord]:
225 """Return all consent events for a subject (oldest first).
227 Args:
228 subject_id: Data subject identifier.
229 """
230 return list(self._consents.get(subject_id, []))
233__all__ = [
234 "GDPRService",
235]