labpack/storage/aws/s3.py
def read_bucket(self, bucket_name):

        '''
            a method to retrieve properties of a bucket in s3

        :param bucket_name: string with name of bucket
        :return: dictionary with details of bucket
        '''

        title = '%s.read_bucket' % self.__class__.__name__
        
    # validate inputs
        input_fields = {
            'bucket_name': bucket_name
        }
        for key, value in input_fields.items():
            object_title = '%s(%s=%s)' % (title, key, str(value))
            self.fields.validate(value, '.%s' % key, object_title)

    # validate existence of bucket
        if not bucket_name in self.bucket_list:
            if not bucket_name in self.list_buckets():
                raise ValueError('S3 Bucket "%s" does not exist in aws region %s.' % (bucket_name, self.iam.region_name))

    # create details dictionary
        bucket_details = {
            'bucket_name': bucket_name,
            'access_control': 'private',
            'version_control': False,
            'log_destination': {},
            'lifecycle_rules': [],
            'tag_list': [],
            'notification_settings': [],
            'region_replication': {},
            'access_policy': {}
        }

    # retrieve access control details
        try:
            response = self.connection.get_bucket_acl( Bucket=bucket_name )
            if len(response['Grants']) > 1:
                log_user = 'http://acs.amazonaws.com/groups/s3/LogDelivery'
                log_delivery = False
                public_user = 'http://acs.amazonaws.com/groups/global/AllUsers'
                public_perm = []
                for grant in response['Grants']:
                    if 'URI' in grant['Grantee']:
                        if grant['Grantee']['URI'] == log_user:
                            log_delivery = True
                        if grant['Grantee']['URI'] == public_user:
                            public_perm.append(grant['Permission'])
                if public_perm:
                    if len(public_perm) > 1:
                        bucket_details['access_control'] = 'public-read-write'
                    else:
                        bucket_details['access_control'] = 'public-read'
                elif log_delivery:
                    bucket_details['access_control'] = 'log-delivery-write'
                else:
                    bucket_details['access_control'] = 'authenticated-read'
        except:
            raise AWSConnectionError(title)

    # retrieve version control details
        try:
            response = self.connection.get_bucket_versioning( Bucket=bucket_name )
            if 'Status' in response.keys():
                if response['Status'] == 'Enabled':
                    bucket_details['version_control'] = True
        except:
            raise AWSConnectionError(title)

    # retrieve log destination details
        try:
            response = self.connection.get_bucket_logging( Bucket=bucket_name )
            if 'LoggingEnabled' in response:
                res = response['LoggingEnabled']
                bucket_details['log_destination']['name'] = res['TargetBucket']
                bucket_details['log_destination']['prefix'] = ''
                if 'TargetPrefix' in res.keys():
                    bucket_details['log_destination']['prefix'] = res['TargetPrefix']
        except:
            raise AWSConnectionError(title)

    # retrieve lifecycle rules details
        try:
            response = self.connection.get_bucket_lifecycle( Bucket=bucket_name )
            for rule in response['Rules']:
                if rule['Status'] == 'Enabled':
                    details = { "prefix": rule['Prefix'] }
                    if 'Transition' in rule.keys():
                        details['longevity'] = rule['Transition']['Days']
                        details['current_version'] = True
                        details['action'] = 'archive'
                    elif 'Expiration' in rule.keys():
                        details['longevity'] = rule['Expiration']['Days']
                        details['current_version'] = True
                        details['action'] = 'delete'
                    elif 'NoncurrentVersionTransition' in rule.keys():
                        details['longevity'] = rule['NoncurrentVersionTransition']['NoncurrentDays']
                        details['current_version'] = False
                        details['action'] = 'archive'
                    elif 'NoncurrentVersionExpiration' in rule.keys():
                        details['longevity'] = rule['NoncurrentVersionExpiration']['NoncurrentDays']
                        details['current_version'] = False
                        details['action'] = 'delete'
                    bucket_details['lifecycle_rules'].append(details)
        except:
            pass

    # retrieve bucket tag details
        try:
            response = self.connection.get_bucket_tagging( Bucket=bucket_name )
            for tag in response['TagSet']:
                bucket_details['tag_list'].append(tag)
        except:
            pass

    # retrieve notification settings details
        try:
            response = self.connection.get_bucket_notification_configuration( Bucket=bucket_name)
            if 'TopicConfigurations' in response.keys():
                for notification in response['TopicConfigurations']:
                    details = {
                        'service': 'sns',
                        'arn': notification['TopicArn'],
                        'event': notification['Events'][0],
                        'filters': {}
                    }
                    if 'Filter' in notification.keys():
                        for rule in notification['Filter']['Key']['FilterRules']:
                            details['filters'][rule['Name']] = rule['Value']
                    bucket_details['notification_settings'].append(details)
            if 'QueueConfigurations' in response.keys():
                for notification in response['QueueConfigurations']:
                    details = {
                        'service': 'sqs',
                        'arn': notification['QueueArn'],
                        'event': notification['Events'][0],
                        'filters': {}
                    }
                    if 'Filter' in notification.keys():
                        for rule in notification['Filter']['Key']['FilterRules']:
                            details['filters'][rule['Name']] = rule['Value']
                    bucket_details['notification_settings'].append(details)
            if 'LambdaFunctionConfigurations' in response.keys():
                for notification in response['LambdaFunctionConfigurations']:
                    details = {
                        'service': 'lambda',
                        'arn': notification['LambdaFunctionArn'],
                        'event': notification['Events'][0],
                        'filters': {}
                    }
                    if 'Filter' in notification.keys():
                        for rule in notification['Filter']['Key']['FilterRules']:
                            details['filters'][rule['Name']] = rule['Value']
                    bucket_details['notification_settings'].append(details)
        except:
            raise AWSConnectionError(title)

    # TODO: retrieve region replication details
    #     try:
    #         response = self.connection.get_bucket_replication( Bucket=bucket_name )
    #     except:
    #         pass

    # TODO: retrieve access policy details
    #     try:
    #         response = self.connection.get_bucket_policy( Bucket=bucket_name )
    #     except:
    #         pass

        return self.iam.ingest(bucket_details)

            a method to retrieve properties of a bucket in s3

        :param bucket_name: string with name of bucket
        :return: dictionary with details of bucket
        
a method to retrieve properties of a bucket in s3
a method to retrieve properties of a bucket in s3
def read_bucket(self, bucket_name):
        title = '%s.read_bucket' % self.__class__.__name__
        input_fields = {
            'bucket_name': bucket_name
        }
        for key, value in input_fields.items():
            object_title = '%s(%s=%s)' % (title, key, str(value))
            self.fields.validate(value, '.%s' % key, object_title)
        if not bucket_name in self.bucket_list:
            if not bucket_name in self.list_buckets():
                raise ValueError('S3 Bucket "%s" does not exist in aws region %s.' % (bucket_name, self.iam.region_name))
        bucket_details = {
            'bucket_name': bucket_name,
            'access_control': 'private',
            'version_control': False,
            'log_destination': {},
            'lifecycle_rules': [],
            'tag_list': [],
            'notification_settings': [],
            'region_replication': {},
            'access_policy': {}
        }
        try:
            response = self.connection.get_bucket_acl( Bucket=bucket_name )
            if len(response['Grants']) > 1:
                log_user = 'http://acs.amazonaws.com/groups/s3/LogDelivery'
                log_delivery = False
                public_user = 'http://acs.amazonaws.com/groups/global/AllUsers'
                public_perm = []
                for grant in response['Grants']:
                    if 'URI' in grant['Grantee']:
                        if grant['Grantee']['URI'] == log_user:
                            log_delivery = True
                        if grant['Grantee']['URI'] == public_user:
                            public_perm.append(grant['Permission'])
                if public_perm:
                    if len(public_perm) > 1:
                        bucket_details['access_control'] = 'public-read-write'
                    else:
                        bucket_details['access_control'] = 'public-read'
                elif log_delivery:
                    bucket_details['access_control'] = 'log-delivery-write'
                else:
                    bucket_details['access_control'] = 'authenticated-read'
        except:
            raise AWSConnectionError(title)
        try:
            response = self.connection.get_bucket_versioning( Bucket=bucket_name )
            if 'Status' in response.keys():
                if response['Status'] == 'Enabled':
                    bucket_details['version_control'] = True
        except:
            raise AWSConnectionError(title)
        try:
            response = self.connection.get_bucket_logging( Bucket=bucket_name )
            if 'LoggingEnabled' in response:
                res = response['LoggingEnabled']
                bucket_details['log_destination']['name'] = res['TargetBucket']
                bucket_details['log_destination']['prefix'] = ''
                if 'TargetPrefix' in res.keys():
                    bucket_details['log_destination']['prefix'] = res['TargetPrefix']
        except:
            raise AWSConnectionError(title)
        try:
            response = self.connection.get_bucket_lifecycle( Bucket=bucket_name )
            for rule in response['Rules']:
                if rule['Status'] == 'Enabled':
                    details = { "prefix": rule['Prefix'] }
                    if 'Transition' in rule.keys():
                        details['longevity'] = rule['Transition']['Days']
                        details['current_version'] = True
                        details['action'] = 'archive'
                    elif 'Expiration' in rule.keys():
                        details['longevity'] = rule['Expiration']['Days']
                        details['current_version'] = True
                        details['action'] = 'delete'
                    elif 'NoncurrentVersionTransition' in rule.keys():
                        details['longevity'] = rule['NoncurrentVersionTransition']['NoncurrentDays']
                        details['current_version'] = False
                        details['action'] = 'archive'
                    elif 'NoncurrentVersionExpiration' in rule.keys():
                        details['longevity'] = rule['NoncurrentVersionExpiration']['NoncurrentDays']
                        details['current_version'] = False
                        details['action'] = 'delete'
                    bucket_details['lifecycle_rules'].append(details)
        except:
            pass
        try:
            response = self.connection.get_bucket_tagging( Bucket=bucket_name )
            for tag in response['TagSet']:
                bucket_details['tag_list'].append(tag)
        except:
            pass
        try:
            response = self.connection.get_bucket_notification_configuration( Bucket=bucket_name)
            if 'TopicConfigurations' in response.keys():
                for notification in response['TopicConfigurations']:
                    details = {
                        'service': 'sns',
                        'arn': notification['TopicArn'],
                        'event': notification['Events'][0],
                        'filters': {}
                    }
                    if 'Filter' in notification.keys():
                        for rule in notification['Filter']['Key']['FilterRules']:
                            details['filters'][rule['Name']] = rule['Value']
                    bucket_details['notification_settings'].append(details)
            if 'QueueConfigurations' in response.keys():
                for notification in response['QueueConfigurations']:
                    details = {
                        'service': 'sqs',
                        'arn': notification['QueueArn'],
                        'event': notification['Events'][0],
                        'filters': {}
                    }
                    if 'Filter' in notification.keys():
                        for rule in notification['Filter']['Key']['FilterRules']:
                            details['filters'][rule['Name']] = rule['Value']
                    bucket_details['notification_settings'].append(details)
            if 'LambdaFunctionConfigurations' in response.keys():
                for notification in response['LambdaFunctionConfigurations']:
                    details = {
                        'service': 'lambda',
                        'arn': notification['LambdaFunctionArn'],
                        'event': notification['Events'][0],
                        'filters': {}
                    }
                    if 'Filter' in notification.keys():
                        for rule in notification['Filter']['Key']['FilterRules']:
                            details['filters'][rule['Name']] = rule['Value']
                    bucket_details['notification_settings'].append(details)
        except:
            raise AWSConnectionError(title)
        return self.iam.ingest(bucket_details)
def read_bucket ( self , bucket_name ) : title = '%s.read_bucket' % self . __class__ . __name__ input_fields = { 'bucket_name' : bucket_name } for key , value in input_fields . items ( ) : object_title = '%s(%s=%s)' % ( title , key , str ( value ) ) self . fields . validate ( value , '.%s' % key , object_title ) if not bucket_name in self . bucket_list : if not bucket_name in self . list_buckets ( ) : raise ValueError ( 'S3 Bucket "%s" does not exist in aws region %s.' % ( bucket_name , self . iam . region_name ) ) bucket_details = { 'bucket_name' : bucket_name , 'access_control' : 'private' , 'version_control' : False , 'log_destination' : { } , 'lifecycle_rules' : [ ] , 'tag_list' : [ ] , 'notification_settings' : [ ] , 'region_replication' : { } , 'access_policy' : { } } try : response = self . connection . get_bucket_acl ( Bucket = bucket_name ) if len ( response [ 'Grants' ] ) > 1 : log_user = 'http://acs.amazonaws.com/groups/s3/LogDelivery' log_delivery = False public_user = 'http://acs.amazonaws.com/groups/global/AllUsers' public_perm = [ ] for grant in response [ 'Grants' ] : if 'URI' in grant [ 'Grantee' ] : if grant [ 'Grantee' ] [ 'URI' ] == log_user : log_delivery = True if grant [ 'Grantee' ] [ 'URI' ] == public_user : public_perm . append ( grant [ 'Permission' ] ) if public_perm : if len ( public_perm ) > 1 : bucket_details [ 'access_control' ] = 'public-read-write' else : bucket_details [ 'access_control' ] = 'public-read' elif log_delivery : bucket_details [ 'access_control' ] = 'log-delivery-write' else : bucket_details [ 'access_control' ] = 'authenticated-read' except : raise AWSConnectionError ( title ) try : response = self . connection . get_bucket_versioning ( Bucket = bucket_name ) if 'Status' in response . keys ( ) : if response [ 'Status' ] == 'Enabled' : bucket_details [ 'version_control' ] = True except : raise AWSConnectionError ( title ) try : response = self . connection . get_bucket_logging ( Bucket = bucket_name ) if 'LoggingEnabled' in response : res = response [ 'LoggingEnabled' ] bucket_details [ 'log_destination' ] [ 'name' ] = res [ 'TargetBucket' ] bucket_details [ 'log_destination' ] [ 'prefix' ] = '' if 'TargetPrefix' in res . keys ( ) : bucket_details [ 'log_destination' ] [ 'prefix' ] = res [ 'TargetPrefix' ] except : raise AWSConnectionError ( title ) try : response = self . connection . get_bucket_lifecycle ( Bucket = bucket_name ) for rule in response [ 'Rules' ] : if rule [ 'Status' ] == 'Enabled' : details = { "prefix" : rule [ 'Prefix' ] } if 'Transition' in rule . keys ( ) : details [ 'longevity' ] = rule [ 'Transition' ] [ 'Days' ] details [ 'current_version' ] = True details [ 'action' ] = 'archive' elif 'Expiration' in rule . keys ( ) : details [ 'longevity' ] = rule [ 'Expiration' ] [ 'Days' ] details [ 'current_version' ] = True details [ 'action' ] = 'delete' elif 'NoncurrentVersionTransition' in rule . keys ( ) : details [ 'longevity' ] = rule [ 'NoncurrentVersionTransition' ] [ 'NoncurrentDays' ] details [ 'current_version' ] = False details [ 'action' ] = 'archive' elif 'NoncurrentVersionExpiration' in rule . keys ( ) : details [ 'longevity' ] = rule [ 'NoncurrentVersionExpiration' ] [ 'NoncurrentDays' ] details [ 'current_version' ] = False details [ 'action' ] = 'delete' bucket_details [ 'lifecycle_rules' ] . append ( details ) except : pass try : response = self . connection . get_bucket_tagging ( Bucket = bucket_name ) for tag in response [ 'TagSet' ] : bucket_details [ 'tag_list' ] . append ( tag ) except : pass try : response = self . connection . get_bucket_notification_configuration ( Bucket = bucket_name ) if 'TopicConfigurations' in response . keys ( ) : for notification in response [ 'TopicConfigurations' ] : details = { 'service' : 'sns' , 'arn' : notification [ 'TopicArn' ] , 'event' : notification [ 'Events' ] [ 0 ] , 'filters' : { } } if 'Filter' in notification . keys ( ) : for rule in notification [ 'Filter' ] [ 'Key' ] [ 'FilterRules' ] : details [ 'filters' ] [ rule [ 'Name' ] ] = rule [ 'Value' ] bucket_details [ 'notification_settings' ] . append ( details ) if 'QueueConfigurations' in response . keys ( ) : for notification in response [ 'QueueConfigurations' ] : details = { 'service' : 'sqs' , 'arn' : notification [ 'QueueArn' ] , 'event' : notification [ 'Events' ] [ 0 ] , 'filters' : { } } if 'Filter' in notification . keys ( ) : for rule in notification [ 'Filter' ] [ 'Key' ] [ 'FilterRules' ] : details [ 'filters' ] [ rule [ 'Name' ] ] = rule [ 'Value' ] bucket_details [ 'notification_settings' ] . append ( details ) if 'LambdaFunctionConfigurations' in response . keys ( ) : for notification in response [ 'LambdaFunctionConfigurations' ] : details = { 'service' : 'lambda' , 'arn' : notification [ 'LambdaFunctionArn' ] , 'event' : notification [ 'Events' ] [ 0 ] , 'filters' : { } } if 'Filter' in notification . keys ( ) : for rule in notification [ 'Filter' ] [ 'Key' ] [ 'FilterRules' ] : details [ 'filters' ] [ rule [ 'Name' ] ] = rule [ 'Value' ] bucket_details [ 'notification_settings' ] . append ( details ) except : raise AWSConnectionError ( title ) return self . iam . ingest ( bucket_details )
a method to retrieve properties of a bucket in s3
a method to retrieve properties of a bucket in s3
'''
            a method to retrieve properties of a bucket in s3

        :param bucket_name: string with name of bucket
        :return: dictionary with details of bucket
        ''' # validate inputs # validate existence of bucket # create details dictionary # retrieve access control details # retrieve version control details # retrieve log destination details # retrieve lifecycle rules details # retrieve bucket tag details # retrieve notification settings details # TODO: retrieve region replication details #     try: #         response = self.connection.get_bucket_replication( Bucket=bucket_name ) #     except: #         pass # TODO: retrieve access policy details #     try: #         response = self.connection.get_bucket_policy( Bucket=bucket_name ) #     except: #         pass
dictionary with details of bucket
dictionary with details of bucket
self
bucket_name
string with name of bucket
string with name of bucket