Python flask 模块,json() 实例源码
我们从Python开源项目中,提取了以下25个代码示例,用于说明如何使用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, 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, 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, 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, limit=limit))
return render_template(
"result.html", count=count, results=results,
result_formater=result_dump.result_formater,
project=project, limit=limit, json=json
)
def on(self, 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,``'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, namespace, sid,
*args)
if self.server:
self.server.on(message, _handler, namespace=namespace)
else:
self.handlers.append((message, 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', namespace=namespace, room=room,
skip_sid=skip_sid, callback=callback, **kwargs)
else:
self.emit('message', **kwargs)
def convert_application_json(self, 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 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,
socketio_path=resource)
app.wsgi_app = self.sockio_mw
def emit(event, **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,
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)