def response_from_exception(self, exc):
status_code, error_code = 500, 'UNEXPECTED_ERROR'
if isinstance(exc, self.expected_exceptions):
if type(exc) in self.mapped_errors:
status_code, error_code = self.mapped_errors[type(exc)]
else:
status_code = 400
error_code = 'BAD_REQUEST'
return Response(
json.dumps({
'error': error_code,
'message': safe_for_serialization(exc),
}),
status=status_code,
mimetype='application/json'
)
python类Response()的实例源码
def jsonify(*args, **kwargs):
"""jsonify with support for MongoDB ObjectId"""
return Response(json.dumps(*args, cls=MongoJsonEncoder), mimetype='application/json', **kwargs)
def jsonify(*args, **kwargs):
"""
jsonify that uses hoplite's serial encoder
"""
return werkzeug.Response(
hoplite_dumps(dict(*args, **kwargs)), mimetype='application/json')
# This gets set by the app factory when the app is created
def jsonify(*args, **kwargs):
""" jsonify with support for MongoDB ObjectId
See https://gist.github.com/akhenakh/2954605
"""
return Response(json.dumps(dict(*args, **kwargs),
default=json_util.default,
indent=2,
cls=MongoJsonEncoder),
mimetype='application/json')
def humanify(obj, status_code=200):
""" Gets an obj and possibly a status code and returns a flask Resonse
with a jsonified obj, not suitable for humans
>>> humanify({"a": 1})
<Response 8 bytes [200 OK]>
>>> humanify({"a": 1}, 404)
<Response 8 bytes [404 NOT FOUND]>
>>> humanify({"a": 1}).get_data()
'{"a": 1}'
>>> humanify([1,2,3]).get_data()
'[1, 2, 3]'
"""
# TODO: refactor the name to `response`
# jsonify function doesn't work with lists
if type(obj) == list:
data = json.dumps(obj, default=json_util.default)
elif type(obj) == pymongo.cursor.Cursor:
rv = []
for doc in obj:
doc['_id'] = str(doc['_id'])
rv.append(dumps(doc))
data = '[' + ',\n'.join(rv) + ']' + '\n'
else:
data = dumps(obj,
default=json_util.default,
cls=MongoJsonEncoder)
resp = Response(data, mimetype='application/json')
resp.status_code = status_code
return resp
def get_product(self, request, product_id):
"""Gets product by `product_id`
"""
product = self.products_rpc.get(product_id)
return Response(
ProductSchema().dumps(product).data,
mimetype='application/json'
)
def create_product(self, request):
"""Create a new product - product data is posted as json
Example request ::
{
"id": "the_odyssey",
"title": "The Odyssey",
"passenger_capacity": 101,
"maximum_speed": 5,
"in_stock": 10
}
The response contains the new product ID in a json document ::
{"id": "the_odyssey"}
"""
schema = ProductSchema(strict=True)
try:
# load input data through a schema (for validation)
# Note - this may raise `ValueError` for invalid json,
# or `ValidationError` if data is invalid.
product_data = schema.loads(request.get_data(as_text=True)).data
except ValueError as exc:
raise BadRequest("Invalid json: {}".format(exc))
# Create the product
self.products_rpc.create(product_data)
return Response(
json.dumps({'id': product_data['id']}), mimetype='application/json'
)
def get_order(self, request, order_id):
"""Gets the order details for the order given by `order_id`.
Enhances the order details with full product details from the
products-service.
"""
order = self._get_order(order_id)
return Response(
GetOrderSchema().dumps(order).data,
mimetype='application/json'
)
def create_order(self, request):
"""Create a new order - order data is posted as json
Example request ::
{
"order_details": [
{
"product_id": "the_odyssey",
"price": "99.99",
"quantity": 1
},
{
"price": "5.99",
"product_id": "the_enigma",
"quantity": 2
},
]
}
The response contains the new order ID in a json document ::
{"id": 1234}
"""
schema = CreateOrderSchema(strict=True)
try:
# load input data through a schema (for validation)
# Note - this may raise `ValueError` for invalid json,
# or `ValidationError` if data is invalid.
order_data = schema.loads(request.get_data(as_text=True)).data
except ValueError as exc:
raise BadRequest("Invalid json: {}".format(exc))
# Create the order
# Note - this may raise `ProductNotFound`
id_ = self._create_order(order_data)
return Response(json.dumps({'id': id_}), mimetype='application/json')
def jsonify(*args, **kwargs):
""" jsonify with support for MongoDB ObjectId
"""
return Response(json.dumps(dict(*args, **kwargs), cls=MongoJsonEncoder), mimetype='application/json')