Skip to content

Aws connection

AWSConnector

AWSConnector - Connects to AWS using Credentials File or IAM Role

Source code in src/super_gradients/common/aws_connection/aws_connector.py
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class AWSConnector:
    """
    AWSConnector - Connects to AWS using Credentials File or IAM Role
    """

    @staticmethod
    def __create_boto_3_session(profile_name: str):
        """
        __create_boto_3_session
            :param profile_name:
            :return:
        """
        current_class_name = __class__.__name__
        logger = logging.getLogger(current_class_name)

        try:
            try:
                if profile_name and boto3.session.Session(profile_name=profile_name).get_credentials():
                    # TRY USING A SPECIFIC PROFILE_NAME (USING A CREDENTIALS FILE)
                    logger.info("Trying to connect to AWS using Credentials File with profile_name: " + profile_name)

                    session = boto3.Session(profile_name=profile_name)
                    return session

            except ProfileNotFound as profileNotFoundException:
                logger.debug(
                    "[" + current_class_name + "] - Could not find profile name - Trying using Default Profile/IAM Role" + str(profileNotFoundException)
                )

            # TRY USING AN IAM ROLE (OR *DEFAULT* CREDENTIALS - USING A CREDENTIALS FILE)
            logger.info("Trying to connect to AWS using IAM role or Default Credentials")
            session = boto3.Session()
            return session

        except Exception as ex:
            logger.critical("[" + current_class_name + "] - Caught Exception while trying to connect to AWS Credentials Manager " + str(ex))
            return None

    @staticmethod
    def get_aws_session(profile_name: str) -> boto3.Session:
        """
        get_aws_session - Connects to AWS to retrieve an AWS Session
            :param      profile_name: The Config Profile (Environment Name in Credentials file)
            :return:    boto3 Session
        """
        current_class_name = __class__.__name__
        logger = logging.getLogger(current_class_name)

        aws_session = AWSConnector.__create_boto_3_session(profile_name=profile_name)
        if aws_session is None:
            logger.error("Failed to initiate an AWS Session")

        return aws_session

    @staticmethod
    def get_aws_client_for_service_name(profile_name: str, service_name: str) -> boto3.Session.client:
        """
        get_aws_client_for_service_name - Connects to AWS to retrieve the relevant Client
            :param      profile_name: The Config Profile (Environment Name in Credentials file)
            :param      service_name: The AWS Service name to get the Client for
            :return:    Service client instance
        """
        current_class_name = __class__.__name__
        logger = logging.getLogger(current_class_name)

        aws_session = AWSConnector.__create_boto_3_session(profile_name=profile_name)
        if aws_session is None:
            logger.error("Failed to connect to AWS client: " + str(service_name))

        return aws_session.client(service_name=service_name)

    @staticmethod
    def get_aws_resource_for_service_name(profile_name: str, service_name: str) -> boto3.Session.resource:
        """
        Connects to AWS to retrieve the relevant Resource (More functionality then Client)
            :param      profile_name: The Config Profile (Environment Name in Credentials file)
            :param      service_name: The AWS Service name to get the Client for
            :return:    Service client instance
        """
        current_class_name = __class__.__name__
        logger = logging.getLogger(current_class_name)

        aws_session = AWSConnector.__create_boto_3_session(profile_name=profile_name)
        if aws_session is None:
            logger.error("Failed to connect to AWS client: " + str(service_name))

        return aws_session.resource(service_name=service_name)

    @staticmethod
    def is_client_error(code):
        e = sys.exc_info()[1]
        if isinstance(e, ClientError) and e.response["Error"]["Code"] == code:
            return ClientError
        return type("NeverEverRaisedException", (Exception,), {})

__create_boto_3_session(profile_name) staticmethod

__create_boto_3_session :param profile_name: :return:

Source code in src/super_gradients/common/aws_connection/aws_connector.py
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@staticmethod
def __create_boto_3_session(profile_name: str):
    """
    __create_boto_3_session
        :param profile_name:
        :return:
    """
    current_class_name = __class__.__name__
    logger = logging.getLogger(current_class_name)

    try:
        try:
            if profile_name and boto3.session.Session(profile_name=profile_name).get_credentials():
                # TRY USING A SPECIFIC PROFILE_NAME (USING A CREDENTIALS FILE)
                logger.info("Trying to connect to AWS using Credentials File with profile_name: " + profile_name)

                session = boto3.Session(profile_name=profile_name)
                return session

        except ProfileNotFound as profileNotFoundException:
            logger.debug(
                "[" + current_class_name + "] - Could not find profile name - Trying using Default Profile/IAM Role" + str(profileNotFoundException)
            )

        # TRY USING AN IAM ROLE (OR *DEFAULT* CREDENTIALS - USING A CREDENTIALS FILE)
        logger.info("Trying to connect to AWS using IAM role or Default Credentials")
        session = boto3.Session()
        return session

    except Exception as ex:
        logger.critical("[" + current_class_name + "] - Caught Exception while trying to connect to AWS Credentials Manager " + str(ex))
        return None

get_aws_client_for_service_name(profile_name, service_name) staticmethod

get_aws_client_for_service_name - Connects to AWS to retrieve the relevant Client :param profile_name: The Config Profile (Environment Name in Credentials file) :param service_name: The AWS Service name to get the Client for :return: Service client instance

Source code in src/super_gradients/common/aws_connection/aws_connector.py
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@staticmethod
def get_aws_client_for_service_name(profile_name: str, service_name: str) -> boto3.Session.client:
    """
    get_aws_client_for_service_name - Connects to AWS to retrieve the relevant Client
        :param      profile_name: The Config Profile (Environment Name in Credentials file)
        :param      service_name: The AWS Service name to get the Client for
        :return:    Service client instance
    """
    current_class_name = __class__.__name__
    logger = logging.getLogger(current_class_name)

    aws_session = AWSConnector.__create_boto_3_session(profile_name=profile_name)
    if aws_session is None:
        logger.error("Failed to connect to AWS client: " + str(service_name))

    return aws_session.client(service_name=service_name)

get_aws_resource_for_service_name(profile_name, service_name) staticmethod

Connects to AWS to retrieve the relevant Resource (More functionality then Client) :param profile_name: The Config Profile (Environment Name in Credentials file) :param service_name: The AWS Service name to get the Client for :return: Service client instance

Source code in src/super_gradients/common/aws_connection/aws_connector.py
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@staticmethod
def get_aws_resource_for_service_name(profile_name: str, service_name: str) -> boto3.Session.resource:
    """
    Connects to AWS to retrieve the relevant Resource (More functionality then Client)
        :param      profile_name: The Config Profile (Environment Name in Credentials file)
        :param      service_name: The AWS Service name to get the Client for
        :return:    Service client instance
    """
    current_class_name = __class__.__name__
    logger = logging.getLogger(current_class_name)

    aws_session = AWSConnector.__create_boto_3_session(profile_name=profile_name)
    if aws_session is None:
        logger.error("Failed to connect to AWS client: " + str(service_name))

    return aws_session.resource(service_name=service_name)

get_aws_session(profile_name) staticmethod

get_aws_session - Connects to AWS to retrieve an AWS Session :param profile_name: The Config Profile (Environment Name in Credentials file) :return: boto3 Session

Source code in src/super_gradients/common/aws_connection/aws_connector.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@staticmethod
def get_aws_session(profile_name: str) -> boto3.Session:
    """
    get_aws_session - Connects to AWS to retrieve an AWS Session
        :param      profile_name: The Config Profile (Environment Name in Credentials file)
        :return:    boto3 Session
    """
    current_class_name = __class__.__name__
    logger = logging.getLogger(current_class_name)

    aws_session = AWSConnector.__create_boto_3_session(profile_name=profile_name)
    if aws_session is None:
        logger.error("Failed to initiate an AWS Session")

    return aws_session