Get notifications configured for the given bucket.

        :param bucket_name: Bucket name.
2820
def get_bucket_notification(self, bucket_name):
        
        is_valid_bucket_name(bucket_name)

        response = self._url_open(
            "GET",
            bucket_name=bucket_name,
            query={"notification": ""},
        )
        data = response.data.decode('utf-8')
        return parse_get_bucket_notification(data)
662459
def notification(
        self,
        topic_name,
        topic_project=None,
        custom_attributes=None,
        event_types=None,
        blob_name_prefix=None,
        payload_format=NONE_PAYLOAD_FORMAT,
    ):
        
        return BucketNotification(
            self,
            topic_name,
            topic_project=topic_project,
            custom_attributes=custom_attributes,
            event_types=event_types,
            blob_name_prefix=blob_name_prefix,
            payload_format=payload_format,
        )
295502
def _notifications(self):
        
        if not self.username or not self.auth_token:
            if not self.notification_warning:
                self.py3.notify_user(
                    "Github module needs username and "
                    "auth_token to check notifications."
                )
                self.notification_warning = True
            return "?"
        if self.notifications == "all" or not self.repo:
            url = self.url_api + "/notifications"
        else:
            url = self.url_api + "/repos/" + self.repo + "/notifications"
        url += "?per_page=100"
        try:
            info = self.py3.request(url, auth=(self.username, self.auth_token))
        except (self.py3.RequestException):
            return
        if info.status_code == 200:
            links = info.headers.get("Link")

            if not links:
                return len(info.json())

            last_page = 1
            for link in links.split(","):
                if 'rel="last"' in link:
                    last_url = link[link.find("<") + 1 : link.find(">")]
                    parsed = urlparse.urlparse(last_url)
                    last_page = int(urlparse.parse_qs(parsed.query)["page"][0])

            if last_page == 1:
                return len(info.json())
            try:
                last_page_info = self.py3.request(
                    last_url, auth=(self.username, self.auth_token)
                )
            except self.py3.RequestException:
                return

            return len(info.json()) * (last_page - 1) + len(last_page_info.json())

        if info.status_code == 404:
            if not self.repo_warning:
                self.py3.notify_user("Github repo cannot be found.")
                self.repo_warning = True
70610
def _empty_notification(self):
        
        sess = cherrypy.session
        username = sess.get(SESSION_KEY, None)
        if username in self.notifications:
            ret = self.notifications[username]
        else:
            ret = []
        self.notifications[username] = []
        return ret
1239997
def get_object(self, bucket, object_name):
        
        details = self._details(
            method=b"GET",
            url_context=self._url_context(bucket=bucket, object_name=object_name),
        )
        d = self._submit(self._query_factory(details))
        d.addCallback(itemgetter(1))
        return d
1252563
def _gcs_get_keys(bucket, pattern):
  
  return [obj for obj in list(bucket.objects()) if fnmatch.fnmatch(obj.metadata.name, pattern)]
799157
def notifications(self):
        """"""
        self.__init()
        items = []
        for n in self._notifications:
            if "id" in n:
                url = "%s/%s" % (self.root, n['id'])
                items.append(self.Notification(url=url,
                                               securityHandler=self._securityHandler,
                                               proxy_url=self._proxy_url,
                                               proxy_port=self._proxy_port))
        return items
533430
def get_notifications(self, login=None, **kwargs):
        

        _login = kwargs.get(
            'login',
            login or self._login
        )
        _notif_url = NOTIF_URL.format(login=_login)
        return self._request_api(url=_notif_url).json()
326675
def get_notification(self, notification_id, **params):
        """"""
        response = self._get('v2', 'notifications', notification_id, params=params)
        return self._make_api_object(response, Notification)
1196896
def get_messages(self, name):
        

        return self._loop.run_coroutine(self._client.get_messages(name))
835085
def notifications_get(input_params={}, always_retry=True, **kwargs):
    
    return DXHTTPRequest('/notifications/get', input_params, always_retry=always_retry, **kwargs)
312826
def _storage_get_keys(bucket, pattern):
  
  return [item for item in list(bucket.items()) if fnmatch.fnmatch(item.metadata.name, pattern)]
633643
def read_bucket(self, bucket_name):

        '''
            
        '''

        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)
30659
def _get_bucket_location(self, bucket_name):
        
        method = 'GET'
        url = self._endpoint_url + '/' + bucket_name + '?location='
        headers = {}
        # default for all requests.
        region = 'us-east-1'

        # Region is set override.
        if self._region:
            return self._region

        # For anonymous requests no need to get bucket location.
        if self._access_key is None or self._secret_key is None:
            return 'us-east-1'

        # Get signature headers if any.
        headers = sign_v4(method, url, region,
                          headers, self._access_key,
                          self._secret_key,
                          self._session_token,
                          None)

        response = self._http.urlopen(method, url,
                                      body=None,
                                      headers=headers)

        if self._trace_output_stream:
            dump_http(method, url, headers, response,
                      self._trace_output_stream)

        if response.status != 200:
            raise ResponseError(response, method, bucket_name).get_exception()

        location = parse_location_constraint(response.data)
        # location is empty for 'US standard region'
        if not location:
            return 'us-east-1'
        # location can be 'EU' convert it to meaningful 'eu-west-1'
        if location == 'EU':
            return 'eu-west-1'
        return location
910206
def update_notifications(factory_id, cloud_notification_settings_body, opts = {})
      data, _status_code, _headers = update_notifications_with_http_info(factory_id, cloud_notification_settings_body, opts)
      return data
    end
1066452
def instrument(name, payload = {}, &block)
      suffixed_name = /\.middleman$/.match?(name) ? name.dup : "#{name}.middleman"
      ::ActiveSupport::Notifications.instrument(suffixed_name, payload, &block)
    end
1018247
def check_bucket_exists(self, bucket: str) -> bool:
        
        bucket_obj = self.gcp_client.bucket(bucket)  # type: Bucket
        return bucket_obj.exists()