def s3()

in luigi/contrib/s3.py [0:0]

• 39 lines of code
• 12 McCabe index (conditional complexity)

    def s3(self):
        # only import boto3 when needed to allow top-lvl s3 module import
        import boto3

        options = dict(self._options)

        if self._s3:
            return self._s3

        aws_access_key_id = options.get('aws_access_key_id')
        aws_secret_access_key = options.get('aws_secret_access_key')

        # Removing key args would break backwards compatibility
        role_arn = options.get('aws_role_arn')
        role_session_name = options.get('aws_role_session_name')

        # In case the aws_session_token is provided use it
        aws_session_token = options.get('aws_session_token')

        if role_arn and role_session_name:
            sts_client = boto3.client('sts')
            assumed_role = sts_client.assume_role(RoleArn=role_arn,
                                                  RoleSessionName=role_session_name)
            aws_secret_access_key = assumed_role['Credentials'].get(
                'SecretAccessKey')
            aws_access_key_id = assumed_role['Credentials'].get('AccessKeyId')
            aws_session_token = assumed_role['Credentials'].get('SessionToken')
            logger.debug('using aws credentials via assumed role {} as defined in luigi config'
                         .format(role_session_name))

        for key in ['aws_access_key_id', 'aws_secret_access_key',
                    'aws_role_session_name', 'aws_role_arn', 'aws_session_token']:
            if key in options:
                options.pop(key)

        # At this stage, if no credentials provided, boto3 would handle their resolution for us
        # For finding out about the order in which it tries to find these credentials
        # please see here details
        # http://boto3.readthedocs.io/en/latest/guide/configuration.html#configuring-credentials

        if not (aws_access_key_id and aws_secret_access_key):
            logger.debug('no credentials provided, delegating credentials resolution to boto3')

        try:
            self._s3 = boto3.resource('s3',
                                      aws_access_key_id=aws_access_key_id,
                                      aws_secret_access_key=aws_secret_access_key,
                                      aws_session_token=aws_session_token,
                                      **options)
        except TypeError as e:
            logger.error(e.args[0])
            if 'got an unexpected keyword argument' in e.args[0]:
                raise DeprecatedBotoClientException(
                    "Now using boto3. Check that you're passing the correct arguments")
            raise

        return self._s3