Source code for dxaws_core.providers.aws

# dxaws_core/providers/aws.py

import re
import uuid

from dataclasses import dataclass
from ..aws_session import AwsSession
from dxaws_core.utils.tokens import token_hashed as _token_hashed


_SAFE = re.compile(r"[^a-zA-Z0-9.-]+")
def _clean(part: str) -> str:
    return _SAFE.sub("-", part).strip("-").lower()




@dataclass(frozen=False, slots=True, kw_only=True)
class AwsProviderBase:
    """
    Minimal base class for AWS providers.

    Purpose:
    - Provide consistent access to AwsSession
    - Provide a consistent idempotency token helper

    Non-goals:
    - No retries/backoff logic
    - No caching
    - No service-specific behavior
    """

    aws: AwsSession
    provider_ns: str  # e.g. "dns.route53", "ssm", "acm"



    def client(self, service: str):
        return self.aws.client(service)




    def token(self, *parts: str, prefix: str = "dxaws", rand_len: int = 6) -> str:
        """
        Generate a readable short idempotency token, automatically namespaced.

        Example:
            self.token("create_zone", "example.com")
            -> dxaws-dns.route53-create_zone-example.com-a1b2c3
        """
        cleaned = [_clean(p) for p in parts if p]
        suffix = uuid.uuid4().hex[:rand_len]
        return "-".join([prefix, self.provider_ns, *cleaned, suffix])




    def token_hashed(
        self,
        *parts: str,
        prefix: str = "dxaws",
        algo: str = "sha256",
        max_len: int = 32,
        include_ns: bool = True,
    ) -> str:
        """Generate a deterministic, length-bounded token.

        Implementation is delegated to `dxaws_core.utils.tokens.token_hashed`.
        """
        return _token_hashed(
            *parts,
            prefix=prefix,
            algo=algo,
            max_len=max_len,
            namespace=self.provider_ns if include_ns else None,
        )