Source code for dxaws_accounts.providers.aws

from __future__ import annotations

from dataclasses import asdict
from typing import Any, Iterable, List, Optional

from dxaws_core.providers.aws import AwsProviderBase

from dxaws_accounts.models import AwsOrgAccount




class AwsProvider(AwsProviderBase):
    """AWS implementation for dxaws-accounts.

    This provider is intentionally thin: it only knows how to talk to AWS
    Organizations and convert AWS responses into our internal models.
    """

    def __init__(self, *, aws) -> None:
        # provider_ns is derived from the module name, e.g. "cloudfront", "acm", "dns.route53"
        super().__init__(aws=aws, provider_ns="accounts.organizations")

    # ---------------------------------------------------------------------
    # Internal helpers
    # ---------------------------------------------------------------------

    def _client(self, service_name: str):
        """Best-effort client accessor compatible with different aws wrappers."""
        if hasattr(self.aws, "client") and callable(getattr(self.aws, "client")):
            return self.aws.client(service_name)

        # Common alternative: aws.session.client(...)
        sess = getattr(self.aws, "session", None)
        if sess is not None and hasattr(sess, "client") and callable(getattr(sess, "client")):
            return sess.client(service_name)

        raise TypeError(
            "aws object must provide .client(service_name) or .session.client(service_name)"
        )

    @staticmethod
    def _to_org_account(d: dict[str, Any]) -> AwsOrgAccount:
        """Convert an AWS Organizations Account dict into our AwsOrgAccount model."""
        # AWS returns keys like: Id, Arn, Email, Name, Status, JoinedMethod, JoinedTimestamp
        return AwsOrgAccount(
            id=str(d.get("Id") or ""),
            name=str(d.get("Name") or ""),
            email=str(d.get("Email") or ""),
            status=str(d.get("Status") or ""),
        )

    # ---------------------------------------------------------------------
    # Public API used by the Accounts manager
    # ---------------------------------------------------------------------



    def list_org_accounts(self) -> list[AwsOrgAccount]:
        """List all accounts visible via AWS Organizations."""
        org = self._client("organizations")

        accounts: list[AwsOrgAccount] = []
        paginator = getattr(org, "get_paginator", None)
        if callable(paginator):
            for page in org.get_paginator("list_accounts").paginate():
                for a in page.get("Accounts", []) or []:
                    accounts.append(self._to_org_account(a))
            return accounts

        # Fallback (non-paginated)
        resp = org.list_accounts()
        for a in resp.get("Accounts", []) or []:
            accounts.append(self._to_org_account(a))
        return accounts




    def describe_org_account(self, *, account_id: str) -> AwsOrgAccount:
        """Fetch a single account from AWS Organizations."""
        if not account_id:
            raise ValueError("account_id is required")

        org = self._client("organizations")
        resp = org.describe_account(AccountId=account_id)
        acct = resp.get("Account") or {}
        return self._to_org_account(acct)


    # Convenience (sometimes handy for debugging / o11y)


    def list_org_accounts_dict(self) -> list[dict[str, Any]]:
        """Return org accounts as plain dicts (useful for logging/tests)."""
        return [asdict(a) for a in self.list_org_accounts()]