Python flask 模块，json() 实例源码

def serve_app(similarities, fragments, pharmacophores, internal_port=8084, external_url='http://localhost:8084/kripo'):
    """Serve webservice forever

    Args:
        similarities: Filename of similarity matrix hdf5 file
        fragments: Filename of fragments database file
        pharmacophores: Filename of pharmacophores hdf5 file
        internal_port: TCP port on which to listen
        external_url (str): URL which should be used in Swagger spec
    """
    sim_matrix = open_similarity_matrix(similarities)
    pharmacophores_db = PharmacophoresDb(pharmacophores)
    app = wsgi_app(sim_matrix, fragments, pharmacophores_db, external_url)
    LOGGER.setLevel(logging.INFO)
    LOGGER.addHandler(logging.StreamHandler())
    LOGGER.info(' * Swagger spec at {}/swagger.json'.format(external_url))
    LOGGER.info(' * Swagger ui at {}/ui'.format(external_url))
    try:
        app.run(port=internal_port)
    finally:
        sim_matrix.close()

def myPaymentEngine():

    stripe.api_key = "YOUR_TEST_SECRET_KEY"

    json = request.json

    token = json['stripeToken']
    amount = json['amount']
    description = json['description']

    try:
        charge = stripe.Charge.create(amount=amount, currency="usd", card=token, description=description)
    except stripe.CardError, e:
        pass

    return "Your payment is done successfully!"

def dump_result(project, _format):
    resultdb = app.config['resultdb']
    # force update project list
    resultdb.get(project, 'any')
    if project not in resultdb.projects:
        return "no such project.", 404

    offset = int(request.args.get('offset', 0)) or None
    limit = int(request.args.get('limit', 0)) or None
    results = resultdb.select(project, offset=offset, limit=limit)

    if _format == 'json':
        valid = request.args.get('style', 'rows') == 'full'
        return Response(result_dump.dump_as_json(results, valid),
                        mimetype='application/json')
    elif _format == 'txt':
        return Response(result_dump.dump_as_txt(results),
                        mimetype='text/plain')
    elif _format == 'csv':
        return Response(result_dump.dump_as_csv(results),
                        mimetype='text/csv')

def get(self, _device_):
        #Lookup Company
        company = "Flyball-Labs"
        assetSearch = company + "_" + _device_ + "_*"
        assetFullQueryURL = assetQueryURL + "/" + assetSearch
        print(assetFullQueryURL)
        try:
            response = requests.get(assetFullQueryURL, headers={"Accept" : "application/json"})
            jData = response.json()
        except:
            return "Server Down"
        counter = 0
        decodedList  = []
        for row in jData['Row']:
            # Decode the key into ascii
            #rowKey = base64.b64decode(row['key']).decode('ascii')
            dColumn = {}
            for cell in row['Cell']:
               columnname = base64.b64decode(cell['column']).decode('ascii')
               value = base64.b64decode(cell['$']).decode('ascii')
               dColumn[columnname] = value 
            decodedList.append (dColumn) 
        return jsonify(decodedList)

def on_event(self, message, handler, namespace=None):
        """Register a SocketIO event handler.

        ``on_event`` is the non-decorator version of ``'on'``.

        Example::

            def on_foo_event(json):
                print('received json: ' + str(json))

            socketio.on_event('my event', on_foo_event, namespace='/chat')

        :param message: The name of the event. This is normally a user defined
                        string, but a few event names are already defined. Use
                        ``'message'`` to define a handler that takes a string
                        payload, ``'json'`` to define a handler that takes a
                        JSON blob payload, ``'connect'`` or ``'disconnect'``
                        to create handlers for connection and disconnection
                        events.
        :param handler: The function that handles the event.
        :param namespace: The namespace on which the handler is to be
                          registered. Defaults to the global namespace.
        """
        self.on(message, namespace=namespace)(handler)

def __init__(self, data=None, status=200, content_type=None, converters=None):
        default_content_type = 'application/json;charset=utf-8'
        converters = dict(converters or [])
        converters[default_content_type] = self.convert_application_json

        if content_type is None:
            # If the content type is not explicitly given, detect it
            content_type = request.accept_mimetypes.best_match([default_content_type] + list(converters.keys()))

        try:
            converter = converters[content_type]
        except KeyError:
            converter = converters[default_content_type]

        final_data = converter(data, status, content_type)
        super(BaseApiResponse, self).__init__(final_data, content_type=content_type, status=status)

def create_task():
    print("before adding phone number")
    if not request.json or not 'phoneNumber' in request.json:
        abort(400)
    try:
        task = {
            'id': phoneNumbers[-1]['id'] + 1,
            'name': request.json['name'],
            'phoneNumber': str(int(request.json.get('phoneNumber', ""))),
        }
        phoneNumbers.append(task)
        print("added data")
        print(task)
        return jsonify({'task': task}), 201
    except ValueError:
        abort(415)

# send an sms to all phone numbers registered
# curl -X POST  http://localhost:5000/sendsms
# curl -X POST -H "Content-Type: application/json" -d '{"message": "who wants toast"}' http://localhost:5000/sendsms

def dispatch_request(self, *args, **kwargs):
        response = super().dispatch_request(*args, **kwargs)

        if isinstance(response, flask.Response):
            return response

        try:
            response = self.prepare_response(response)
        except Exception as exc:
            LOG.error("Cannot build model response: %s", exc)
            raise exceptions.UnknownReturnValueError from exc

        try:
            response = flask.json.jsonify(response)
        except Exception as exc:
            LOG.error("Cannot convert %s to JSON: %s", response, exc)
            raise exceptions.CannotConvertResultToJSONError() from exc

        return response

def webhook(self,
                action=None,
                parameters={}):
        """
        Uses a regular Flask test client in order to post valid webhook messages.

        :param action: action to tested
        :param parameters: parameters of the action as dictionary
        :return: returns a response object as a regular Flask test client
        """
        req = {
            "result": {
                "action": action,
                "parameters": parameters
            }
        }

        return self.post(
            self.apiai_webhook.webhook_url,
            data=json.dumps(req),
            content_type="application/json",
            headers={self.apiai_webhook.api_key_header: self.apiai_webhook.api_key_value})

def callapi(url, method='post', payload=None, headers=None, auth=None, check=True):
    """
    Simple wrapper around `requests.post`, with excessive logging.
    Returns a Flask-friendly tuple on success or failure.
    Logs and re-raises any exceptions.
    """
    if not headers:
        headers = {'Content-type': 'application/json'}
    try:
        logging.info("URL=%s" % url)
        logging.info("Auth=%s" % str(auth))
        logging.info("Headers=%s" % headers)
        logging.info("Body=%s" % payload)
        logging.info("Check=%s" % check)
        if (auth is not None):
            r = requests.request(method, url, auth=auth, headers=headers, data=payload, verify=bool(strtobool(str(check))))
        else:
            r = requests.request(method, url, headers=headers, data=payload, verify=check)
        if r.status_code >= 200 and r.status_code < 300:
            if (payload is None):
                return r.text
            else:
                return ("OK", r.status_code, None)
    except:
        logging.exception("Can't create new payload. Check code and try again.")
        raise
    return ("%s" % r.text, r.status_code, None)

def result():
    resultdb = app.config['resultdb']
    project = request.args.get('project')
    offset = int(request.args.get('offset', 0))
    limit = int(request.args.get('limit', 20))

    count = resultdb.count(project)
    results = list(resultdb.select(project, offset=offset, limit=limit))

    return render_template(
        "result.html", count=count, results=results,
        result_formater=result_dump.result_formater,
        project=project, offset=offset, limit=limit, json=json
    )

def on(self, message, namespace=None):
        """Decorator to register a SocketIO event handler.

        This decorator must be applied to SocketIO event handlers. Example::

            @socketio.on('my event', namespace='/chat')
            def handle_my_custom_event(json):
                print('received json: ' + str(json))

        :param message: The name of the event. This is normally a user defined
                        string, but a few event names are already defined. Use
                        ``'message'`` to define a handler that takes a string
                        payload, ``'json'`` to define a handler that takes a
                        JSON blob payload, ``'connect'`` or ``'disconnect'``
                        to create handlers for connection and disconnection
                        events.
        :param namespace: The namespace on which the handler is to be
                          registered. Defaults to the global namespace.
        """
        namespace = namespace or '/'

        def decorator(handler):
            def _handler(sid, *args):
                return self._handle_event(handler, message, namespace, sid,
                                          *args)

            if self.server:
                self.server.on(message, _handler, namespace=namespace)
            else:
                self.handlers.append((message, _handler, namespace))
            return handler
        return decorator

def send(self, data, json=False, namespace=None, room=None,
             callback=None, include_self=True, skip_sid=None, **kwargs):
        """Send a server-generated SocketIO message.

        This function sends a simple SocketIO message to one or more connected
        clients. The message can be a string or a JSON blob. This is a simpler
        version of ``emit()``, which should be preferred. This function can be
        used outside of a SocketIO event context, so it is appropriate to use
        when the server is the originator of an event.

        :param message: The message to send, either a string or a JSON blob.
        :param json: ``True`` if ``message`` is a JSON blob, ``False``
                     otherwise.
        :param namespace: The namespace under which the message is to be sent.
                          Defaults to the global namespace.
        :param room: Send the message only to the users in the given room. If
                     this parameter is not included, the message is sent to
                     all connected users.
        :param skip_sid: The session id of a client to ignore when broadcasting
                         or addressing a room. This is typically set to the
                         originator of the message, so that everyone except
                         that client receive the message.
        :param callback: If given, this function will be called to acknowledge
                         that the client has received the message. The
                         arguments that will be passed to the function are
                         those provided by the client. Callback functions can
                         only be used when addressing an individual client.
        """
        skip_sid = flask.request.sid if not include_self else skip_sid
        if json:
            self.emit('json', data, namespace=namespace, room=room,
                      skip_sid=skip_sid, callback=callback, **kwargs)
        else:
            self.emit('message', data, namespace=namespace, room=room,
                      skip_sid=skip_sid, callback=callback, **kwargs)

def as_json_hint(json, **kwargs):
        kwargs.update({'message': 'FROG EXPECTED JSON OF THE FORM: {0}'.format(json)})
        return ApiError(**kwargs)

def convert_application_json(self, data, status, content_type):
        if data is None:
            return

        return flask.json.dumps(obj=data, indent=None, ensure_ascii=False, default=default_encoder, encoding='utf-8')

def after_request(response):
    response.headers.add("Access-Control-Allow-Origin", "*")
    response.headers.add("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS")
    response.headers.add('Access-Control-Allow-Headers', 'X-Requested-With, Content-Type, Authorization')
    response.headers.add('Access-Control-Allow-Credentials', 'true')
    response.headers.add('Content-Type', 'application/json')
    return response

def get_body(self, environ=None):
        error = self.error_name or self.__class__.__name__
        error = str(error)

        error_message = {
            "code": self.code,
            "error": error,
            "message": self.get_description(environ)
        }
        json_error = flask.json.dumps(error_message)

        return json_error

def get_headers(self, environ=None):
        return [("Content-Type", "application/json")]

def send_message_back( payload ):
    resp = Response(
        json.dumps( payload ),
        status=200,
        mimetype='application/json')
    return resp

def __init__(self, import_name, **kwargs):
        super(FleakerJSONApp, self).__init__(import_name, **kwargs)

        # @TODO (json): Should be class level to make this easier to override.
        self.json_encoder = FleakerJSONEncoder
        self.json = flask.json
        self.config.setdefault('JSON_SORT_KEYS', True)

def json(func):
    """
    Decorator to render as JSON
    :param func:
    :return:
    """
    if inspect.isclass(func):
        apply_function_to_members(func, json)
        return func
    else:
        @functools.wraps(func)
        def decorated_view(*args, **kwargs):
            data = func(*args, **kwargs)
            return _build_response(data, jsonify)
        return decorated_view

def get_fragments(fragment_ids=None, pdb_codes=None):
    """Retrieve fragments based on their identifier or PDB code.

    Args:
        fragment_ids (List[str]): List of fragment identifiers
        pdb_codes (List[str]): List of PDB codes

    Returns:
        list[dict]: List of fragment information

    Raises:
        werkzeug.exceptions.NotFound: When one of the fragments_ids or pdb_code could not be found
    """
    fragments_db_filename = current_app.config['fragments']
    with FragmentsDb(fragments_db_filename) as fragmentsdb:
        fragments = []
        missing_ids = []
        if fragment_ids:
            for frag_id in fragment_ids:
                try:
                    fragments.append(fragmentsdb[frag_id])
                except LookupError:
                    missing_ids.append(frag_id)

        if pdb_codes:
            for pdb_code in pdb_codes:
                try:
                    for fragment in fragmentsdb.by_pdb_code(pdb_code):
                        fragments.append(fragment)
                except LookupError:
                    missing_ids.append(pdb_code)
        # TODO if fragment_ids and pdb_codes are both None then return paged list of all fragments
        if missing_ids:
            title = 'Not found'
            label = 'identifiers'
            if pdb_codes:
                label = 'PDB codes'
            description = 'Fragments with {1} \'{0}\' not found'.format(','.join(missing_ids), label)
            # connexion.problem is using json.dumps instead of flask custom json encoder, so performing convert myself
            # TODO remove mol2string conversion when https://github.com/zalando/connexion/issues/266 is fixed
            for fragment in fragments:
                if fragment['mol']:
                    fragment['mol'] = MolToMolBlock(fragment['mol'])
            ext = {'absent_identifiers': missing_ids, 'fragments': fragments}
            return connexion.problem(404, title, description, ext=ext)
        return fragments

def init_app(self, app, **kwargs):
        if app is not None:
            if not hasattr(app, 'extensions'):
                app.extensions = {}  # pragma: no cover
            app.extensions['socketio'] = self
        self.server_options.update(kwargs)

        if 'client_manager' not in self.server_options:
            url = kwargs.pop('message_queue', None)
            channel = kwargs.pop('channel', 'flask-socketio')
            write_only = app is None
            if url:
                if url.startswith('redis://'):
                    queue_class = socketio.RedisManager
                elif url.startswith('zmq'):
                    queue_class = socketio.ZmqManager
                else:
                    queue_class = socketio.KombuManager
                queue = queue_class(url, channel=channel,
                                    write_only=write_only)
                self.server_options['client_manager'] = queue

        if 'json' in self.server_options and \
                self.server_options['json'] == flask_json:
            # flask's json module is tricky to use because its output
            # changes when it is invoked inside or outside the app context
            # so here to prevent any ambiguities we replace it with wrappers
            # that ensure that the app context is always present
            class FlaskSafeJSON(object):
                @staticmethod
                def dumps(*args, **kwargs):
                    with app.app_context():
                        return flask_json.dumps(*args, **kwargs)

                @staticmethod
                def loads(*args, **kwargs):
                    with app.app_context():
                        return flask_json.loads(*args, **kwargs)

            self.server_options['json'] = FlaskSafeJSON

        resource = kwargs.pop('path', kwargs.pop('resource', 'socket.io'))
        if resource.startswith('/'):
            resource = resource[1:]
        self.server = socketio.Server(**self.server_options)
        self.async_mode = self.server.async_mode
        for handler in self.handlers:
            self.server.on(handler[0], handler[1], namespace=handler[2])
        for namespace_handler in self.namespace_handlers:
            self.server.register_namespace(namespace_handler)

        if app is not None:
            # here we attach the SocketIO middlware to the SocketIO object so it
            # can be referenced later if debug middleware needs to be inserted
            self.sockio_mw = _SocketIOMiddleware(self.server, app,
                                                 socketio_path=resource)
            app.wsgi_app = self.sockio_mw

def emit(event, *args, **kwargs):
    """Emit a SocketIO event.

    This function emits a SocketIO event to one or more connected clients. A
    JSON blob can be attached to the event as payload. This is a function that
    can only be called from a SocketIO event handler, as in obtains some
    information from the current client context. Example::

        @socketio.on('my event')
        def handle_my_custom_event(json):
            emit('my response', {'data': 42})

    :param event: The name of the user event to emit.
    :param args: A dictionary with the JSON data to send as payload.
    :param namespace: The namespace under which the message is to be sent.
                      Defaults to the namespace used by the originating event.
                      A ``'/'`` can be used to explicitly specify the global
                      namespace.
    :param callback: Callback function to invoke with the client's
                     acknowledgement.
    :param broadcast: ``True`` to send the message to all clients, or ``False``
                      to only reply to the sender of the originating event.
    :param room: Send the message to all the users in the given room. If this
                 argument is set, then broadcast is implied to be ``True``.
    :param include_self: ``True`` to include the sender when broadcasting or
                         addressing a room, or ``False`` to send to everyone
                         but the sender.
    """
    if 'namespace' in kwargs:
        namespace = kwargs['namespace']
    else:
        namespace = flask.request.namespace
    callback = kwargs.get('callback')
    broadcast = kwargs.get('broadcast')
    room = kwargs.get('room')
    if room is None and not broadcast:
        room = flask.request.sid
    include_self = kwargs.get('include_self', True)

    socketio = flask.current_app.extensions['socketio']
    return socketio.emit(event, *args, namespace=namespace, room=room,
                         include_self=include_self, callback=callback)

def default(self, obj):
        """Encode individual objects into their JSON representation.

        This method is used by :class:`flask.json.JSONEncoder` to encode
        individual items in the JSON object.

        Args:
            obj (object): Any Python object we wish to convert to JSON.

        Returns:
            str: The stringified, valid JSON representation of our provided
                object.
        """
        if isinstance(obj, decimal.Decimal):
            obj = format(obj, 'f')
            str_digit = text_type(obj)

            return (str_digit.rstrip('0').rstrip('.')
                    if '.' in str_digit
                    else str_digit)

        elif isinstance(obj, phonenumbers.PhoneNumber):
            return phonenumbers.format_number(
                obj,
                phonenumbers.PhoneNumberFormat.E164
            )

        elif isinstance(obj, pendulum.Pendulum):
            return text_type(obj)

        elif isinstance(obj, arrow.Arrow):
            return text_type(obj)

        elif isinstance(obj, (datetime.datetime, datetime.date)):
            return obj.isoformat()

        try:
            return list(iter(obj))
        except TypeError:
            pass

        return super(FleakerJSONEncoder, self).default(obj)