from collections import defaultdict from collections.abc import Iterable, Mapping, Sequence from typing import Protocol from .adapters import ( DNS_RECORDS, NormalizedObject, Observation, ) ObservationGroupKey = tuple[str, str] class ReconciliationConflict(RuntimeError): """Raised when observations cannot be reconciled safely.""" class ReconciliationPolicy(Protocol): object_type: str def reconcile( self, logical_key: str, observations: Sequence[Observation], ) -> NormalizedObject: ... def _observation_sort_key( observation: Observation, ) -> tuple[str, str, str]: return ( observation.endpoint_id, observation.object.local_key or "", observation.observed_at.isoformat(), ) def group_observations( observations: Iterable[Observation], ) -> dict[ObservationGroupKey, tuple[Observation, ...]]: grouped: dict[ ObservationGroupKey, list[Observation], ] = defaultdict(list) for observation in observations: normalized = observation.object if not observation.endpoint_id: raise RuntimeError("Observation endpoint_id cannot be empty") if not normalized.object_type: raise RuntimeError("Observation object_type cannot be empty") if not normalized.logical_key: raise RuntimeError("Observation logical_key cannot be empty") key = ( normalized.object_type, normalized.logical_key, ) grouped[key].append(observation) return { key: tuple(sorted(rows, key=_observation_sort_key)) for key, rows in sorted(grouped.items()) } class DnsUnionPolicy: object_type = DNS_RECORDS def reconcile( self, logical_key: str, observations: Sequence[Observation], ) -> NormalizedObject: if not observations: raise RuntimeError( "DNS union policy requires at least one observation" ) first = observations[0].object conflicting_endpoints = { observation.endpoint_id for observation in observations if observation.object.payload != first.payload } if conflicting_endpoints: conflicting_endpoints.add(observations[0].endpoint_id) endpoint_list = ", ".join( sorted(conflicting_endpoints) ) raise ReconciliationConflict( f"Conflicting {DNS_RECORDS} observations for " f"{logical_key!r} from {endpoint_list}" ) return NormalizedObject( object_type=DNS_RECORDS, logical_key=logical_key, payload=dict(first.payload), ) DEFAULT_POLICIES: Mapping[str, ReconciliationPolicy] = { DNS_RECORDS: DnsUnionPolicy(), } def reconcile_observations( observations: Iterable[Observation], policies: Mapping[str, ReconciliationPolicy] | None = None, ) -> list[NormalizedObject]: selected_policies = ( DEFAULT_POLICIES if policies is None else policies ) desired: list[NormalizedObject] = [] for (object_type, logical_key), rows in ( group_observations(observations).items() ): policy = selected_policies.get(object_type) if policy is None: raise RuntimeError( f"No reconciliation policy for {object_type!r}" ) if policy.object_type != object_type: raise RuntimeError( f"Policy for {object_type!r} advertises " f"{policy.object_type!r}" ) desired.append(policy.reconcile(logical_key, rows)) return desired