wip

local-app/python-tools/gfl-resource-actions/aws_resource_management/account_manager.py

@@ -0,0 +1,111 @@
+"""Account management for AWS Resource Management."""
+
+from typing import Dict, List, Optional
+
+from aws_resource_management.utils import (
+    ensure_valid_account_name,
+    get_account_list,
+    get_credentials,
+    get_organization_accounts,
+    logger,
+)
+
+
+class AccountManager:
+    """Manages AWS account discovery and filtering."""
+
+    def __init__(
+        self,
+        profile_name: Optional[str] = None,
+        use_profiles: bool = False,
+        max_concurrent_accounts: int = 3
+    ) -> None:
+        """Initialize the account manager.
+        
+        Args:
+            profile_name: AWS profile name
+            use_profiles: Whether to use AWS profiles
+            max_concurrent_accounts: Maximum number of concurrent account operations
+        """
+        self.profile_name = profile_name
+        self.use_profiles = use_profiles
+        self.MAX_CONCURRENT_ACCOUNTS = max_concurrent_accounts
+
+    def get_accounts(self) -> List[Dict[str, str]]:
+        """Get list of accounts from the current organization or profiles."""
+        if self.use_profiles:
+            logger.info("Using AWS SSO profiles for account discovery")
+            return get_organization_accounts()
+        else:
+            logger.info("Using AWS Organizations API for account discovery")
+            return get_account_list()
+
+    def pre_validate_accounts(
+        self, accounts: List[Dict[str, str]]
+    ) -> List[Dict[str, str]]:
+        """Pre-validate which accounts we can authenticate with."""
+        valid_accounts = []
+        for account in accounts:
+            account_id = account.get("account_id")
+            try:
+                credentials = get_credentials(account_id, self.profile_name)
+                if credentials:
+                    valid_accounts.append(account)
+                else:
+                    logger.debug(
+                        f"No credentials available for account {account_id}, skipping"
+                    )
+            except Exception as e:
+                logger.debug(f"Cannot authenticate with account {account_id}: {str(e)}")
+        return valid_accounts
+
+    def filter_accounts(
+        self, 
+        accounts: List[Dict[str, str]], 
+        exclude_accounts: List[str],
+        specific_account: Optional[str] = None
+    ) -> List[Dict[str, str]]:
+        """Filter accounts based on exclusions and specific account."""
+        filtered_accounts = accounts
+
+        # Filter for specific account if provided
+        if specific_account:
+            filtered_accounts = [
+                acc for acc in filtered_accounts 
+                if acc.get("account_id") == specific_account
+            ]
+            if not filtered_accounts:
+                logger.error(f"Specified account {specific_account} not found or not accessible")
+                return []
+
+        # Apply exclusions
+        return [
+            acc for acc in filtered_accounts 
+            if acc.get("account_id") not in exclude_accounts
+        ]
+
+    def normalize_account_names(
+        self, accounts: List[Dict[str, str]]
+    ) -> List[Dict[str, str]]:
+        """Ensure all accounts have valid names."""
+        for account in accounts:
+            account_id = account.get("account_id")
+            account_name = ensure_valid_account_name(
+                account_id, account.get("account_name")
+            )
+            account["account_name"] = account_name
+        return accounts
+
+    def log_account_summary(
+        self, all_accounts: List[Dict[str, str]], 
+        valid_accounts: List[Dict[str, str]],
+        exclude_accounts: List[str]
+    ) -> None:
+        """Log summary of accounts being processed."""
+        logger.info(f"Found {len(valid_accounts)} accessible accounts to process")
+        if len(all_accounts) > len(valid_accounts):
+            logger.info(
+                f"Skipped {len(all_accounts) - len(valid_accounts)} inaccessible accounts"
+            )
+        if exclude_accounts:
+            logger.info(f"Excluding {len(exclude_accounts)} accounts")