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)
python类jsonresponse()的实例源码
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_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 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 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_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 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 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 _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 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, 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 _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
