1156 lines
40 KiB
Python
1156 lines
40 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_addresses.backend.db.models import (
|
|
AddressBook,
|
|
AddressImportProfile,
|
|
AddressImportRun,
|
|
AddressList,
|
|
AddressListEntry,
|
|
AddressSyncConflict,
|
|
AddressSyncSource,
|
|
AddressSyncTombstone,
|
|
Contact,
|
|
ContactChannelRule,
|
|
ContactEmail,
|
|
ContactFieldProvenance,
|
|
ContactMergeRecord,
|
|
ContactPhone,
|
|
ContactPointQualityDecision,
|
|
ContactPointSnapshot,
|
|
ContactPostalAddress,
|
|
ContactRedirect,
|
|
)
|
|
from govoplan_core.core.dsar import (
|
|
DsarErasureActionRef,
|
|
DsarExecutionResultRef,
|
|
DsarRecordRef,
|
|
DsarSubjectRef,
|
|
dsar_capability_name,
|
|
)
|
|
|
|
|
|
ADDRESSES_DSAR_CAPABILITY = dsar_capability_name("addresses")
|
|
_MAX_RECORDS = 5_000
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _SubjectSelectors:
|
|
account_id: str | None
|
|
email: str | None
|
|
references: Mapping[str, str]
|
|
|
|
|
|
class AddressesDsarProvider:
|
|
provider_id = "addresses"
|
|
module_id = "addresses"
|
|
|
|
def search_subject(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
) -> Sequence[DsarRecordRef]:
|
|
db = _session(session)
|
|
selectors = _subject_selectors(subject)
|
|
if selectors is None:
|
|
return ()
|
|
|
|
contact_ids = _subject_contact_ids(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
selectors=selectors,
|
|
)
|
|
if contact_ids is None:
|
|
return ()
|
|
|
|
records: list[DsarRecordRef] = []
|
|
seen: set[tuple[str, str]] = set()
|
|
|
|
def append(record: DsarRecordRef) -> None:
|
|
key = (record.resource_type, record.resource_id)
|
|
if key in seen:
|
|
return
|
|
if len(records) >= _MAX_RECORDS:
|
|
raise ValueError(
|
|
"Addresses DSAR result limit exceeded; narrow the subject selectors."
|
|
)
|
|
seen.add(key)
|
|
records.append(record)
|
|
|
|
contacts = _rows_by_ids(
|
|
db,
|
|
Contact,
|
|
tenant_id=tenant_id,
|
|
ids=contact_ids,
|
|
)
|
|
for contact in contacts:
|
|
match_fields = []
|
|
if selectors.references.get("contact") == contact.id:
|
|
match_fields.append("reference")
|
|
if selectors.email and any(
|
|
_normalized_email(item.email) == selectors.email
|
|
or _normalized_email(item.normalized_email) == selectors.email
|
|
for item in contact.emails
|
|
):
|
|
match_fields.append("email")
|
|
append(
|
|
_record(
|
|
"addresses_contact",
|
|
contact.id,
|
|
"contact_identity",
|
|
f"Address contact {contact.display_name}",
|
|
{
|
|
"match_fields": match_fields,
|
|
"display_name": _bounded_text(contact.display_name, 255),
|
|
"given_name": _bounded_text(contact.given_name, 255),
|
|
"family_name": _bounded_text(contact.family_name, 255),
|
|
"organization": _bounded_text(contact.organization, 255),
|
|
"role_title": _bounded_text(contact.role_title, 255),
|
|
"note": _bounded_text(contact.note, 2_000),
|
|
"tags": [str(value)[:120] for value in contact.tags[:100]],
|
|
"source_kind": contact.source_kind,
|
|
"source_revision": _bounded_text(
|
|
contact.source_revision,
|
|
255,
|
|
),
|
|
"deleted_at": _iso(contact.deleted_at),
|
|
},
|
|
observed_at=contact.updated_at,
|
|
)
|
|
)
|
|
|
|
for email in _contact_children(
|
|
db,
|
|
ContactEmail,
|
|
tenant_id=tenant_id,
|
|
contact_ids=contact_ids,
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_contact_email",
|
|
email.id,
|
|
"electronic_contact_point",
|
|
"Address email contact point",
|
|
{
|
|
"contact_id": email.contact_id,
|
|
"match_fields": _contact_point_match_fields(
|
|
email,
|
|
selectors=selectors,
|
|
reference_kind="contact_email",
|
|
),
|
|
"label": _bounded_text(email.label, 80),
|
|
"email": _bounded_text(email.email, 320),
|
|
"original_email": _bounded_text(email.original_email, 320),
|
|
"normalized_email": _bounded_text(
|
|
email.normalized_email,
|
|
320,
|
|
),
|
|
"is_primary": email.is_primary,
|
|
"order_index": email.order_index,
|
|
},
|
|
observed_at=email.updated_at,
|
|
)
|
|
)
|
|
|
|
for phone in _contact_children(
|
|
db,
|
|
ContactPhone,
|
|
tenant_id=tenant_id,
|
|
contact_ids=contact_ids,
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_contact_phone",
|
|
phone.id,
|
|
"telephone_contact_point",
|
|
"Address telephone contact point",
|
|
{
|
|
"contact_id": phone.contact_id,
|
|
"match_fields": _contact_point_match_fields(
|
|
phone,
|
|
selectors=selectors,
|
|
reference_kind="contact_phone",
|
|
),
|
|
"label": _bounded_text(phone.label, 80),
|
|
"phone": _bounded_text(phone.phone, 100),
|
|
"original_phone": _bounded_text(phone.original_phone, 100),
|
|
"normalized_phone": _bounded_text(
|
|
phone.normalized_phone,
|
|
100,
|
|
),
|
|
"is_primary": phone.is_primary,
|
|
"order_index": phone.order_index,
|
|
},
|
|
observed_at=phone.updated_at,
|
|
)
|
|
)
|
|
|
|
for postal in _contact_children(
|
|
db,
|
|
ContactPostalAddress,
|
|
tenant_id=tenant_id,
|
|
contact_ids=contact_ids,
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_contact_postal_address",
|
|
postal.id,
|
|
"postal_contact_point",
|
|
"Address postal contact point",
|
|
{
|
|
"contact_id": postal.contact_id,
|
|
"match_fields": _contact_point_match_fields(
|
|
postal,
|
|
selectors=selectors,
|
|
reference_kind="contact_postal_address",
|
|
),
|
|
"label": _bounded_text(postal.label, 80),
|
|
"street": _bounded_text(postal.street, 500),
|
|
"postal_code": _bounded_text(postal.postal_code, 40),
|
|
"locality": _bounded_text(postal.locality, 255),
|
|
"region": _bounded_text(postal.region, 255),
|
|
"country": _bounded_text(postal.country, 255),
|
|
"is_primary": postal.is_primary,
|
|
"order_index": postal.order_index,
|
|
},
|
|
observed_at=postal.updated_at,
|
|
)
|
|
)
|
|
|
|
for entry in _list_entries(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
contact_ids=contact_ids,
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_list_membership",
|
|
entry.id,
|
|
"address_list_use",
|
|
"Address-list membership",
|
|
{
|
|
"contact_id": entry.contact_id,
|
|
"address_list_id": entry.address_list_id,
|
|
"address_list_name": _bounded_text(
|
|
entry.address_list.name,
|
|
255,
|
|
),
|
|
"target_kind": entry.target_kind,
|
|
"contact_email_id": entry.contact_email_id,
|
|
"contact_postal_address_id": (entry.contact_postal_address_id),
|
|
"label": _bounded_text(entry.label, 255),
|
|
"order_index": entry.order_index,
|
|
},
|
|
observed_at=entry.updated_at,
|
|
)
|
|
)
|
|
|
|
for rule in _related_or_attributed(
|
|
db,
|
|
ContactChannelRule,
|
|
tenant_id=tenant_id,
|
|
related_field=ContactChannelRule.contact_id,
|
|
related_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
actor_fields=(ContactChannelRule.created_by_account_id,),
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_channel_rule",
|
|
rule.id,
|
|
"communication_governance_evidence",
|
|
"Address communication-governance decision",
|
|
{
|
|
"match_fields": _related_actor_match_fields(
|
|
rule,
|
|
contact_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
actor_fields=("created_by_account_id",),
|
|
),
|
|
"contact_id": (
|
|
rule.contact_id if rule.contact_id in contact_ids else None
|
|
),
|
|
"channel": rule.channel,
|
|
"purpose": _bounded_text(rule.purpose, 120),
|
|
"contact_point_id": (
|
|
rule.contact_point_id
|
|
if rule.contact_id in contact_ids
|
|
else None
|
|
),
|
|
"decision": rule.decision,
|
|
"legal_basis": _bounded_text(rule.legal_basis, 255),
|
|
"reason": _bounded_text(rule.reason, 2_000),
|
|
"preference_rank": rule.preference_rank,
|
|
"locale": _bounded_text(rule.locale, 20),
|
|
"effective_from": _iso(rule.effective_from),
|
|
"effective_until": _iso(rule.effective_until),
|
|
},
|
|
observed_at=rule.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Communication-governance decisions are effective-dated "
|
|
"institutional evidence and must be corrected by a new decision."
|
|
),
|
|
)
|
|
)
|
|
|
|
for decision in _related_or_attributed(
|
|
db,
|
|
ContactPointQualityDecision,
|
|
tenant_id=tenant_id,
|
|
related_field=ContactPointQualityDecision.contact_id,
|
|
related_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
actor_fields=(ContactPointQualityDecision.created_by_account_id,),
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_quality_decision",
|
|
decision.id,
|
|
"contact_quality_evidence",
|
|
"Address contact-point quality decision",
|
|
{
|
|
"match_fields": _related_actor_match_fields(
|
|
decision,
|
|
contact_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
actor_fields=("created_by_account_id",),
|
|
),
|
|
"contact_id": (
|
|
decision.contact_id
|
|
if decision.contact_id in contact_ids
|
|
else None
|
|
),
|
|
"channel": decision.channel,
|
|
"contact_point_id": (
|
|
decision.contact_point_id
|
|
if decision.contact_id in contact_ids
|
|
else None
|
|
),
|
|
"state": decision.state,
|
|
"reason_code": decision.reason_code,
|
|
"reason": _bounded_text(decision.reason, 2_000),
|
|
"effective_from": _iso(decision.effective_from),
|
|
"effective_until": _iso(decision.effective_until),
|
|
},
|
|
observed_at=decision.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Quality decisions explain recipient eligibility and are "
|
|
"retained as effective-dated evidence."
|
|
),
|
|
)
|
|
)
|
|
|
|
for provenance in _related_or_attributed(
|
|
db,
|
|
ContactFieldProvenance,
|
|
tenant_id=tenant_id,
|
|
related_field=ContactFieldProvenance.contact_id,
|
|
related_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
actor_fields=(ContactFieldProvenance.created_by_account_id,),
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_field_provenance",
|
|
provenance.id,
|
|
"contact_provenance_evidence",
|
|
"Address field provenance",
|
|
{
|
|
"match_fields": _related_actor_match_fields(
|
|
provenance,
|
|
contact_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
actor_fields=("created_by_account_id",),
|
|
),
|
|
"contact_id": (
|
|
provenance.contact_id
|
|
if provenance.contact_id in contact_ids
|
|
else None
|
|
),
|
|
"field_path": _bounded_text(provenance.field_path, 255),
|
|
"source_kind": provenance.source_kind,
|
|
"source_revision": _bounded_text(
|
|
provenance.source_revision,
|
|
255,
|
|
),
|
|
"precedence": provenance.precedence,
|
|
"selected": provenance.selected,
|
|
"reason_code": provenance.reason_code,
|
|
"explanation": _bounded_text(
|
|
provenance.explanation,
|
|
2_000,
|
|
),
|
|
"visibility": provenance.visibility,
|
|
},
|
|
observed_at=provenance.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Field provenance is retained to explain source authority, "
|
|
"normalization, and merge survivorship."
|
|
),
|
|
)
|
|
)
|
|
|
|
for merge in _matching_merges(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
contact_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
):
|
|
related_ids = {
|
|
merge.winner_contact_id,
|
|
*(str(item) for item in merge.loser_contact_ids),
|
|
} & contact_ids
|
|
actor_fields = _actor_match_fields(
|
|
merge,
|
|
selectors.account_id,
|
|
(
|
|
"created_by_account_id",
|
|
"recovered_by_account_id",
|
|
),
|
|
)
|
|
append(
|
|
_record(
|
|
"addresses_merge_record",
|
|
merge.id,
|
|
"contact_merge_evidence",
|
|
"Address contact merge record",
|
|
{
|
|
"match_fields": (
|
|
(["contact_id"] if related_ids else []) + actor_fields
|
|
),
|
|
"related_contact_ids": sorted(related_ids),
|
|
"status": merge.status,
|
|
"reason": _bounded_text(merge.reason, 2_000),
|
|
"before_hash": merge.before_hash,
|
|
"after_hash": merge.after_hash,
|
|
"recovered_at": _iso(merge.recovered_at),
|
|
"recovery_action": merge.recovery_action,
|
|
"recovery_reason": _bounded_text(
|
|
merge.recovery_reason,
|
|
2_000,
|
|
),
|
|
},
|
|
observed_at=merge.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Merge records and hashes are retained so redirects, list "
|
|
"repairs, undo, and split operations remain explainable."
|
|
),
|
|
)
|
|
)
|
|
|
|
for redirect in _matching_redirects(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
contact_ids=contact_ids,
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_contact_redirect",
|
|
redirect.id,
|
|
"contact_merge_evidence",
|
|
"Address contact redirect",
|
|
{
|
|
"source_contact_id": redirect.source_contact_id,
|
|
"target_contact_id": redirect.target_contact_id,
|
|
"merge_record_id": redirect.merge_record_id,
|
|
"ended_at": _iso(redirect.ended_at),
|
|
},
|
|
observed_at=redirect.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Contact redirects preserve resolution of historical references."
|
|
),
|
|
)
|
|
)
|
|
|
|
for tombstone in _related_rows(
|
|
db,
|
|
AddressSyncTombstone,
|
|
tenant_id=tenant_id,
|
|
related_field=AddressSyncTombstone.contact_id,
|
|
related_ids=contact_ids,
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_sync_tombstone",
|
|
tombstone.id,
|
|
"address_sync_evidence",
|
|
"Address synchronization tombstone",
|
|
{
|
|
"contact_id": tombstone.contact_id,
|
|
"sync_source_id": tombstone.sync_source_id,
|
|
"address_book_id": tombstone.address_book_id,
|
|
"local_deleted_at": _iso(tombstone.local_deleted_at),
|
|
"remote_deleted_at": _iso(tombstone.remote_deleted_at),
|
|
"synced_at": _iso(tombstone.synced_at),
|
|
},
|
|
observed_at=tombstone.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Synchronization tombstones prevent accidental recreation "
|
|
"and document source lifecycle outcomes."
|
|
),
|
|
)
|
|
)
|
|
|
|
for conflict in _related_or_attributed(
|
|
db,
|
|
AddressSyncConflict,
|
|
tenant_id=tenant_id,
|
|
related_field=AddressSyncConflict.contact_id,
|
|
related_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
actor_fields=(AddressSyncConflict.resolved_by_account_id,),
|
|
):
|
|
append(
|
|
_record(
|
|
"addresses_sync_conflict",
|
|
conflict.id,
|
|
"address_sync_evidence",
|
|
"Address synchronization conflict",
|
|
{
|
|
"match_fields": _related_actor_match_fields(
|
|
conflict,
|
|
contact_ids=contact_ids,
|
|
account_id=selectors.account_id,
|
|
actor_fields=("resolved_by_account_id",),
|
|
),
|
|
"contact_id": (
|
|
conflict.contact_id
|
|
if conflict.contact_id in contact_ids
|
|
else None
|
|
),
|
|
"sync_source_id": conflict.sync_source_id,
|
|
"address_book_id": conflict.address_book_id,
|
|
"field_path": _bounded_text(conflict.field_path, 500),
|
|
"status": conflict.status,
|
|
"resolution": conflict.resolution,
|
|
"resolved_at": _iso(conflict.resolved_at),
|
|
},
|
|
observed_at=conflict.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Synchronization conflict outcomes are retained to explain "
|
|
"which authoritative value was selected."
|
|
),
|
|
)
|
|
)
|
|
|
|
_append_actor_attribution_records(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
account_id=selectors.account_id,
|
|
append=append,
|
|
)
|
|
|
|
return tuple(records)
|
|
|
|
def plan_erasure(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
records: Sequence[DsarRecordRef],
|
|
) -> Sequence[DsarErasureActionRef]:
|
|
del tenant_id
|
|
_session(session)
|
|
if _subject_selectors(subject) is None:
|
|
raise ValueError("Addresses DSAR subject selectors conflict.")
|
|
actions: list[DsarErasureActionRef] = []
|
|
for record in records:
|
|
_validate_record(record)
|
|
if record.immutable_evidence:
|
|
kind = "retain"
|
|
title = f"Retain {record.title}"
|
|
rationale = record.retention_reason or (
|
|
"Addresses governance evidence must be retained."
|
|
)
|
|
else:
|
|
kind = "manual_review"
|
|
title = f"Review {record.title}"
|
|
rationale = (
|
|
"Reusable address data can be shared, synchronized, merged, "
|
|
"or referenced by immutable recipient evidence. An authorized "
|
|
"operator must correct, archive, or erase it through the owning "
|
|
"Addresses workflow after reviewing those dependencies."
|
|
)
|
|
actions.append(
|
|
DsarErasureActionRef(
|
|
action_id=(
|
|
f"addresses:{kind}:{record.resource_type}:{record.resource_id}"
|
|
),
|
|
provider_id=self.provider_id,
|
|
module_id=self.module_id,
|
|
kind=kind,
|
|
resource_type=record.resource_type,
|
|
resource_id=record.resource_id,
|
|
title=title,
|
|
rationale=rationale,
|
|
executable=False,
|
|
)
|
|
)
|
|
return tuple(actions)
|
|
|
|
def execute_erasure(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
actions: Sequence[DsarErasureActionRef],
|
|
request_id: str,
|
|
) -> Sequence[DsarExecutionResultRef]:
|
|
del tenant_id
|
|
_session(session)
|
|
if _subject_selectors(subject) is None:
|
|
raise ValueError("Addresses DSAR subject selectors conflict.")
|
|
results: list[DsarExecutionResultRef] = []
|
|
for action in actions:
|
|
_validate_action(action)
|
|
if action.executable:
|
|
raise ValueError(
|
|
"Addresses DSAR does not publish executable erasure actions."
|
|
)
|
|
results.append(
|
|
DsarExecutionResultRef(
|
|
action_id=action.action_id,
|
|
status="blocked",
|
|
summary=(
|
|
"Use the authorized Addresses correction, archive, source, "
|
|
"merge, or governance workflow after dependency review."
|
|
),
|
|
evidence={"request_id": request_id},
|
|
)
|
|
)
|
|
return tuple(results)
|
|
|
|
|
|
def _subject_contact_ids(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
selectors: _SubjectSelectors,
|
|
) -> set[str] | None:
|
|
email_ids: set[str] = set()
|
|
if selectors.email:
|
|
email_ids = {
|
|
row[0]
|
|
for row in _bounded_rows(
|
|
session.query(Contact.id)
|
|
.join(ContactEmail, ContactEmail.contact_id == Contact.id)
|
|
.filter(
|
|
Contact.tenant_id == tenant_id,
|
|
or_(
|
|
func.lower(ContactEmail.email) == selectors.email,
|
|
func.lower(ContactEmail.normalized_email) == selectors.email,
|
|
),
|
|
)
|
|
.order_by(Contact.id)
|
|
)
|
|
}
|
|
|
|
direct_ids: set[str] = set()
|
|
direct_queries = (
|
|
("contact", Contact, Contact.id),
|
|
("contact_email", ContactEmail, ContactEmail.id),
|
|
("contact_phone", ContactPhone, ContactPhone.id),
|
|
(
|
|
"contact_postal_address",
|
|
ContactPostalAddress,
|
|
ContactPostalAddress.id,
|
|
),
|
|
)
|
|
for kind, model, id_field in direct_queries:
|
|
reference = selectors.references.get(kind)
|
|
if not reference:
|
|
continue
|
|
query = session.query(Contact.id)
|
|
if model is not Contact:
|
|
query = query.join(model, model.contact_id == Contact.id)
|
|
row = (
|
|
query.filter(
|
|
Contact.tenant_id == tenant_id,
|
|
id_field == reference,
|
|
)
|
|
.order_by(Contact.id)
|
|
.one_or_none()
|
|
)
|
|
if row is None:
|
|
return None
|
|
direct_ids.add(row[0])
|
|
|
|
if selectors.email and direct_ids and not direct_ids.issubset(email_ids):
|
|
return None
|
|
return email_ids | direct_ids
|
|
|
|
|
|
def _append_actor_attribution_records(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
account_id: str | None,
|
|
append: object,
|
|
) -> None:
|
|
if not account_id:
|
|
return
|
|
|
|
definitions = (
|
|
(
|
|
AddressBook,
|
|
"addresses_address_book_attribution",
|
|
"address_book_configuration",
|
|
("created_by_account_id", "updated_by_account_id"),
|
|
lambda row: {
|
|
"scope_type": row.scope_type,
|
|
"scope_id": row.scope_id,
|
|
"name": _bounded_text(row.name, 255),
|
|
"source_kind": row.source_kind,
|
|
"read_only": row.read_only,
|
|
"deleted_at": _iso(row.deleted_at),
|
|
},
|
|
),
|
|
(
|
|
Contact,
|
|
"addresses_contact_attribution",
|
|
"contact_configuration",
|
|
("created_by_account_id", "updated_by_account_id"),
|
|
lambda row: {
|
|
"address_book_id": row.address_book_id,
|
|
"source_kind": row.source_kind,
|
|
"deleted_at": _iso(row.deleted_at),
|
|
},
|
|
),
|
|
(
|
|
AddressList,
|
|
"addresses_list_attribution",
|
|
"address_list_configuration",
|
|
("created_by_account_id", "updated_by_account_id"),
|
|
lambda row: {
|
|
"address_book_id": row.address_book_id,
|
|
"name": _bounded_text(row.name, 255),
|
|
"source_kind": row.source_kind,
|
|
"read_only": row.read_only,
|
|
"deleted_at": _iso(row.deleted_at),
|
|
},
|
|
),
|
|
(
|
|
AddressSyncSource,
|
|
"addresses_sync_source_attribution",
|
|
"address_source_configuration",
|
|
("created_by_account_id", "updated_by_account_id"),
|
|
lambda row: {
|
|
"address_book_id": row.address_book_id,
|
|
"connector_type": row.connector_type,
|
|
"display_name": _bounded_text(row.display_name, 255),
|
|
"sync_direction": row.sync_direction,
|
|
"read_only": row.read_only,
|
|
"enabled": row.enabled,
|
|
"status": row.status,
|
|
},
|
|
),
|
|
(
|
|
AddressImportProfile,
|
|
"addresses_import_profile_attribution",
|
|
"address_import_configuration",
|
|
("created_by_account_id",),
|
|
lambda row: {
|
|
"profile_key": row.profile_key,
|
|
"version": row.version,
|
|
"scope_type": row.scope_type,
|
|
"scope_id": row.scope_id,
|
|
"name": _bounded_text(row.name, 255),
|
|
"source_format": row.source_format,
|
|
"is_current": row.is_current,
|
|
"superseded_at": _iso(row.superseded_at),
|
|
},
|
|
),
|
|
(
|
|
AddressImportRun,
|
|
"addresses_import_run_attribution",
|
|
"address_import_evidence",
|
|
("created_by_account_id",),
|
|
lambda row: {
|
|
"address_book_id": row.address_book_id,
|
|
"profile_id": row.profile_id,
|
|
"source_filename": _bounded_text(row.source_filename, 500),
|
|
"source_format": row.source_format,
|
|
"input_hash": row.input_hash,
|
|
"plan_hash": row.plan_hash,
|
|
"status": row.status,
|
|
"row_count": row.row_count,
|
|
"applied_at": _iso(row.applied_at),
|
|
"rolled_back_at": _iso(row.rolled_back_at),
|
|
},
|
|
),
|
|
(
|
|
ContactPointSnapshot,
|
|
"addresses_snapshot_attribution",
|
|
"recipient_snapshot_evidence",
|
|
("created_by_account_id",),
|
|
lambda row: {
|
|
"source_id": row.source_id,
|
|
"contract_version": row.contract_version,
|
|
"source_revision": row.source_revision,
|
|
"purpose": _bounded_text(row.purpose, 120),
|
|
"effective_at": _iso(row.effective_at),
|
|
"generated_at": _iso(row.generated_at),
|
|
"recipient_count": row.recipient_count,
|
|
"excluded_count": row.excluded_count,
|
|
"snapshot_hash": row.snapshot_hash,
|
|
},
|
|
),
|
|
)
|
|
append_record = append
|
|
for model, resource_type, category, actor_fields, data_factory in definitions:
|
|
conditions = [getattr(model, field) == account_id for field in actor_fields]
|
|
rows = _bounded_rows(
|
|
session.query(model)
|
|
.filter(model.tenant_id == tenant_id, or_(*conditions))
|
|
.order_by(model.id)
|
|
)
|
|
for row in rows:
|
|
data = data_factory(row)
|
|
data["match_fields"] = _actor_match_fields(
|
|
row,
|
|
account_id,
|
|
actor_fields,
|
|
)
|
|
append_record( # type: ignore[operator]
|
|
_record(
|
|
resource_type,
|
|
row.id,
|
|
category,
|
|
"Addresses operator attribution",
|
|
data,
|
|
observed_at=row.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Operator attribution is retained with the governed "
|
|
"configuration or evidence record for accountability."
|
|
),
|
|
)
|
|
)
|
|
|
|
|
|
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
|
|
groups = {
|
|
"account_id": (
|
|
subject.account_id,
|
|
subject.external_references.get("addresses.account"),
|
|
subject.external_references.get("access.account"),
|
|
),
|
|
"email": (
|
|
subject.email,
|
|
subject.external_references.get("addresses.email"),
|
|
),
|
|
}
|
|
normalized: dict[str, str | None] = {}
|
|
for key, values in groups.items():
|
|
distinct = {
|
|
value
|
|
for item in values
|
|
if (
|
|
value := (
|
|
_normalized_email(item) if key == "email" else _normalized_id(item)
|
|
)
|
|
)
|
|
}
|
|
if len(distinct) > 1:
|
|
return None
|
|
normalized[key] = next(iter(distinct), None)
|
|
|
|
aliases = {
|
|
"addresses.contact": "contact",
|
|
"addresses.contact_email": "contact_email",
|
|
"addresses.contact_phone": "contact_phone",
|
|
"addresses.contact_postal_address": "contact_postal_address",
|
|
}
|
|
references = {
|
|
target: value
|
|
for source, target in aliases.items()
|
|
if (value := _normalized_id(subject.external_references.get(source)))
|
|
}
|
|
return _SubjectSelectors(references=references, **normalized)
|
|
|
|
|
|
def _rows_by_ids(
|
|
session: Session,
|
|
model: type,
|
|
*,
|
|
tenant_id: str,
|
|
ids: set[str],
|
|
) -> list[object]:
|
|
if not ids:
|
|
return []
|
|
return _bounded_rows(
|
|
session.query(model)
|
|
.filter(model.tenant_id == tenant_id, model.id.in_(ids))
|
|
.order_by(model.id)
|
|
)
|
|
|
|
|
|
def _contact_children(
|
|
session: Session,
|
|
model: type,
|
|
*,
|
|
tenant_id: str,
|
|
contact_ids: set[str],
|
|
) -> list[object]:
|
|
if not contact_ids:
|
|
return []
|
|
return _bounded_rows(
|
|
session.query(model)
|
|
.join(Contact, model.contact_id == Contact.id)
|
|
.filter(
|
|
Contact.tenant_id == tenant_id,
|
|
model.contact_id.in_(contact_ids),
|
|
)
|
|
.order_by(model.id)
|
|
)
|
|
|
|
|
|
def _list_entries(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
contact_ids: set[str],
|
|
) -> list[AddressListEntry]:
|
|
if not contact_ids:
|
|
return []
|
|
return _bounded_rows(
|
|
session.query(AddressListEntry)
|
|
.join(AddressList, AddressListEntry.address_list_id == AddressList.id)
|
|
.filter(
|
|
AddressList.tenant_id == tenant_id,
|
|
AddressListEntry.contact_id.in_(contact_ids),
|
|
)
|
|
.order_by(AddressListEntry.id)
|
|
)
|
|
|
|
|
|
def _related_rows(
|
|
session: Session,
|
|
model: type,
|
|
*,
|
|
tenant_id: str,
|
|
related_field: object,
|
|
related_ids: set[str],
|
|
) -> list[object]:
|
|
if not related_ids:
|
|
return []
|
|
return _bounded_rows(
|
|
session.query(model)
|
|
.filter(
|
|
model.tenant_id == tenant_id,
|
|
related_field.in_(related_ids), # type: ignore[attr-defined]
|
|
)
|
|
.order_by(model.id)
|
|
)
|
|
|
|
|
|
def _related_or_attributed(
|
|
session: Session,
|
|
model: type,
|
|
*,
|
|
tenant_id: str,
|
|
related_field: object,
|
|
related_ids: set[str],
|
|
account_id: str | None,
|
|
actor_fields: Sequence[object],
|
|
) -> list[object]:
|
|
conditions = []
|
|
if related_ids:
|
|
conditions.append(related_field.in_(related_ids)) # type: ignore[attr-defined]
|
|
if account_id:
|
|
conditions.extend(field == account_id for field in actor_fields)
|
|
if not conditions:
|
|
return []
|
|
return _bounded_rows(
|
|
session.query(model)
|
|
.filter(model.tenant_id == tenant_id, or_(*conditions))
|
|
.order_by(model.id)
|
|
)
|
|
|
|
|
|
def _matching_merges(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
contact_ids: set[str],
|
|
account_id: str | None,
|
|
) -> list[ContactMergeRecord]:
|
|
rows = _bounded_rows(
|
|
session.query(ContactMergeRecord)
|
|
.filter(ContactMergeRecord.tenant_id == tenant_id)
|
|
.order_by(ContactMergeRecord.id)
|
|
)
|
|
return [
|
|
row
|
|
for row in rows
|
|
if row.winner_contact_id in contact_ids
|
|
or bool({str(item) for item in row.loser_contact_ids} & contact_ids)
|
|
or bool(
|
|
_actor_match_fields(
|
|
row,
|
|
account_id,
|
|
("created_by_account_id", "recovered_by_account_id"),
|
|
)
|
|
)
|
|
]
|
|
|
|
|
|
def _matching_redirects(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
contact_ids: set[str],
|
|
) -> list[ContactRedirect]:
|
|
if not contact_ids:
|
|
return []
|
|
return _bounded_rows(
|
|
session.query(ContactRedirect)
|
|
.filter(
|
|
ContactRedirect.tenant_id == tenant_id,
|
|
or_(
|
|
ContactRedirect.source_contact_id.in_(contact_ids),
|
|
ContactRedirect.target_contact_id.in_(contact_ids),
|
|
),
|
|
)
|
|
.order_by(ContactRedirect.id)
|
|
)
|
|
|
|
|
|
def _contact_point_match_fields(
|
|
row: object,
|
|
*,
|
|
selectors: _SubjectSelectors,
|
|
reference_kind: str,
|
|
) -> list[str]:
|
|
fields = []
|
|
if selectors.references.get(reference_kind) == getattr(row, "id"):
|
|
fields.append("reference")
|
|
if reference_kind == "contact_email" and selectors.email:
|
|
if selectors.email in {
|
|
_normalized_email(getattr(row, "email", None)),
|
|
_normalized_email(getattr(row, "normalized_email", None)),
|
|
}:
|
|
fields.append("email")
|
|
return fields
|
|
|
|
|
|
def _related_actor_match_fields(
|
|
row: object,
|
|
*,
|
|
contact_ids: set[str],
|
|
account_id: str | None,
|
|
actor_fields: Sequence[str],
|
|
) -> list[str]:
|
|
fields = []
|
|
if getattr(row, "contact_id", None) in contact_ids:
|
|
fields.append("contact_id")
|
|
fields.extend(_actor_match_fields(row, account_id, actor_fields))
|
|
return fields
|
|
|
|
|
|
def _actor_match_fields(
|
|
row: object,
|
|
account_id: str | None,
|
|
fields: Sequence[str],
|
|
) -> list[str]:
|
|
if not account_id:
|
|
return []
|
|
return [field for field in fields if getattr(row, field, None) == account_id]
|
|
|
|
|
|
def _validate_record(record: DsarRecordRef) -> None:
|
|
if record.provider_id != "addresses" or record.module_id != "addresses":
|
|
raise ValueError("Addresses DSAR received a foreign provider record.")
|
|
|
|
|
|
def _validate_action(action: DsarErasureActionRef) -> None:
|
|
if action.provider_id != "addresses" or action.module_id != "addresses":
|
|
raise ValueError("Addresses DSAR received a foreign provider action.")
|
|
|
|
|
|
def _record(
|
|
resource_type: str,
|
|
resource_id: str,
|
|
category: str,
|
|
title: str,
|
|
data: Mapping[str, object],
|
|
*,
|
|
observed_at: datetime | None = None,
|
|
immutable: bool = False,
|
|
retention_reason: str | None = None,
|
|
) -> DsarRecordRef:
|
|
return DsarRecordRef(
|
|
provider_id="addresses",
|
|
module_id="addresses",
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
category=category,
|
|
title=title,
|
|
data=data,
|
|
observed_at=observed_at,
|
|
immutable_evidence=immutable,
|
|
retention_reason=retention_reason,
|
|
source_path="/address-book",
|
|
)
|
|
|
|
|
|
def _session(value: object) -> Session:
|
|
if not isinstance(value, Session):
|
|
raise TypeError("Addresses DSAR provider requires a SQLAlchemy session.")
|
|
return value
|
|
|
|
|
|
def _bounded_rows(query: object) -> list[object]:
|
|
rows = query.limit(_MAX_RECORDS + 1).all() # type: ignore[attr-defined]
|
|
if len(rows) > _MAX_RECORDS:
|
|
raise ValueError(
|
|
"Addresses DSAR match limit exceeded; narrow the subject selectors."
|
|
)
|
|
return rows
|
|
|
|
|
|
def _bounded_text(value: str | None, limit: int) -> str | None:
|
|
return value[:limit] if value else None
|
|
|
|
|
|
def _normalized_email(value: object) -> str | None:
|
|
if not isinstance(value, str):
|
|
return None
|
|
value = value.strip().casefold()
|
|
return value or None
|
|
|
|
|
|
def _normalized_id(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
value = str(value).strip()
|
|
return value or None
|
|
|
|
|
|
def _iso(value: datetime | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=timezone.utc)
|
|
return value.isoformat()
|
|
|
|
|
|
__all__ = ["ADDRESSES_DSAR_CAPABILITY", "AddressesDsarProvider"]
|