Coverage for src/lexigram/admin/gdpr/anonymizer.py: 93%
30 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Anonymization helpers — strategy application and record anonymization."""
3from __future__ import annotations
5import hashlib
6from typing import Any
8from lexigram.admin.gdpr.models import AnonymizationRule, AnonymizationStrategy
11def _apply_strategy(value: Any, strategy: str | AnonymizationStrategy) -> Any:
12 """Apply a single anonymization strategy to a value.
14 Args:
15 value: Original field value.
16 strategy: Strategy to apply.
18 Returns:
19 Anonymized value.
20 """
21 s = AnonymizationStrategy(strategy) if isinstance(strategy, str) else strategy
22 match s:
23 case AnonymizationStrategy.HASH:
24 return hashlib.sha256(str(value).encode()).hexdigest()
25 case AnonymizationStrategy.CLEAR:
26 return ""
27 case AnonymizationStrategy.REDACT:
28 return "[REDACTED]"
29 case AnonymizationStrategy.ZERO:
30 return 0
31 case AnonymizationStrategy.NULLIFY:
32 return None
33 case AnonymizationStrategy.FAKE_EMAIL:
34 return "anonymized@example.com"
35 case _:
36 return value
39def anonymize_record(
40 record: dict[str, Any],
41 rule: AnonymizationRule,
42) -> dict[str, Any]:
43 """Return a copy of *record* with PII fields anonymized per *rule*.
45 Fields not listed in the rule are passed through unchanged.
47 Args:
48 record: Original record dict.
49 rule: Anonymization rule defining per-field strategies.
51 Returns:
52 New dict with anonymized values.
53 """
54 result = dict(record)
55 for fname, strategy in rule.fields.items():
56 if fname in result:
57 if rule.preserve_id and fname == "id":
58 continue
59 result[fname] = _apply_strategy(result[fname], strategy)
60 return result
63__all__ = [
64 "anonymize_record",
65]