Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
_validate_not_none('queue_name', queue_name)
_validate_not_none('message_id', message_id)
_validate_not_none('pop_receipt', pop_receipt)
_validate_not_none('visibility_timeout', visibility_timeout)
request = HTTPRequest()
request.method = 'PUT'
request.host_locations = self._get_host_locations()
request.path = _get_path(queue_name, True, message_id)
request.query = {
'popreceipt': _to_str(pop_receipt),
'visibilitytimeout': _int_to_str(visibility_timeout),
'timeout': _int_to_str(timeout)
}
if content is not None:
request.body = _get_request_body(_convert_queue_message_xml(content, self.encode_function,
self.key_encryption_key))
return self._perform_request(request, _parse_queue_message_from_headers)
_validate_encryption_required(self.require_encryption, self.key_encryption_key)
_validate_not_none('queue_name', queue_name)
_validate_not_none('content', content)
request = HTTPRequest()
request.method = 'POST'
request.host_locations = self._get_host_locations()
request.path = _get_path(queue_name, True)
request.query = {
'visibilitytimeout': _to_str(visibility_timeout),
'messagettl': _to_str(time_to_live),
'timeout': _int_to_str(timeout)
}
request.body = _get_request_body(_convert_queue_message_xml(content, self.encode_function,
self.key_encryption_key))
message_list = self._perform_request(request, _convert_xml_to_queue_messages,
[self.decode_function, False,
None, None, content])
return message_list[0]