Source code for dxaws_cloudfront.models
from __future__ import annotations
"""Models for dxaws-cloudfront.
These models are intentionally small and stable. The planner/executor/provider
layers build on these.
MVP focus: a single CloudFront distribution fronting an S3 origin using OAC.
"""
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
[docs]
class PriceClass(str, Enum):
ALL = "PriceClass_All"
P200 = "PriceClass_200"
P100 = "PriceClass_100"
[docs]
class ViewerProtocolPolicy(str, Enum):
REDIRECT_TO_HTTPS = "redirect-to-https"
HTTPS_ONLY = "https-only"
ALLOW_ALL = "allow-all"
[docs]
class OriginProtocolPolicy(str, Enum):
HTTP_ONLY = "http-only"
HTTPS_ONLY = "https-only"
MATCH_VIEWER = "match-viewer"
[docs]
class HttpVersion(str, Enum):
HTTP1_1 = "http1.1"
HTTP2 = "http2"
HTTP2_AND_3 = "http2and3"
HTTP3 = "http3"
# ---------------------------------------------------------------------------
# Desired State
# ---------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class S3OriginDesired:
"""S3 origin settings.
Provide either `bucket_name` or `domain_name`. If both are provided, bucket_name
is the canonical identity and domain_name is used as an explicit override.
"""
bucket_name: str | None = None
domain_name: str | None = None
# If you want to lock this down later:
# origin_path: str | None = None
[docs]
@dataclass(frozen=True)
class CacheBehaviorDesired:
"""Desired default cache behavior (MVP subset)."""
viewer_protocol_policy: ViewerProtocolPolicy = ViewerProtocolPolicy.REDIRECT_TO_HTTPS
# TTL defaults (CloudFront accepts ints in seconds)
min_ttl: int = 0
default_ttl: int = 3600
max_ttl: int = 86400
# Whether to enable compression at edge
compress: bool = True
[docs]
@dataclass(frozen=True)
class CustomErrorResponsesDesired:
"""SPA-friendly error responses (optional)."""
enabled: bool = False
response_page_path: str = "/index.html"
error_codes: tuple[int, ...] = (403, 404)
response_code: int = 200
error_caching_min_ttl: int = 0
[docs]
@dataclass(frozen=True)
class ViewerCertificateDesired:
"""Viewer certificate settings.
If `acm_certificate_arn` is provided, we will use it with SNI-only.
Otherwise, we will use the CloudFront default certificate.
"""
acm_certificate_arn: str | None = None
minimum_protocol_version: str = "TLSv1.2_2021"
ssl_support_method: str = "sni-only" # only relevant when custom cert is used
[docs]
@dataclass(frozen=True)
class DistributionDesired:
"""Top-level desired state for a single distribution.
Public callers can use a compact interface (`aliases`, `origin_domain`,
`certificate_arn`, `tags`) while advanced callers can still provide the
richer nested `origin` and `viewer_certificate` models directly.
The dataclass normalizes the compact fields into the nested models so the
rest of the planner/provider stack can continue to work with a stable
internal shape.
"""
# A stable logical identifier for idempotency/discovery.
# If omitted, we derive it from the first alias or origin domain.
name: str | None = None
present: bool = True
enabled: bool = True
comment: str | None = None
# Aliases (CNAMEs)
aliases: tuple[str, ...] = ()
# Compact public interface for common cases.
origin_domain: str | None = None
certificate_arn: str | None = None
# Advanced / normalized nested forms.
origin: S3OriginDesired = field(default_factory=S3OriginDesired)
# Default root object
default_root_object: str = "index.html"
# Cache behavior
default_cache_behavior: CacheBehaviorDesired = field(default_factory=CacheBehaviorDesired)
# Optional SPA-style error responses
custom_errors: CustomErrorResponsesDesired = field(default_factory=CustomErrorResponsesDesired)
# Certificate / TLS
viewer_certificate: ViewerCertificateDesired = field(default_factory=ViewerCertificateDesired)
# Global settings
price_class: PriceClass = PriceClass.P100
http_version: HttpVersion = HttpVersion.HTTP2
ipv6_enabled: bool = True
# Tags applied to the distribution (resource tags)
tags: dict[str, str] = field(default_factory=dict)
# Origin Access Control (OAC)
# If not provided, we derive a name from distribution `name`.
oac_name: str | None = None
def __post_init__(self) -> None:
derived_name = self.name or self._derive_name()
if not derived_name:
raise ValueError(
"DistributionDesired requires a name, at least one alias, or an origin_domain"
)
object.__setattr__(self, "name", derived_name)
normalized_origin_domain = self._normalize_origin_domain()
if normalized_origin_domain:
object.__setattr__(
self,
"origin",
S3OriginDesired(
bucket_name=self.origin.bucket_name,
domain_name=normalized_origin_domain,
),
)
object.__setattr__(self, "origin_domain", normalized_origin_domain)
normalized_certificate_arn = self._normalize_certificate_arn()
if normalized_certificate_arn:
object.__setattr__(
self,
"viewer_certificate",
ViewerCertificateDesired(
acm_certificate_arn=normalized_certificate_arn,
minimum_protocol_version=self.viewer_certificate.minimum_protocol_version,
ssl_support_method=self.viewer_certificate.ssl_support_method,
),
)
object.__setattr__(self, "certificate_arn", normalized_certificate_arn)
def _derive_name(self) -> str | None:
if self.aliases:
return self.aliases[0]
if self.origin_domain:
return self.origin_domain
if self.origin.domain_name:
return self.origin.domain_name
if self.origin.bucket_name:
return self.origin.bucket_name
return None
def _normalize_origin_domain(self) -> str | None:
flat = self.origin_domain
nested = self.origin.domain_name
if flat and nested and flat != nested:
raise ValueError(
"DistributionDesired origin_domain conflicts with origin.domain_name"
)
return flat or nested
def _normalize_certificate_arn(self) -> str | None:
flat = self.certificate_arn
nested = self.viewer_certificate.acm_certificate_arn
if flat and nested and flat != nested:
raise ValueError(
"DistributionDesired certificate_arn conflicts with "
"viewer_certificate.acm_certificate_arn"
)
return flat or nested
@property
def effective_origin_domain(self) -> str | None:
return self.origin.domain_name or self.origin_domain
@property
def effective_certificate_arn(self) -> str | None:
return (
self.viewer_certificate.acm_certificate_arn or self.certificate_arn
)
# ---------------------------------------------------------------------------
# Current State
# ---------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class DistributionCurrent:
"""Observed current state for a distribution.
This is a normalized representation. We do NOT store the entire AWS config here
unless needed. Keep it stable for planner comparisons.
"""
exists: bool
name: str
id: str | None = None
arn: str | None = None
domain_name: str | None = None
status: str | None = None
# Snapshot of key config parts we care about
enabled: bool | None = None
comment: str | None = None
aliases: tuple[str, ...] | None = None
origin_domain_name: str | None = None
default_root_object: str | None = None
# This will be derived from config and normalized for comparisons
viewer_protocol_policy: str | None = None
# Cert-related
acm_certificate_arn: str | None = None
using_default_certificate: bool | None = None
price_class: str | None = None
http_version: str | None = None
ipv6_enabled: bool | None = None
# Tags, best-effort (CloudFront tags are per-resource and require ARN)
tags: dict[str, str] | None = None
# Keep the raw config/etag if provider needs it for updates (optional).
raw_config: dict[str, Any] | None = None
etag: str | None = None
# ---------------------------------------------------------------------------
# Plan / Actions
# ---------------------------------------------------------------------------
[docs]
class ActionType(str, Enum):
CREATE_OAC = "create_oac"
CREATE_DISTRIBUTION = "create_distribution"
UPDATE_DISTRIBUTION = "update_distribution"
DISABLE_DISTRIBUTION = "disable_distribution"
DELETE_DISTRIBUTION = "delete_distribution"
TAG_DISTRIBUTION = "tag_distribution"
WAIT_DEPLOYED = "wait_deployed"
[docs]
@dataclass(frozen=True)
class Action:
type: ActionType
reason: str
payload: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass(frozen=True)
class Plan:
desired: DistributionDesired
current: DistributionCurrent
actions: list[Action] = field(default_factory=list)
@property
def is_noop(self) -> bool:
return len(self.actions) == 0
# ---------------------------------------------------------------------------
# Apply result / outputs
# ---------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class DistributionResult:
name: str
id: str
arn: str | None
domain_name: str | None
# Convenience for the DNS module later
hosted_zone_id: str | None = None # CloudFront's zone id is constant but we keep it explicit
[docs]
@dataclass(frozen=True)
class ManagerResult:
desired: DistributionDesired
current: DistributionCurrent
plan: Plan
result: DistributionResult
outcome: str