API Reference

This page is generated directly from the Python source using Sphinx autodoc. It reflects the current public contract of the dxaws-s3 module.

The API is intentionally small and centered around the declarative convergence flow:

Desired → Plan → Apply → Result

Manager

The Manager - Public module facade for dxaws-s3.

class dxaws_s3.manager.ApplyOptions(emit_legacy_events: 'bool' = True)

Bases: object

emit_legacy_events: bool = True

class dxaws_s3.manager.ExecuteOptions(plan_options: 'PlanOptions | None' = None, apply_options: 'ApplyOptions | None' = None, emit_legacy_events: 'bool' = True)

Bases: object

apply_options: ApplyOptions | None = None

emit_legacy_events: bool = True

plan_options: PlanOptions | None = None

class dxaws_s3.manager.S3Manager(provider: 'S3Provider', plan_options: 'PlanOptions' = <factory>, o11y: 'O11y' = <factory>)

Bases: object

apply(desired: S3BucketDesired, *, options: ApplyOptions | None = None) → S3BucketResult

apply_plan(p: S3BucketPlan, *, options: ApplyOptions | None = None) → S3BucketResult

execute(desired: S3BucketDesired, *, options: ExecuteOptions | None = None) → S3ManagerResult

get_current(desired: S3BucketDesired) → S3BucketCurrent

o11y: O11y

plan(desired: S3BucketDesired, current: S3BucketCurrent | None = None, *, options: PlanOptions | None = None) → S3BucketPlan

plan_options: PlanOptions

provider: S3Provider

class dxaws_s3.manager.S3ManagerResult(desired: 'S3BucketDesired', current: 'S3BucketCurrent', plan: 'S3BucketPlan', result: 'S3BucketResult | None', outcome: 'str')

Bases: object

current: S3BucketCurrent

desired: S3BucketDesired

outcome: str

plan: S3BucketPlan

result: S3BucketResult | None

Planner

Planner for dxaws-s3.

Computes a deterministic list of actions by comparing desired vs current state.

class dxaws_s3.planner.PlanOptions(strict_cloudfront: bool = False)

Bases: object

Planner behavior flags.

strict_cloudfront: bool = False

dxaws_s3.planner.plan(desired: S3BucketDesired, current: S3BucketCurrent, *, options: PlanOptions | None = None) → S3BucketPlan

Executor

Executor for dxaws-s3.

dxaws_s3.executor.execute(plan: S3BucketPlan, provider: S3Provider, *, on_op_start=None, on_op_done=None) → S3BucketResult

Models

dxaws-s3 models.

Desired -> Current -> Plan(operations) -> Result

class dxaws_s3.models.ActionType(*values)

Bases: str, Enum

CREATE_BUCKET = 'create_bucket'

PUT_BUCKET_POLICY = 'put_bucket_policy'

PUT_ENCRYPTION = 'put_encryption'

PUT_PUBLIC_ACCESS_BLOCK = 'put_public_access_block'

PUT_TAGS = 'put_tags'

PUT_VERSIONING = 'put_versioning'

class dxaws_s3.models.BucketPurpose(*values)

Bases: str, Enum

CLOUDFRONT_ORIGIN = 'cloudfront_origin'

EncryptionMode = "typing.Literal['SSE-S3', 'SSE-KMS', 'NONE']"

GENERAL = 'general'

class dxaws_s3.models.CloudFrontOriginDesired(distribution_arn: 'str | None' = None, distribution_arns: 'list[str]' = <factory>, oai_canonical_user_id: 'str | None' = None)

Bases: object

distribution_arn: str | None = None

distribution_arns: list[str]

oai_canonical_user_id: str | None = None

class dxaws_s3.models.OperationType(*values)

Bases: str, Enum

create = 'create'

delete = 'delete'

noop = 'noop'

update = 'update'

class dxaws_s3.models.S3BucketCurrent(exists: 'bool', name: 'str', arn: 'str | None' = None, regional_domain_name: 'str | None' = None, tags: 'dict[str, str]'=<factory>, block_public_access: 'bool | None' = None, encryption: 'EncryptionMode | None' = None, kms_key_arn: 'str | None' = None, versioning: 'bool | None' = None, bucket_policy_json: 'str | None' = None)

Bases: object

arn: str | None = None

block_public_access: bool | None = None

bucket_policy_json: str | None = None

encryption: EncryptionMode | None = None

exists: bool

kms_key_arn: str | None = None

name: str

regional_domain_name: str | None = None

tags: dict[str, str]

versioning: bool | None = None

class dxaws_s3.models.S3BucketDesired(name: 'str', region: 'str | None' = None, purpose: 'BucketPurpose' = <BucketPurpose.GENERAL: 'general'>, tags: 'dict[str, str]'=<factory>, block_public_access: 'bool' = True, enforce_tls: 'bool' = True, encryption: 'EncryptionMode' = 'SSE-S3', kms_key_arn: 'str | None' = None, versioning: 'bool' = False, lifecycle_rules: 'list[dict[str, Any]]'=<factory>, cors_rules: 'list[dict[str, Any]]'=<factory>, cloudfront: 'CloudFrontOriginDesired | None' = None)

Bases: object

block_public_access: bool = True

cloudfront: CloudFrontOriginDesired | None = None

property cloudfront_distribution_arns: list[str]

cors_rules: list[dict[str, Any]]

encryption: EncryptionMode = 'SSE-S3'

enforce_tls: bool = True

kms_key_arn: str | None = None

lifecycle_rules: list[dict[str, Any]]

name: str

purpose: BucketPurpose = 'general'

region: str | None = None

tags: dict[str, str]

versioning: bool = False

class dxaws_s3.models.S3BucketOperation(op_type: 'OperationType', type: 'ActionType', target: 'str', reason: 'str', payload: 'dict[str, Any]'=<factory>)

Bases: object

static for_bucket(*, op_type: OperationType, type: ActionType, bucket_name: str, reason: str, payload: dict[str, Any] | None = None) → S3BucketOperation

op_type: OperationType

payload: dict[str, Any]

reason: str

target: str

type: ActionType

class dxaws_s3.models.S3BucketPlan(desired: 'S3BucketDesired', current: 'S3BucketCurrent', operations: 'list[S3BucketOperation]', metadata: 'dict[str, Any]'=<factory>)

Bases: object

current: S3BucketCurrent

desired: S3BucketDesired

property is_noop: bool

metadata: dict[str, Any]

operations: list[S3BucketOperation]

class dxaws_s3.models.S3BucketResult(name: 'str', arn: 'str', regional_domain_name: 'str')

Bases: object

arn: str

name: str

regional_domain_name: str

Constants

dxaws-s3 constants.

Keep this intentionally small: defaults and well-known keys.

Exceptions

dxaws-s3 exceptions.

exception dxaws_s3.exceptions.BucketNameError

Bases: DxAwsS3Error

Bucket name is invalid for S3.

exception dxaws_s3.exceptions.DxAwsS3Error

Bases: Exception

Base error for dxaws-s3.

exception dxaws_s3.exceptions.InvalidDesiredStateError

Bases: DxAwsS3Error

Desired state is invalid or internally inconsistent.

exception dxaws_s3.exceptions.MissingCloudFrontInfoError

Bases: DxAwsS3Error

Reserved - CloudFront-origin bucket was requested but required CloudFront info was missing.

Providers

Base Interface

Provider interfaces for dxaws-s3.

Providers translate between AWS APIs (or other backends) and dxaws-s3 models.

This file defines the stable provider contract used by the manager/executor.

class dxaws_s3.providers.base.S3Provider(*args, **kwargs)

Bases: Protocol

create_bucket(desired: S3BucketDesired) → None

Create the bucket.

delete_all_objects(name: str) → None

Delete all objects (and versions, if present) from the bucket.

delete_bucket(name: str) → None

Delete the bucket (must be empty first).

destroy_bucket(name: str) → None

Best-effort bucket destroy: empty then delete.

get_current(name: str) → S3BucketCurrent

Return the current observed state of the bucket.

get_outputs(name: str) → S3BucketResult

Return outputs for a bucket that exists.

put_bucket_policy(name: str, policy_json: str) → None

Put/replace the bucket policy.

put_encryption(name: str, encryption: str, kms_key_arn: str | None) → None

Configure default bucket encryption.

put_public_access_block(name: str, enabled: bool) → None

Configure the bucket public access block.

put_tags(name: str, tags: dict[str, str]) → None

Put/replace bucket tags.

put_versioning(name: str, enabled: bool) → None

Configure bucket versioning.

AWS Implementation

AWS provider for dxaws-s3.

This provider is responsible for translating boto3 state into our AWS-agnostic S3BucketCurrent model, and applying changes requested by the executor.

We implement get_current() first because it’s the most subtle/important part of convergence.

class dxaws_s3.providers.aws.AwsS3Provider(*, region: str | None = None, aws: AwsSession | None = None)

Bases: S3Provider

create_bucket(desired: S3BucketDesired) → None

Create the bucket.

delete_all_objects(name: str) → None

Delete all objects (and versions, if present) from the bucket.

delete_bucket(name: str) → None

Delete the bucket (must be empty first).

destroy_bucket(name: str) → None

Best-effort bucket destroy: empty then delete.

get_current(name: str) → S3BucketCurrent

Return the current observed state of the bucket.

get_outputs(name: str) → S3BucketResult

Return outputs for a bucket that exists.

put_bucket_policy(name: str, policy_json: str) → None

Put/replace the bucket policy.

put_encryption(name: str, encryption: str, kms_key_arn: str | None) → None

Configure default bucket encryption.

put_public_access_block(name: str, enabled: bool) → None

Configure the bucket public access block.

put_tags(name: str, tags: dict[str, str]) → None

Put/replace bucket tags.

put_versioning(name: str, enabled: bool) → None

Configure bucket versioning.