Python werkzeug 模块，Response() 实例源码

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'
        )

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')