GovOPlaN/govoplan-core
1
0
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects Releases Wiki Activity

Add contact point resolution contract

Browse Source
This commit is contained in:
zemion
2026-08-02 06:20:18 +02:00
parent ca35aad286
commit 9ecdc6d713
3 changed files with 248 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/MODULE_ARCHITECTURE.md
+2 -1
View File
@@ -286,7 +286,8 @@ such as `calendar:sync-source:<id>`.
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`,
src/govoplan_core/core/contact_points.py
+191
View File
@@ -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",
]
tests/test_contact_point_contract.py
+55
View File
@@ -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()