from __future__ import annotations
from .config import DnsRecordSpec, DnsZoneSpec
from .providers.aws import RecordSetInfo, Route53Provider, ZoneInfo
from .types import RecordPlan, RecordStep, StepAction, ZonePlan, ZoneStep, ZoneVisibility
def _canonical_zone_name(name: str) -> str:
n = name.strip().rstrip(".").lower()
return f"{n}."
def _same_values(a: list[str] | None, b: list[str] | None) -> bool:
if a is None or b is None:
return a == b
return sorted(a) == sorted(b)
[docs]
def plan_zones(*, provider: Route53Provider, desired: list[DnsZoneSpec]) -> ZonePlan:
steps: list[ZoneStep] = []
observed: dict[str, ZoneInfo] = {}
for spec in desired:
zone_name = _canonical_zone_name(spec.name)
hz = provider.get_zone_by_name(zone_name)
if hz is None:
steps.append(
ZoneStep(
action=StepAction.CREATE_PUBLIC_ZONE,
zone_name=zone_name,
summary=f"+ create public zone: {zone_name}",
)
)
continue
# Found a zone with the same name.
observed[zone_name] = hz
if hz.visibility == ZoneVisibility.PRIVATE:
# DNS name exists but only privately; we refuse because it is ambiguous / risky.
raise ValueError(f"Zone '{zone_name}' exists but is private; refusing to converge")
# Public zone exists: NOOP
steps.append(
ZoneStep(
action=StepAction.NOOP,
zone_name=zone_name,
summary=f"= noop public zone: {zone_name}",
)
)
return ZonePlan(desired=desired, steps=steps, observed=observed)
[docs]
def plan_records(
*,
provider: Route53Provider,
desired: list[DnsRecordSpec],
) -> RecordPlan:
"""Plan record convergence within existing hosted zones.
Supports both:
- Non-alias records: TTL + values
- Alias records: A/AAAA with AliasTarget (dns_name + hosted_zone_id)
"""
steps: list[RecordStep] = []
observed: dict[tuple[str, str], RecordSetInfo] = {}
for spec in desired:
spec.validate()
zone_name = _canonical_zone_name(spec.zone)
fqdn = spec.fqdn().rstrip(".") + "."
rtype = spec.type.upper()
# Resolve zone -> id
hz = provider.get_zone_by_name(zone_name)
if hz is None:
raise ValueError(
f"Zone '{zone_name}' does not exist; create the zone before planning records"
)
if hz.visibility == ZoneVisibility.PRIVATE:
raise ValueError(
f"Zone '{zone_name}' is private; refusing to plan public DNS records"
)
current = provider.get_record_set(
zone_id=hz.zone_id,
fqdn=fqdn,
record_type=rtype,
)
if current is None:
steps.append(
RecordStep(
action=StepAction.UPSERT_RECORD,
zone_name=zone_name,
fqdn=fqdn,
record_type=rtype,
summary=f"+ upsert record: {fqdn} {rtype}",
)
)
continue
observed[(fqdn, rtype)] = current
# ---- Alias desired ----
if spec.alias is not None:
# If current is non-alias or alias target differs -> upsert
if current.alias_target is None:
steps.append(
RecordStep(
action=StepAction.UPSERT_RECORD,
zone_name=zone_name,
fqdn=fqdn,
record_type=rtype,
summary=f"~ upsert record (to alias): {fqdn} {rtype}",
)
)
continue
desired_alias = {
"dns_name": spec.alias.dns_name.rstrip(".") + ".",
"hosted_zone_id": spec.alias.hosted_zone_id,
}
current_alias = {
"dns_name": (current.alias_target.get("dns_name") or "").rstrip(".") + ".",
"hosted_zone_id": current.alias_target.get("hosted_zone_id") or "",
}
if desired_alias == current_alias:
steps.append(
RecordStep(
action=StepAction.NOOP_RECORD,
zone_name=zone_name,
fqdn=fqdn,
record_type=rtype,
summary=f"= noop alias record: {fqdn} {rtype}",
)
)
else:
steps.append(
RecordStep(
action=StepAction.UPSERT_RECORD,
zone_name=zone_name,
fqdn=fqdn,
record_type=rtype,
summary=f"~ upsert alias record: {fqdn} {rtype}",
)
)
continue
# ---- Non-alias desired ----
if current.alias_target is not None:
steps.append(
RecordStep(
action=StepAction.UPSERT_RECORD,
zone_name=zone_name,
fqdn=fqdn,
record_type=rtype,
summary=f"~ upsert record (replace alias): {fqdn} {rtype}",
)
)
continue
if current.ttl == spec.ttl and _same_values(current.values, spec.values):
steps.append(
RecordStep(
action=StepAction.NOOP_RECORD,
zone_name=zone_name,
fqdn=fqdn,
record_type=rtype,
summary=f"= noop record: {fqdn} {rtype}",
)
)
continue
steps.append(
RecordStep(
action=StepAction.UPSERT_RECORD,
zone_name=zone_name,
fqdn=fqdn,
record_type=rtype,
summary=f"~ upsert record: {fqdn} {rtype}",
)
)
return RecordPlan(desired=desired, steps=steps, observed=observed)