Coverage for src / lexigram / admin / gdpr / service.py: 0%

1"""GDPRService — orchestrates anonymization, SAR tracking, and consent.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6from typing import Any 

7 

8from lexigram.admin.gdpr.anonymizer import anonymize_record 

9from lexigram.admin.gdpr.models import ( 

10 AnonymizationRule, 

11 ConsentRecord, 

12 SARStatus, 

13 SubjectAccessRequest, 

14) 

15 

16 

17class GDPRService: 

18 """Orchestrates GDPR compliance operations. 

19 

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 """ 

24 

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 

31 

32 # ------------------------------------------------------------------ 

33 # Anonymization rules 

34 # ------------------------------------------------------------------ 

35 

36 def add_rule(self, rule: AnonymizationRule) -> None: 

37 """Register an anonymization rule for a resource type. 

38 

39 Args: 

40 rule: :class:`AnonymizationRule` to register. 

41 """ 

42 self._rules[rule.resource_type] = rule 

43 

44 def get_rule(self, resource_type: str) -> AnonymizationRule | None: 

45 """Return the rule for *resource_type*, or ``None``. 

46 

47 Args: 

48 resource_type: Resource name. 

49 """ 

50 return self._rules.get(resource_type) 

51 

52 def anonymize(self, resource_type: str, record: dict[str, Any]) -> dict[str, Any]: 

53 """Anonymize *record* using the registered rule for *resource_type*. 

54 

55 If no rule is registered, the record is returned unchanged (safe 

56 default — no data destruction without an explicit rule). 

57 

58 Args: 

59 resource_type: Resource name. 

60 record: Original record dict. 

61 

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) 

69 

70 # ------------------------------------------------------------------ 

71 # Right to erasure 

72 # ------------------------------------------------------------------ 

73 

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. 

78 

79 Does **not** persist anything — callers must apply the returned dict 

80 to their data source. 

81 

82 Args: 

83 resource_type: Resource name. 

84 record: Current record dict. 

85 

86 Returns: 

87 Anonymized / erased record dict. 

88 """ 

89 return self.anonymize(resource_type, record) 

90 

91 # ------------------------------------------------------------------ 

92 # Subject Access Requests 

93 # ------------------------------------------------------------------ 

94 

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. 

99 

100 Args: 

101 subject_id: Identifier of the data subject. 

102 subject_email: Contact email for the response. 

103 notes: Optional notes. 

104 

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 

117 

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. 

122 

123 Args: 

124 sar_id: SAR identifier. 

125 data_snapshot: Exported data for the subject. 

126 

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 

137 

138 def reject_sar(self, sar_id: str, reason: str = "") -> SubjectAccessRequest | None: 

139 """Reject an SAR (e.g. identity verification failed). 

140 

141 Args: 

142 sar_id: SAR identifier. 

143 reason: Reason for rejection (stored in notes). 

144 

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 

155 

156 def get_sar(self, sar_id: str) -> SubjectAccessRequest | None: 

157 """Return an SAR by ID. 

158 

159 Args: 

160 sar_id: SAR identifier. 

161 """ 

162 return self._sars.get(sar_id) 

163 

164 def list_sars( 

165 self, *, status: SARStatus | None = None 

166 ) -> list[SubjectAccessRequest]: 

167 """Return all SARs, optionally filtered by status. 

168 

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 

176 

177 # ------------------------------------------------------------------ 

178 # Consent 

179 # ------------------------------------------------------------------ 

180 

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. 

190 

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.). 

196 

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 

210 

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. 

213 

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 

223 

224 def consent_history(self, subject_id: str) -> list[ConsentRecord]: 

225 """Return all consent events for a subject (oldest first). 

226 

227 Args: 

228 subject_id: Data subject identifier. 

229 """ 

230 return list(self._consents.get(subject_id, [])) 

231 

232 

233__all__ = [ 

234 "GDPRService", 

235]