API Modules

This page is rendered directly from the module’s Python source using Sphinx autodoc.

dxaws_cloudfront.module

dxaws_cloudfront.module.CloudFrontModule

alias of Module

class dxaws_cloudfront.module.Module(*, provider: AwsProvider)

Bases: object

dxaws-cloudfront convergence module.

Note: This is NOT a DxAwsInfraModuleBase module. It does not deploy stacks. It converges a CloudFront distribution directly via AWS APIs.

apply(desired: DistributionDesired, *, wait_deployed: bool = True) → ModuleApplyResult

get_current(desired: DistributionDesired) → DistributionCurrent

Discover current state for a distribution based on deterministic identity.

MVP strategy:

• find by primary alias (first entry in desired.aliases)

• if found, fetch config + etag and normalize into DistributionCurrent

• best-effort tags if ARN is available

Note: comment-based discovery is intentionally not supported.

plan(desired: DistributionDesired, *, wait_deployed: bool = True) → ModulePlanResult

class dxaws_cloudfront.module.ModuleApplyResult(desired: 'DistributionDesired', current: 'DistributionCurrent', plan: 'Plan', result: 'DistributionResult')

Bases: object

current: DistributionCurrent

desired: DistributionDesired

plan: Plan

result: DistributionResult

class dxaws_cloudfront.module.ModulePlanResult(desired: 'DistributionDesired', current: 'DistributionCurrent', plan: 'Plan')

Bases: object

current: DistributionCurrent

desired: DistributionDesired

plan: Plan

dxaws_cloudfront.planner

dxaws_cloudfront.planner.current_summary_for_compare(cur: DistributionCurrent) → dict[str, Any]

dxaws_cloudfront.planner.desired_summary_for_compare(desired: DistributionDesired) → dict[str, Any]

Build a simple summary dict used for comparisons.

This keeps comparisons stable even if AWS returns extra fields in raw_config.

dxaws_cloudfront.planner.normalize_current_for_compare(cur: DistributionCurrent) → DistributionCurrent

Normalize None collections/strings for easier comparisons.

dxaws_cloudfront.planner.plan(desired: DistributionDesired, current: DistributionCurrent, *, oac_id: str | None, wait_deployed: bool = True) → Plan

Plan convergence actions.

oac_id is supplied by the module/executor (ensure/create in AWS) or None if OAC is not used yet.

dxaws_cloudfront.planner.resolve_oac_name(d: DistributionDesired) → str

dxaws_cloudfront.planner.resolve_origin_domain(origin: S3OriginDesired) → str

Resolve origin domain name from desired origin.

dxaws_cloudfront.planner.stable_comment(name: str) → str

Stable comment used for idempotent discovery.

dxaws_cloudfront.planner.synth_custom_errors(ce: CustomErrorResponsesDesired) → dict[str, Any] | None

dxaws_cloudfront.planner.synth_default_cache_behavior(cb: CacheBehaviorDesired, *, origin_id: str) → dict[str, Any]

dxaws_cloudfront.planner.synth_distribution_config(d: DistributionDesired, *, origin_domain: str, oac_id: str | None) → dict[str, Any]

Build a deterministic DistributionConfig dict (MVP).

dxaws_cloudfront.planner.synth_viewer_certificate(vc: ViewerCertificateDesired) → dict[str, Any]

dxaws_cloudfront.planner.truncate_with_hash(s: str, *, max_len: int) → str

Truncate a string to a max length, adding a stable hash suffix.

CloudFront resource names have strict length limits. This helper keeps names deterministic by adding an 8-char hash suffix when truncation is required.

We also sanitize to a conservative character set to avoid API validation issues.

dxaws_cloudfront.executor

class dxaws_cloudfront.executor.ExecuteResult(result: DistributionResult, oac_id: str | None = None)

Bases: object

Return type for executor.execute().

oac_id: str | None = None

result: DistributionResult

dxaws_cloudfront.executor.execute(provider, p: Plan) → ExecuteResult

Execute a plan and return outputs.

dxaws_cloudfront.manager

Manager for dxaws-cloudfront (Base Manager Pattern).

class dxaws_cloudfront.manager.ApplyOptions(*, max_wait_seconds: 'int' = 1800, poll_interval_seconds: 'int' = 30, emit_events: 'bool' = True)

Bases: object

emit_events: bool

max_wait_seconds: int

poll_interval_seconds: int

class dxaws_cloudfront.manager.CloudFrontManager(*, provider: 'ProviderBase | None' = None, o11y: 'O11y' = <factory>)

Bases: object

apply(plan: Plan, *, options: ApplyOptions | None = None) → DistributionResult

destroy(desired: DistributionDesired, *, options: ExecuteOptions | None = None) → ManagerResult

Destroy a distribution by converging to desired.present=False.

Pass the same desired used for creation so aliases can locate the distribution deterministically.

execute(desired: DistributionDesired, *, options: ExecuteOptions | None = None) → ManagerResult

get_current(desired: DistributionDesired) → DistributionCurrent

o11y: O11y

plan(desired: DistributionDesired, current: DistributionCurrent | None = None) → Plan

provider: ProviderBase | None

class dxaws_cloudfront.manager.ExecuteOptions(*, apply_options: 'ApplyOptions | None' = None, emit_events: 'bool' = True)

Bases: object

apply_options: ApplyOptions | None

emit_events: bool

dxaws_cloudfront.models

class dxaws_cloudfront.models.Action(type: 'ActionType', reason: 'str', payload: 'dict[str, Any]'=<factory>)

Bases: object

payload: dict[str, Any]

reason: str

type: ActionType

class dxaws_cloudfront.models.ActionType(*values)

Bases: str, Enum

CREATE_DISTRIBUTION = 'create_distribution'

CREATE_OAC = 'create_oac'

DELETE_DISTRIBUTION = 'delete_distribution'

DISABLE_DISTRIBUTION = 'disable_distribution'

TAG_DISTRIBUTION = 'tag_distribution'

UPDATE_DISTRIBUTION = 'update_distribution'

WAIT_DEPLOYED = 'wait_deployed'

class dxaws_cloudfront.models.CacheBehaviorDesired(viewer_protocol_policy: ViewerProtocolPolicy = ViewerProtocolPolicy.REDIRECT_TO_HTTPS, min_ttl: int = 0, default_ttl: int = 3600, max_ttl: int = 86400, compress: bool = True)

Bases: object

Desired default cache behavior (MVP subset).

compress: bool = True

default_ttl: int = 3600

max_ttl: int = 86400

min_ttl: int = 0

viewer_protocol_policy: ViewerProtocolPolicy = 'redirect-to-https'

class dxaws_cloudfront.models.CustomErrorResponsesDesired(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)

Bases: object

SPA-friendly error responses (optional).

enabled: bool = False

error_caching_min_ttl: int = 0

error_codes: tuple[int, ...] = (403, 404)

response_code: int = 200

response_page_path: str = '/index.html'

class dxaws_cloudfront.models.DistributionCurrent(exists: bool, name: str, id: str | None = None, arn: str | None = None, domain_name: str | None = None, status: str | None = None, 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, viewer_protocol_policy: str | None = None, 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: dict[str, str] | None = None, raw_config: dict[str, Any] | None = None, etag: str | None = None)

Bases: object

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.

acm_certificate_arn: str | None = None

aliases: tuple[str, ...] | None = None

arn: str | None = None

comment: str | None = None

default_root_object: str | None = None

domain_name: str | None = None

enabled: bool | None = None

etag: str | None = None

exists: bool

http_version: str | None = None

id: str | None = None

ipv6_enabled: bool | None = None

name: str

origin_domain_name: str | None = None

price_class: str | None = None

raw_config: dict[str, Any] | None = None

status: str | None = None

tags: dict[str, str] | None = None

using_default_certificate: bool | None = None

viewer_protocol_policy: str | None = None

class dxaws_cloudfront.models.DistributionDesired(name: str | None = None, present: bool = True, enabled: bool = True, comment: str | None = None, aliases: tuple[str, ...]=(), origin_domain: str | None = None, certificate_arn: str | None = None, origin: S3OriginDesired = <factory>, default_root_object: str = 'index.html', default_cache_behavior: CacheBehaviorDesired = <factory>, custom_errors: CustomErrorResponsesDesired = <factory>, viewer_certificate: ViewerCertificateDesired = <factory>, price_class: PriceClass = PriceClass.P100, http_version: HttpVersion = HttpVersion.HTTP2, ipv6_enabled: bool = True, tags: dict[str, str]=<factory>, oac_name: str | None = None)

Bases: object

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.

aliases: tuple[str, ...] = ()

certificate_arn: str | None = None

comment: str | None = None

custom_errors: CustomErrorResponsesDesired

default_cache_behavior: CacheBehaviorDesired

default_root_object: str = 'index.html'

property effective_certificate_arn: str | None

property effective_origin_domain: str | None

enabled: bool = True

http_version: HttpVersion = 'http2'

ipv6_enabled: bool = True

name: str | None = None

oac_name: str | None = None

origin: S3OriginDesired

origin_domain: str | None = None

present: bool = True

price_class: PriceClass = 'PriceClass_100'

tags: dict[str, str]

viewer_certificate: ViewerCertificateDesired

class dxaws_cloudfront.models.DistributionResult(name: 'str', id: 'str', arn: 'str | None', domain_name: 'str | None', hosted_zone_id: 'str | None' = None)

Bases: object

arn: str | None

domain_name: str | None

hosted_zone_id: str | None = None

id: str

name: str

class dxaws_cloudfront.models.HttpVersion(*values)

Bases: str, Enum

HTTP1_1 = 'http1.1'

HTTP2 = 'http2'

HTTP2_AND_3 = 'http2and3'

HTTP3 = 'http3'

class dxaws_cloudfront.models.ManagerResult(desired: 'DistributionDesired', current: 'DistributionCurrent', plan: 'Plan', result: 'DistributionResult', outcome: 'str')

Bases: object

current: DistributionCurrent

desired: DistributionDesired

outcome: str

plan: Plan

result: DistributionResult

class dxaws_cloudfront.models.OriginProtocolPolicy(*values)

Bases: str, Enum

HTTPS_ONLY = 'https-only'

HTTP_ONLY = 'http-only'

MATCH_VIEWER = 'match-viewer'

class dxaws_cloudfront.models.Plan(desired: 'DistributionDesired', current: 'DistributionCurrent', actions: 'list[Action]' = <factory>)

Bases: object

actions: list[Action]

current: DistributionCurrent

desired: DistributionDesired

property is_noop: bool

class dxaws_cloudfront.models.PriceClass(*values)

Bases: str, Enum

ALL = 'PriceClass_All'

P100 = 'PriceClass_100'

P200 = 'PriceClass_200'

class dxaws_cloudfront.models.S3OriginDesired(bucket_name: str | None = None, domain_name: str | None = None)

Bases: object

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

class dxaws_cloudfront.models.ViewerCertificateDesired(acm_certificate_arn: str | None = None, minimum_protocol_version: str = 'TLSv1.2_2021', ssl_support_method: str = 'sni-only')

Bases: object

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'

class dxaws_cloudfront.models.ViewerProtocolPolicy(*values)

Bases: str, Enum

ALLOW_ALL = 'allow-all'

HTTPS_ONLY = 'https-only'

REDIRECT_TO_HTTPS = 'redirect-to-https'

dxaws_cloudfront.constants

dxaws_cloudfront.constants.CLOUDFRONT_HOSTED_ZONE_ID = 'Z2FDTNDATAQYW2'

Module-wide constants.

This file defines stable names and defaults that are shared across the module. Constants should be used for values that would otherwise become repeated string literals or “magic numbers” scattered through the codebase.

dxaws_cloudfront.exceptions

Module-specific exceptions.

Keep exceptions stable once published. Callers may rely on these names for control flow and error handling.

exception dxaws_cloudfront.exceptions.ApplyError

Bases: DxAwsError

Raised when applying a plan fails.

exception dxaws_cloudfront.exceptions.DxAwsError

Bases: Exception

Base error for dxaws-cloudfront.

exception dxaws_cloudfront.exceptions.PlanError

Bases: DxAwsError

Raised when planning fails (e.g., inconsistent current state or diffing failure).

exception dxaws_cloudfront.exceptions.ProviderError

Bases: DxAwsError

Raised when the provider encounters an unexpected or unrecoverable error.

exception dxaws_cloudfront.exceptions.ValidationError

Bases: DxAwsError

Raised when caller-provided desired state or arguments are invalid.

dxaws_cloudfront.providers.base

class dxaws_cloudfront.providers.base.ProviderBase(*args, **kwargs)

Bases: Protocol

Provider interface for dxaws-cloudfront.

create_distribution(*, config: dict, tags: dict | None = None)

Create a distribution and return a ref.

delete_distribution(dist_id: str) → None

Delete a distribution.

disable_distribution(dist_id: str) → None

Disable a distribution (required before delete).

find_distribution_by_alias(alias: str)

Find a distribution by alias (CNAME).

get_distribution(dist_id: str)

Return (DistributionConfig, ETag).

get_distribution_status(dist_id: str) → str

Return distribution Status (Deployed/InProgress).

update_distribution(*, dist_id: str, if_match: str, config: dict)

Update a distribution and return a ref.

wait_deployed(dist_id: str) → None

Wait for distribution to reach Deployed.

dxaws_cloudfront.providers.aws

class dxaws_cloudfront.providers.aws.AwsProvider(*, aws)

Bases: AwsProviderBase

AWS implementation for dxaws-cloudfront.

aws: AwsSession

property cf

CloudFront boto3 client.

create_distribution(*, config: dict[str, Any], tags: dict[str, str] | None = None) → DistributionRef

Create a distribution from a raw DistributionConfig.

delete_distribution(dist_id: str) → None

Delete a distribution.

CloudFront requires the distribution to be disabled and deployed before delete.

disable_distribution(dist_id: str) → None

Disable a distribution (required before delete).

CloudFront is eventually consistent: a distribution can report Status=Deployed from a previous state even after we submit an update. We therefore wait until the config reflects Enabled=False and the distribution is Deployed.

ensure_oac(*, name: str, description: str, signing_behavior: str = 'always', signing_protocol: str = 'sigv4', origin_type: str = 's3') → OacRef

Ensure an OAC exists and return its id.

For MVP we create if missing; if CloudFront says it already exists, we re-query by name and return it.

find_distribution_by_alias(alias: str) → DistributionRef | None

find_oac_by_name(name: str) → OacRef | None

get_distribution(dist_id: str) → tuple[dict[str, Any], str]

Fetch the full distribution config and ETag.

Returns (DistributionConfig, ETag).

get_distribution_status(dist_id: str) → str

list_distributions() → list[DistributionRef]

List distributions (best-effort summary).

CloudFront is a global service; this uses the global endpoint.

list_oacs() → list[OacRef]

list_tags(resource_arn: str) → dict[str, str]

provider_ns: str

tag_resource(resource_arn: str, tags: dict[str, str]) → None

update_distribution(*, dist_id: str, if_match: str, config: dict[str, Any]) → DistributionRef

wait_deployed(dist_id: str) → None

Wait for a distribution to reach Deployed.

class dxaws_cloudfront.providers.aws.DistributionRef(id: str, arn: str | None, domain_name: str | None, status: str | None, comment: str | None)

Bases: object

Minimal reference for a CloudFront distribution.

arn: str | None

comment: str | None

domain_name: str | None

id: str

status: str | None

class dxaws_cloudfront.providers.aws.OacRef(id: str, name: str)

Bases: object

Minimal reference for an Origin Access Control.

id: str

name: str