From 9ecdc6d713f441d46deff8ff1dc51d609364e833 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Sun, 2 Aug 2026 06:20:18 +0200 Subject: [PATCH] Add contact point resolution contract --- docs/MODULE_ARCHITECTURE.md | 3 +- src/govoplan_core/core/contact_points.py | 191 +++++++++++++++++++++++ tests/test_contact_point_contract.py | 55 +++++++ 3 files changed, 248 insertions(+), 1 deletion(-) create mode 100644 src/govoplan_core/core/contact_points.py create mode 100644 tests/test_contact_point_contract.py diff --git a/docs/MODULE_ARCHITECTURE.md b/docs/MODULE_ARCHITECTURE.md index a21b6de..48e7165 100644 --- a/docs/MODULE_ARCHITECTURE.md +++ b/docs/MODULE_ARCHITECTURE.md @@ -286,7 +286,8 @@ such as `calendar:sync-source:`. Current named interfaces, generated from the source manifests by the workspace contract checks, are: -- `addresses.contact_writer`, `addresses.lookup`, `addresses.recipient_source` +- `addresses.contact_point_resolution`, `addresses.contact_writer`, + `addresses.lookup`, `addresses.recipient_source` - `calendar.external_profiles`, `calendar.invitations`, `calendar.outbox`, `calendar.scheduling` - `campaigns.access`, `campaigns.delivery_tasks`, diff --git a/src/govoplan_core/core/contact_points.py b/src/govoplan_core/core/contact_points.py new file mode 100644 index 0000000..8450a99 --- /dev/null +++ b/src/govoplan_core/core/contact_points.py @@ -0,0 +1,191 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from typing import Literal, Protocol, runtime_checkable + +from govoplan_core.core.distribution_lists import ( + DistributionChannel, + DistributionExplanation, + DistributionOutcome, + DistributionSourceReference, +) + + +CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION = "addresses.contact_point_resolution" +CONTACT_POINT_CONTRACT_VERSION = "1.0" + +ContactPointFallbackRule = Literal["none", "primary", "any"] +PostalAddressFormat = Literal["domestic", "international"] + + +@dataclass(frozen=True, slots=True) +class ContactPointResolutionRequest: + tenant_id: str + subject: DistributionSourceReference + effective_at: datetime + purpose: str | None = None + requested_channels: tuple[DistributionChannel, ...] = () + address_purpose: str | None = None + fallback_rule: ContactPointFallbackRule = "primary" + locale: str | None = None + postal_format: PostalAddressFormat = "domestic" + context: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class ContactPointCandidate: + channel: DistributionChannel + target: str + target_key: str + status: DistributionOutcome + contact_point_id: str | None = None + address_purpose: str | None = None + locale: str | None = None + preferred: bool = False + preference_rank: int | None = None + reason_code: str | None = None + explanation: str | None = None + source: DistributionSourceReference | None = None + source_revision: str | None = None + preference_revision: str | None = None + consent_revision: str | None = None + value: Mapping[str, object] = field(default_factory=dict) + provenance: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class ContactPointResolution: + contract_version: str + subject: DistributionSourceReference + status: DistributionOutcome + contact_id: str | None = None + display_name: str | None = None + candidates: tuple[ContactPointCandidate, ...] = () + excluded: tuple[ContactPointCandidate, ...] = () + explanations: tuple[DistributionExplanation, ...] = () + source_revision: str | None = None + source_fingerprint: str | None = None + provenance: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class ContactPointSourceRequest: + tenant_id: str + source_id: str + effective_at: datetime + purpose: str | None = None + requested_channels: tuple[DistributionChannel, ...] = () + address_purpose: str | None = None + fallback_rule: ContactPointFallbackRule = "primary" + locale: str | None = None + postal_format: PostalAddressFormat = "domestic" + max_items: int = 5_000 + context: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class ContactPointSourcePreview: + contract_version: str + source: DistributionSourceReference + request: ContactPointSourceRequest + resolutions: tuple[ContactPointResolution, ...] + total_count: int + usable_count: int + excluded_count: int + offset: int + limit: int + has_more: bool + source_revision: str + source_fingerprint: str + generated_at: datetime + provenance: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class ContactPointSnapshotRef: + id: str + tenant_id: str + contract_version: str + source: DistributionSourceReference + request: ContactPointSourceRequest + resolutions: tuple[ContactPointResolution, ...] + recipient_count: int + excluded_count: int + source_revision: str + source_fingerprint: str + snapshot_hash: str + generated_at: datetime + provenance: Mapping[str, object] = field(default_factory=dict) + + +@runtime_checkable +class ContactPointResolutionProvider(Protocol): + def resolve_contact_points( + self, + session: object, + principal: object, + *, + request: ContactPointResolutionRequest, + ) -> ContactPointResolution: + ... + + def preview_source( + self, + session: object, + principal: object, + *, + request: ContactPointSourceRequest, + offset: int = 0, + limit: int = 100, + ) -> ContactPointSourcePreview: + ... + + def freeze_source( + self, + session: object, + principal: object, + *, + request: ContactPointSourceRequest, + ) -> ContactPointSnapshotRef: + ... + + def get_snapshot( + self, + session: object, + principal: object, + *, + snapshot_id: str, + ) -> ContactPointSnapshotRef | None: + ... + + +def contact_point_resolution_provider( + registry: object | None, +) -> ContactPointResolutionProvider | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not hasattr(registry, "capability") + or not registry.has_capability(CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION) + ): + return None + capability = registry.capability(CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION) + return capability if isinstance(capability, ContactPointResolutionProvider) else None + + +__all__ = [ + "CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION", + "CONTACT_POINT_CONTRACT_VERSION", + "ContactPointCandidate", + "ContactPointFallbackRule", + "ContactPointResolution", + "ContactPointResolutionProvider", + "ContactPointResolutionRequest", + "ContactPointSnapshotRef", + "ContactPointSourcePreview", + "ContactPointSourceRequest", + "PostalAddressFormat", + "contact_point_resolution_provider", +] diff --git a/tests/test_contact_point_contract.py b/tests/test_contact_point_contract.py new file mode 100644 index 0000000..4073faa --- /dev/null +++ b/tests/test_contact_point_contract.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import unittest +from datetime import UTC, datetime + +from govoplan_core.core.contact_points import ( + CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION, + CONTACT_POINT_CONTRACT_VERSION, + ContactPointResolutionRequest, + contact_point_resolution_provider, +) +from govoplan_core.core.distribution_lists import DistributionSourceReference + + +class _Registry: + def __init__(self, capability: object | None = None) -> None: + self._capability = capability + + def has_capability(self, name: str) -> bool: + return ( + name == CAPABILITY_ADDRESSES_CONTACT_POINT_RESOLUTION + and self._capability is not None + ) + + def capability(self, _name: str) -> object: + return self._capability + + +class ContactPointContractTests(unittest.TestCase): + def test_request_is_provider_neutral_and_versioned(self) -> None: + request = ContactPointResolutionRequest( + tenant_id="tenant-1", + subject=DistributionSourceReference( + provider="idm", + resource_type="identity", + resource_id="identity-1", + ), + effective_at=datetime(2026, 8, 2, tzinfo=UTC), + requested_channels=("email", "postal"), + address_purpose="official", + fallback_rule="primary", + postal_format="international", + ) + + self.assertEqual("1.0", CONTACT_POINT_CONTRACT_VERSION) + self.assertEqual("idm", request.subject.provider) + self.assertEqual(("email", "postal"), request.requested_channels) + + def test_accessor_rejects_objects_that_do_not_implement_the_protocol(self) -> None: + self.assertIsNone(contact_point_resolution_provider(None)) + self.assertIsNone(contact_point_resolution_provider(_Registry(object()))) + + +if __name__ == "__main__": + unittest.main()