我们从Python开源项目中，提取了以下39个代码示例，用于说明如何使用boto.jsonresponse()。
def get_response(self, doc_path, action, params, path='/', parent=None, verb='GET', list_marker=None): if not parent: parent = self response = self.make_request(action, params, path, verb) body = response.read() boto.log.debug(body) if response.status == 200: e = boto.jsonresponse.Element( list_marker=list_marker if list_marker else 'Set', pythonize_name=True) h = boto.jsonresponse.XmlHandler(e, parent) h.parse(body) inner = e for p in doc_path: inner = inner.get(p) if not inner: return None if list_marker is None else [] if isinstance(inner, list): return inner else: return dict(**inner) else: raise self.ResponseError(response.status, response.reason, body)
def delete_hosted_zone(self, hosted_zone_id): """ Delete the hosted zone specified by the given id. :type hosted_zone_id: str :param hosted_zone_id: The hosted zone's id """ uri = '/%s/hostedzone/%s' % (self.Version, hosted_zone_id) response = self.make_request('DELETE', uri) body = response.read() boto.log.debug(body) if response.status not in (200, 204): raise exception.DNSServerError(response.status, response.reason, body) e = boto.jsonresponse.Element() h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e # Health checks
def delete_health_check(self, health_check_id): """ Delete a health check :type health_check_id: str :param health_check_id: ID of the health check to delete """ uri = '/%s/healthcheck/%s' % (self.Version, health_check_id) response = self.make_request('DELETE', uri) body = response.read() boto.log.debug(body) if response.status not in (200, 204): raise exception.DNSServerError(response.status, response.reason, body) e = boto.jsonresponse.Element() h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e # Resource Record Sets
def get_change(self, change_id): """ Get information about a proposed set of changes, as submitted by the change_rrsets method. Returns a Python data structure with status information about the changes. :type change_id: str :param change_id: The unique identifier for the set of changes. This ID is returned in the response to the change_rrsets method. """ uri = '/%s/change/%s' % (self.Version, change_id) response = self.make_request('GET', uri) body = response.read() boto.log.debug(body) if response.status >= 300: raise exception.DNSServerError(response.status, response.reason, body) e = boto.jsonresponse.Element() h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e
def get_checker_ip_ranges(self): """ Return a list of Route53 healthcheck IP ranges """ uri = '/%s/checkeripranges' % self.Version response = self.make_request('GET', uri) body = response.read() boto.log.debug(body) if response.status >= 300: raise exception.DNSServerError(response.status, response.reason, body) e = boto.jsonresponse.Element(list_marker='CheckerIpRanges', item_marker=('member',)) h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e
def _make_request(self, action, params=None): """Make a call to the SES API. :type action: string :param action: The API method to use (e.g. SendRawEmail) :type params: dict :param params: Parameters that will be sent as POST data with the API call. """ ct = 'application/x-www-form-urlencoded; charset=UTF-8' headers = {'Content-Type': ct} params = params or {} params['Action'] = action for k, v in params.items(): if isinstance(v, six.text_type): # UTF-8 encode only if it's Unicode params[k] = v.encode('utf-8') response = super(SESConnection, self).make_request( 'POST', '/', headers=headers, data=urllib.parse.urlencode(params) ) body = response.read().decode('utf-8') if response.status == 200: list_markers = ('VerifiedEmailAddresses', 'Identities', 'DkimTokens', 'VerificationAttributes', 'SendDataPoints') item_markers = ('member', 'item', 'entry') e = boto.jsonresponse.Element(list_marker=list_markers, item_marker=item_markers) h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e else: # HTTP codes other than 200 are considered errors. Go through # some error handling to determine which exception gets raised, self._handle_error(response, body)
def get_response(self, action, params, path='/', parent=None, verb='POST', list_marker='Set'): """ Utility method to handle calls to IAM and parsing of responses. """ if not parent: parent = self response = self.make_request(action, params, path, verb) body = response.read() boto.log.debug(body) if response.status == 200: if body: e = boto.jsonresponse.Element(list_marker=list_marker, pythonize_name=True) h = boto.jsonresponse.XmlHandler(e, parent) h.parse(body) return e else: # Support empty responses, e.g. deleting a SAML provider # according to the official documentation. return {} else: boto.log.error('%s %s' % (response.status, response.reason)) boto.log.error('%s' % body) raise self.ResponseError(response.status, response.reason, body) # # Group methods #
def get_website_configuration_with_xml(self, headers=None): """ Returns the current status of website configuration on the bucket as unparsed XML. :rtype: 2-Tuple :returns: 2-tuple containing: 1) A dictionary containing a Python representation \ of the XML response. The overall structure is: * WebsiteConfiguration * IndexDocument * Suffix : suffix that is appended to request that \ is for a "directory" on the website endpoint * ErrorDocument * Key : name of object to serve when an error occurs 2) unparsed XML describing the bucket's website configuration """ body = self.get_website_configuration_xml(headers=headers) e = boto.jsonresponse.Element() h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e, body
def get_all_hosted_zones(self, start_marker=None, zone_list=None): """ Returns a Python data structure with information about all Hosted Zones defined for the AWS account. :param int start_marker: start marker to pass when fetching additional results after a truncated list :param list zone_list: a HostedZones list to prepend to results """ params = {} if start_marker: params = {'marker': start_marker} response = self.make_request('GET', '/%s/hostedzone' % self.Version, params=params) body = response.read() boto.log.debug(body) if response.status >= 300: raise exception.DNSServerError(response.status, response.reason, body) e = boto.jsonresponse.Element(list_marker='HostedZones', item_marker=('HostedZone',)) h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) if zone_list: e['ListHostedZonesResponse']['HostedZones'].extend(zone_list) while 'NextMarker' in e['ListHostedZonesResponse']: next_marker = e['ListHostedZonesResponse']['NextMarker'] zone_list = e['ListHostedZonesResponse']['HostedZones'] e = self.get_all_hosted_zones(next_marker, zone_list) return e
def create_health_check(self, health_check, caller_ref=None): """ Create a new Health Check :type health_check: HealthCheck :param health_check: HealthCheck object :type caller_ref: str :param caller_ref: A unique string that identifies the request and that allows failed CreateHealthCheckRequest requests to be retried without the risk of executing the operation twice. If you don't provide a value for this, boto will generate a Type 4 UUID and use that. """ if caller_ref is None: caller_ref = str(uuid.uuid4()) uri = '/%s/healthcheck' % self.Version params = {'xmlns': self.XMLNameSpace, 'caller_ref': caller_ref, 'health_check': health_check.to_xml() } xml_body = self.POSTHCXMLBody % params response = self.make_request('POST', uri, {'Content-Type': 'text/xml'}, xml_body) body = response.read() boto.log.debug(body) if response.status == 201: e = boto.jsonresponse.Element() h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e else: raise exception.DNSServerError(response.status, response.reason, body)
def get_list_health_checks(self, maxitems=None, marker=None): """ Return a list of health checks :type maxitems: int :param maxitems: Maximum number of items to return :type marker: str :param marker: marker to get next set of items to list """ params = {} if maxitems is not None: params['maxitems'] = maxitems if marker is not None: params['marker'] = marker uri = '/%s/healthcheck' % (self.Version, ) response = self.make_request('GET', uri, params=params) body = response.read() boto.log.debug(body) if response.status >= 300: raise exception.DNSServerError(response.status, response.reason, body) e = boto.jsonresponse.Element(list_marker='HealthChecks', item_marker=('HealthCheck',)) h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e
def _make_request(self, action, params=None): """Make a call to the SES API. :type action: string :param action: The API method to use (e.g. SendRawEmail) :type params: dict :param params: Parameters that will be sent as POST data with the API call. """ ct = 'application/x-www-form-urlencoded; charset=UTF-8' headers = {'Content-Type': ct} params = params or {} params['Action'] = action for k, v in params.items(): if isinstance(v, six.text_type): # UTF-8 encode only if it's Unicode params[k] = v.encode('utf-8') response = super(SESConnection, self).make_request( 'POST', '/', headers=headers, data=urllib.parse.urlencode(params) ) body = response.read().decode('utf-8') if response.status == 200: list_markers = ('VerifiedEmailAddresses', 'Identities', 'DkimTokens', 'DkimAttributes', 'VerificationAttributes', 'SendDataPoints') item_markers = ('member', 'item', 'entry') e = boto.jsonresponse.Element(list_marker=list_markers, item_marker=item_markers) h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e else: # HTTP codes other than 200 are considered errors. Go through # some error handling to determine which exception gets raised, self._handle_error(response, body)
def _make_request(self, action, params=None): """Make a call to the SES API. :type action: string :param action: The API method to use (e.g. SendRawEmail) :type params: dict :param params: Parameters that will be sent as POST data with the API call. """ ct = 'application/x-www-form-urlencoded; charset=UTF-8' headers = {'Content-Type': ct} params = params or {} params['Action'] = action for k, v in params.items(): if isinstance(v, unicode): # UTF-8 encode only if it's Unicode params[k] = v.encode('utf-8') response = super(SESConnection, self).make_request( 'POST', '/', headers=headers, data=urllib.urlencode(params) ) body = response.read() if response.status == 200: list_markers = ('VerifiedEmailAddresses', 'Identities', 'VerificationAttributes', 'SendDataPoints') item_markers = ('member', 'item', 'entry') e = boto.jsonresponse.Element(list_marker=list_markers, item_marker=item_markers) h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e else: # HTTP codes other than 200 are considered errors. Go through # some error handling to determine which exception gets raised, self._handle_error(response, body)
def get_website_configuration_with_xml(self, headers=None): """ Returns the current status of website configuration on the bucket as unparsed XML. :rtype: 2-Tuple :returns: 2-tuple containing: 1) A dictionary containing a Python representation of the XML response. The overall structure is: * WebsiteConfiguration * IndexDocument * Suffix : suffix that is appended to request that is for a "directory" on the website endpoint * ErrorDocument * Key : name of object to serve when an error occurs 2) unparsed XML describing the bucket's website configuration. """ response = self.connection.make_request('GET', self.name, query_args='website', headers=headers) body = response.read() boto.log.debug(body) if response.status != 200: raise self.connection.provider.storage_response_error( response.status, response.reason, body) e = boto.jsonresponse.Element() h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e, body
def create_hosted_zone(self, domain_name, caller_ref=None, comment=''): """ Create a new Hosted Zone. Returns a Python data structure with information about the newly created Hosted Zone. :type domain_name: str :param domain_name: The name of the domain. This should be a fully-specified domain, and should end with a final period as the last label indication. If you omit the final period, Amazon Route 53 assumes the domain is relative to the root. This is the name you have registered with your DNS registrar. It is also the name you will delegate from your registrar to the Amazon Route 53 delegation servers returned in response to this request.A list of strings with the image IDs wanted. :type caller_ref: str :param caller_ref: A unique string that identifies the request and that allows failed CreateHostedZone requests to be retried without the risk of executing the operation twice. If you don't provide a value for this, boto will generate a Type 4 UUID and use that. :type comment: str :param comment: Any comments you want to include about the hosted zone. """ if caller_ref is None: caller_ref = str(uuid.uuid4()) params = {'name': domain_name, 'caller_ref': caller_ref, 'comment': comment, 'xmlns': self.XMLNameSpace} xml_body = HZXML % params uri = '/%s/hostedzone' % self.Version response = self.make_request('POST', uri, {'Content-Type': 'text/xml'}, xml_body) body = response.read() boto.log.debug(body) if response.status == 201: e = boto.jsonresponse.Element(list_marker='NameServers', item_marker=('NameServer',)) h = boto.jsonresponse.XmlHandler(e, None) h.parse(body) return e else: raise exception.DNSServerError(response.status, response.reason, body)
