Profile manager

async_boto.core.profile_manager

AWSProfileManager

AWSProfileManager(logger=None)

Handles AWS profiles similar to boto3 but without dependencies. Supports standard credential files and SSO authentication.

Source code in async_boto/core/profile_manager.py

def __init__(self, logger=None):
    self._profiles = {}
    self._config_files = []
    self._loaded = False
    self._sso_cache = {}

    # Setup logging
    self.logger = logger or logging.getLogger("AWSProfileManager")
    if not self.logger.handlers:
        handler = logging.StreamHandler()
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

logger instance-attribute

logger = logger or getLogger('AWSProfileManager')

get_profile

get_profile(profile_name=None)

Get a specific AWS profile.

Args: profile_name: The name of the profile to get. If None, will try to get the default profile or use AWS environment variables.

Returns: Profile data as a dictionary

Raises: ValueError: If the profile doesn't exist or no default can be found

Source code in async_boto/core/profile_manager.py

def get_profile(self, profile_name: str | None = None) -> dict[str, Any]:
    """
    Get a specific AWS profile.

    Args:
        profile_name: The name of the profile to get.
                      If None, will try to get the default profile
                      or use AWS environment variables.

    Returns:
        Profile data as a dictionary

    Raises:
        ValueError: If the profile doesn't exist or no default can be found
    """
    self._load_profiles()

    # If no profile specified, check environment variables
    if profile_name is None:
        profile_name = os.environ.get("AWS_PROFILE")

    # If still no profile, use default
    if profile_name is None:
        profile_name = "default"

    # Check if profile exists
    if profile_name not in self._profiles:
        raise ValueError(f"Profile '{profile_name}' not found in AWS config files")

    return self._profiles[profile_name].copy()

get_credentials

get_credentials(profile_name=None)

Get AWS credentials from profile or environment variables

Args: profile_name: The name of the profile to get credentials from. If None, will try to get the default profile or use AWS environment variables.

Returns: Dictionary with aws_access_key_id, aws_secret_access_key, aws_session_token (if available), and region_name (if available)

Source code in async_boto/core/profile_manager.py

def get_credentials(self, profile_name: str | None = None) -> dict[str, str]:
    """
    Get AWS credentials from profile or environment variables

    Args:
        profile_name: The name of the profile to get credentials from.
                      If None, will try to get the default profile
                      or use AWS environment variables.

    Returns:
        Dictionary with aws_access_key_id, aws_secret_access_key,
        aws_session_token (if available), and region_name (if available)
    """
    # Check environment variables first (they take precedence)
    credentials = {}

    # Check for access key in environment
    access_key = os.environ.get("AWS_ACCESS_KEY_ID")
    if access_key:
        credentials["aws_access_key_id"] = access_key

        # If access key is in environment, also check for secret key
        secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY")
        if secret_key:
            credentials["aws_secret_access_key"] = secret_key

            # If both keys found, also check for session token
            session_token = os.environ.get("AWS_SESSION_TOKEN")
            if session_token:
                credentials["aws_session_token"] = session_token

    # Check for region in environment
    region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")
    if region:
        credentials["region_name"] = region

    # If we have all required credentials from environment, return them
    if (
        "aws_access_key_id" in credentials
        and "aws_secret_access_key" in credentials
    ):
        return credentials

    # Otherwise, try to get from profile
    try:
        profile = self.get_profile(profile_name)

        # Store profile name for reference
        profile["__name__"] = profile_name or "default"

        # Check if this is an SSO profile
        if "sso_start_url" in profile:
            self.logger.info(
                f"Using SSO authentication for profile {profile['__name__']}"
            )
            # For synchronous credential retrieval, use subprocess with AWS CLI
            return self._get_sso_credentials_sync(profile)

        # Map profile keys to credential keys
        key_mapping = {
            "aws_access_key_id": ["aws_access_key_id", "access_key_id"],
            "aws_secret_access_key": ["aws_secret_access_key", "secret_access_key"],
            "aws_session_token": ["aws_session_token", "session_token"],
            "region_name": ["region"],
        }

        # Some profiles may use different key names
        for cred_key, profile_keys in key_mapping.items():
            for profile_key in profile_keys:
                if profile_key in profile and cred_key not in credentials:
                    credentials[cred_key] = profile[profile_key]

        # Handle assume role if specified in profile
        if "role_arn" in profile:
            # This would involve making an STS call to assume the role
            # For now, we'll raise an error as this requires additional
            # implementation
            raise NotImplementedError(
                "Assuming roles from profiles requires STS implementation"
            )

    except ValueError:
        # If no profile found and no environment credentials, just return empty dict
        pass

    # Validate we have the minimum required credentials
    if not (
        "aws_access_key_id" in credentials
        and "aws_secret_access_key" in credentials
    ):
        raise ValueError(
            "No valid AWS credentials found in environment or profiles"
        )

    return credentials

get_credentials_async async

get_credentials_async(profile_name=None, client_func=None)

Get AWS credentials asynchronously

Args: profile_name: The name of the profile to get credentials from client_func: Function to create an async client

Returns: Credentials dictionary

Source code in async_boto/core/profile_manager.py

async def get_credentials_async(
    self, profile_name: str | None = None, client_func=None
) -> dict[str, str]:
    """
    Get AWS credentials asynchronously

    Args:
        profile_name: The name of the profile to get credentials from
        client_func: Function to create an async client

    Returns:
        Credentials dictionary
    """
    # Check environment variables first
    credentials = {}

    # Check for access key in environment
    access_key = os.environ.get("AWS_ACCESS_KEY_ID")
    if access_key:
        credentials["aws_access_key_id"] = access_key

        # If access key is in environment, also check for secret key
        secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY")
        if secret_key:
            credentials["aws_secret_access_key"] = secret_key

            # If both keys found, also check for session token
            session_token = os.environ.get("AWS_SESSION_TOKEN")
            if session_token:
                credentials["aws_session_token"] = session_token

    # Check for region in environment
    region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")
    if region:
        credentials["region_name"] = region

    # If we have all required credentials from environment, return them
    if (
        "aws_access_key_id" in credentials
        and "aws_secret_access_key" in credentials
    ):
        return credentials

    # Otherwise, try to get from profile
    try:
        profile = self.get_profile(profile_name)

        # Store profile name for reference
        profile["__name__"] = profile_name or "default"

        # Check if this is an SSO profile
        if "sso_start_url" in profile and client_func:
            self.logger.info(
                f"Using SSO authentication for profile {profile['__name__']}"
            )
            return await self._get_sso_credentials_async(profile, client_func)
        elif "sso_start_url" in profile:
            self.logger.info(
                f"Using SSO authentication for profile {profile['__name__']}"
            )
            # Fall back to sync method if client_func not provided
            return self._get_sso_credentials_sync(profile)

        # Map profile keys to credential keys
        key_mapping = {
            "aws_access_key_id": ["aws_access_key_id", "access_key_id"],
            "aws_secret_access_key": ["aws_secret_access_key", "secret_access_key"],
            "aws_session_token": ["aws_session_token", "session_token"],
            "region_name": ["region"],
        }

        # Some profiles may use different key names
        for cred_key, profile_keys in key_mapping.items():
            for profile_key in profile_keys:
                if profile_key in profile and cred_key not in credentials:
                    credentials[cred_key] = profile[profile_key]

        # Handle assume role if specified in profile
        if "role_arn" in profile:
            raise NotImplementedError(
                "Assuming roles from profiles requires STS implementation"
            )

    except ValueError:
        # If no profile found and no environment credentials, just return empty dict
        pass

    # Validate we have the minimum required credentials
    if not (
        "aws_access_key_id" in credentials
        and "aws_secret_access_key" in credentials
    ):
        raise ValueError(
            "No valid AWS credentials found in environment or profiles"
        )

    return credentials

list_profiles

list_profiles()

List all available AWS profiles

Returns: List of profile names

Source code in async_boto/core/profile_manager.py

def list_profiles(self) -> list[str]:
    """
    List all available AWS profiles

    Returns:
        List of profile names
    """
    self._load_profiles()
    return list(self._profiles.keys())

get_available_regions

get_available_regions(service_name='s3')

Get available regions for a service. This is a simplified implementation that returns common regions.

Args: service_name: AWS service name (ignored in this implementation)

Returns: List of region names

Source code in async_boto/core/profile_manager.py

def get_available_regions(self, service_name: str = "s3") -> list[str]:
    """
    Get available regions for a service.
    This is a simplified implementation that returns common regions.

    Args:
        service_name: AWS service name (ignored in this implementation)

    Returns:
        List of region names
    """
    # This is a simplified list of common regions
    # A full implementation would need to check AWS's region data
    return [
        "us-east-1",
        "us-east-2",
        "us-west-1",
        "us-west-2",
        "ca-central-1",
        "eu-west-1",
        "eu-west-2",
        "eu-west-3",
        "eu-central-1",
        "eu-north-1",
        "ap-northeast-1",
        "ap-northeast-2",
        "ap-northeast-3",
        "ap-southeast-1",
        "ap-southeast-2",
        "ap-south-1",
        "sa-east-1",
    ]