def on_post(self, req, resp):
"""
Accept twilio POST that has message delivery status, and pass it
to iris-api
"""
try:
re = self.iclient.post(self.endpoint, req.context['body'], raw=True)
except MaxRetryError:
logger.exception('Failed posting data to iris-api')
raise falcon.HTTPInternalServerError('Internal Server Error', 'API call failed')
if re.status is not 204:
logger.error('Invalid response from API for delivery status update: %s', re.status)
raise falcon.HTTPBadRequest('Likely bad params passed', 'Invalid response from API')
resp.status = falcon.HTTP_204
python类HTTPBadRequest()的实例源码
def on_post(self, req, resp):
if req.get_param('output') == 'parquet-mr':
json_schema = req.bounded_stream.read()
file = tempfile.NamedTemporaryFile(delete=False, suffix='.json')
try:
file.write(json_schema)
file.close()
output = subprocess.check_output(
[JSONSCHEMA_PARQUET_PATH,
'parquet',
'--deref',
file.name],
)
resp.body = json.dumps({'parquet-mr': output.decode()})
finally:
os.remove(file.name)
else:
raise falcon.HTTPBadRequest(
'Bad request', 'Invalid or missing output parameter.')
def from_json(self, datastring):
try:
return json.loads(datastring, object_hook=self._sanitizer)
except ValueError:
msg = 'Malformed JSON in request body.'
raise falcon.HTTPBadRequest(
title='Malformed JSON', description=msg)
def default(self, request):
if self.has_body(request):
body = request.stream.read(request.content_length)
return {'body': self.from_json(body.decode("utf-8"))}
else:
raise falcon.HTTPBadRequest('Empty request body',
'A valid JSON doc is required')
def get_queryset(self, req, resp):
query = self.get_base_query(req, resp)
conditions = {}
if 'doc' in req.context:
conditions = dict(req.context['doc'])
# ignore any special params except SEARCH and ORDER
for param in self.get_special_params():
conditions.pop(param, None)
conditions.update(req.params)
if self.PARAM_SEARCH in conditions:
search = conditions.pop(self.PARAM_SEARCH)
try:
conditions.update(json.loads(search) if isinstance(search, str) else search)
except ValueError:
raise HTTPBadRequest('Invalid attribute',
'Value of {} filter attribute is invalid'.format(self.PARAM_SEARCH))
order = conditions.pop(self.PARAM_ORDER, None)
if not order:
return self.filter_by(query, conditions)
if isinstance(order, str):
if (order[0] == '{' and order[-1] == '}') or (order[0] == '[' and order[-1] == ']'):
try:
order = json.loads(order)
except ValueError:
# not valid json, ignore and try to parse as an ordinary list of attributes
pass
if not isinstance(order, list) and not isinstance(order, dict):
order = [order]
order_expressions = self._build_order_expressions(order)
if order_expressions:
query = query.sort(*order_expressions)
return self.filter_by(query, conditions, order_criteria=order_expressions)
def _parse_logical_op(self, arg, value, default_op, relationships):
"""
:param arg: condition name
:type arg: str
:param value: condition value
:type value: dict | list
:param default_op: a default operator to join all filter expressions
:type default_op: function
:param relationships: a dict with all joins to apply, describes current state in recurrent calls
:type relationships: dict
:return: expressions list
:rtype: list
"""
if isinstance(value, dict):
return self._build_filter_expressions(value, default_op, relationships)
if not isinstance(value, list):
raise HTTPBadRequest('Invalid attribute', 'Filter attribute {} is invalid'.format(arg))
expressions = []
for subconditions in value:
if not isinstance(subconditions, dict):
raise HTTPBadRequest('Invalid attribute', 'Filter attribute {} is invalid'.format(arg))
subexpressions = self._build_filter_expressions(subconditions, and_, relationships)
if subexpressions is not None:
expressions.append(subexpressions)
result = None
if len(expressions) > 1:
result = default_op(*expressions) if default_op != not_ else not_(and_(*expressions))
elif len(expressions) == 1:
result = expressions[0] if default_op != not_ else not_(expressions[0])
return result
def process_request(self, req, resp):
"""
Converts request input data from JSON to a dict.
:param req: Falcon request
:type req: falcon.request.Request
:param resp: Falcon response
:type resp: falcon.response.Response
"""
# req.stream corresponds to the WSGI wsgi.input environ variable,
# and allows you to read bytes from the request body.
#
# See also: PEP 3333
if req.content_length in (None, 0):
# Nothing to do
return
body = req.stream.read()
if not body:
raise falcon.HTTPBadRequest('Empty request body',
'A valid JSON document is required.')
try:
req.context['doc'] = json.loads(body.decode('utf-8'))
except (ValueError, UnicodeDecodeError):
raise falcon.HTTPError(falcon.HTTP_753,
'Malformed JSON',
'Could not decode the request body. The '
'JSON was incorrect or not encoded as '
'UTF-8.')
def process_resource(self, req, resp, resource, params):
"""
:param req: Falcon request
:type req: falcon.request.Request
:param resp: Falcon response
:type resp: falcon.response.Response
:param resource:
:type resource: falcon_dbapi.resources.base.BaseCollectionResource|
falcon_dbapi.resources.base.BaseSingleResource
:param params: parameters dict
:type params: dict
"""
if resource is None or req.method not in ['POST', 'PUT', 'PATCH']:
return
body = req.stream.read()
if not body:
raise falcon.HTTPBadRequest('Empty request body', 'At least one value is required')
try:
body = body.decode('utf-8')
except UnicodeDecodeError:
raise falcon.HTTPBadRequest('Invalid request body', 'A valid UTF-8 encoded document is required')
req.context['doc'] = parse_query_string(body, keep_blank_qs_values=True, parse_qs_csv=False)
def on_post(self, req, resp, revision_id, validation_name):
validation_data = req.stream.read(req.content_length or 0)
try:
validation_data = yaml.safe_load(validation_data)
except yaml.YAMLError as e:
error_msg = ("Could not parse the validation into YAML data. "
"Details: %s." % e)
LOG.error(error_msg)
raise falcon.HTTPBadRequest(description=six.text_type(e))
if not validation_data:
error_msg = 'Validation payload must be provided.'
LOG.error(error_msg)
raise falcon.HTTPBadRequest(description=error_msg)
if not all([validation_data.get(x) for x in ('status', 'validator')]):
error_msg = 'Validation payload must contain keys: %s.' % (
', '.join(['"status"', '"validator"']))
LOG.error(error_msg)
raise falcon.HTTPBadRequest(description=error_msg)
try:
resp_body = db_api.validation_create(
revision_id, validation_name, validation_data)
except errors.RevisionNotFound as e:
raise falcon.HTTPNotFound(description=e.format_message())
resp.status = falcon.HTTP_201
resp.append_header('Content-Type', 'application/x-yaml')
resp.body = self.view_builder.show(resp_body)
def on_put(self, req, resp, bucket_name=None):
document_data = req.stream.read(req.content_length or 0)
try:
documents = list(yaml.safe_load_all(document_data))
except yaml.YAMLError as e:
error_msg = ("Could not parse the document into YAML data. "
"Details: %s." % e)
LOG.error(error_msg)
raise falcon.HTTPBadRequest(description=six.text_type(e))
# NOTE: Must validate documents before doing policy enforcement,
# because we expect certain formatting of the documents while doing
# policy enforcement. If any documents fail basic schema validaiton
# raise an exception immediately.
try:
doc_validator = document_validation.DocumentValidation(documents)
validations = doc_validator.validate_all()
except (deckhand_errors.InvalidDocumentFormat,
deckhand_errors.InvalidDocumentSchema) as e:
LOG.exception(e.format_message())
raise falcon.HTTPBadRequest(description=e.format_message())
for document in documents:
if document['metadata'].get('storagePolicy') == 'encrypted':
policy.conditional_authorize(
'deckhand:create_encrypted_documents', req.context)
break
self._prepare_secret_documents(documents)
created_documents = self._create_revision_documents(
bucket_name, documents, validations)
resp.body = self.view_builder.list(created_documents)
resp.status = falcon.HTTP_200
def test_required(self, simulate_request, client, resource, method_name):
client.app.add_route('/', resource)
query_string = ''
simulate_request(client=client, path='/', query_string=query_string)
req = resource.captured_req
try:
getattr(req, method_name)('marker', required=True)
pytest.fail('falcon.HTTPMissingParam not raised')
except falcon.HTTPMissingParam as ex:
assert isinstance(ex, falcon.HTTPBadRequest)
assert ex.title == 'Missing parameter'
expected_desc = 'The "marker" parameter is required.'
assert ex.description == expected_desc
def test_boolean(self, simulate_request, client, resource):
client.app.add_route('/', resource)
query_string = ('echo=true&doit=false&bogus=bar&bogus2=foo&'
't1=True&f1=False&t2=yes&f2=no&blank&one=1&zero=0&'
'checkbox1=on&checkbox2=off')
simulate_request(client=client, path='/', query_string=query_string)
req = resource.captured_req
with pytest.raises(falcon.HTTPBadRequest):
req.get_param_as_bool('bogus')
try:
req.get_param_as_bool('bogus2')
except Exception as ex:
assert isinstance(ex, falcon.HTTPInvalidParam)
assert ex.title == 'Invalid parameter'
expected_desc = ('The "bogus2" parameter is invalid. '
'The value of the parameter must be "true" '
'or "false".')
assert ex.description == expected_desc
assert req.get_param_as_bool('echo') is True
assert req.get_param_as_bool('doit') is False
assert req.get_param_as_bool('t1') is True
assert req.get_param_as_bool('t2') is True
assert req.get_param_as_bool('f1') is False
assert req.get_param_as_bool('f2') is False
assert req.get_param_as_bool('one') is True
assert req.get_param_as_bool('zero') is False
assert req.get_param('blank') is None
assert req.get_param_as_bool('checkbox1') is True
assert req.get_param_as_bool('checkbox2') is False
store = {}
assert req.get_param_as_bool('echo', store=store) is True
assert store['echo'] is True
def test_http_error_repr():
error = falcon.HTTPBadRequest()
_repr = '<%s: %s>' % (error.__class__.__name__, error.status)
assert error.__repr__() == _repr
def test_misc(self, client):
self._misc_test(client, falcon.HTTPBadRequest, falcon.HTTP_400)
self._misc_test(client, falcon.HTTPNotAcceptable, falcon.HTTP_406,
needs_title=False)
self._misc_test(client, falcon.HTTPConflict, falcon.HTTP_409)
self._misc_test(client, falcon.HTTPPreconditionFailed, falcon.HTTP_412)
self._misc_test(client, falcon.HTTPUnsupportedMediaType, falcon.HTTP_415,
needs_title=False)
self._misc_test(client, falcon.HTTPUnprocessableEntity, falcon.HTTP_422)
self._misc_test(client, falcon.HTTPUnavailableForLegalReasons, falcon.HTTP_451,
needs_title=False)
self._misc_test(client, falcon.HTTPInternalServerError, falcon.HTTP_500)
self._misc_test(client, falcon.HTTPBadGateway, falcon.HTTP_502)
def test_required_header(self, client):
resource = testing.SimpleTestResource(body=SAMPLE_BODY)
client.app.add_route('/', resource)
client.simulate_get()
try:
req = resource.captured_req
req.get_header('X-Not-Found', required=True)
pytest.fail('falcon.HTTPMissingHeader not raised')
except falcon.HTTPMissingHeader as ex:
assert isinstance(ex, falcon.HTTPBadRequest)
assert ex.title == 'Missing header value'
expected_desc = 'The X-Not-Found header is required.'
assert ex.description == expected_desc
def test_jsonschema_validation_failure():
req = RequestStub()
req.media = {}
res = SampleResource()
with pytest.raises(falcon.HTTPBadRequest) as err:
res.on_get(req, None)
assert err.value.description == '\'message\' is a required property'
def validate(req, resp, params):
raise falcon.HTTPBadRequest('Invalid thing', 'Your thing was not '
'formatted correctly.')
def validate_field(req, resp, params):
try:
params['id'] = int(params['id'])
except ValueError:
raise falcon.HTTPBadRequest('Invalid ID', 'ID was not valid.')
def __init__(self, title=None, description=None, **kwargs):
super(HTTPBadRequest, self).__init__(status.HTTP_400, title,
description, **kwargs)
def validate(schema):
"""Decorator that validates ``req.media`` using JSON Schema
Args:
schema (dict): A dictionary that follows the JSON Schema specification.
See `json-schema.org <http://json-schema.org/>`_ for more
information on defining a compatible dictionary.
Example:
.. code:: python
from falcon.media.validators import jsonschema
# -- snip --
@jsonschema.validate(my_post_schema)
def on_post(self, req, resp):
# -- snip --
Note:
This validator requires the ``jsonschema`` library available via
PyPI. The library also requires Python 2.7+.
"""
def decorator(func):
def wrapper(self, req, resp, *args, **kwargs):
try:
jsonschema.validate(req.media, schema)
except jsonschema.ValidationError as e:
raise falcon.HTTPBadRequest(
'Failed data validation',
description=e.message
)
return func(self, req, resp, *args, **kwargs)
return wrapper
return decorator
