如何将 JSON 作为 python tornado "post" 方法的输入
How to take in JSON as input to python tornado "post" method
我正在尝试使用龙卷风做一个简单的获取和 post 方法。龙卷风框架很新。对于 post 我想接受一个 json 作为输入，使用该输入输入另一个函数，我必须执行另一部分代码。但是，即使使用简单的 self.write().
，我也无法使用 tornado post 方法
对于我的 get 方法，我从 SQL 数据库中读取以获取传感器的状态并将其写入 json 格式。 get 方法非常有效！当我转到 localhost:port# 时，它会读出我的 json 文件。对于我的 post 方法，我想采用一个简单的 json，其中只有一个 key:value，它是一个浮点数。我想获取用户在 json 中指定的浮点数，并在我的 flowMKS.set() 函数中使用它，这将更改传感器的设定点参数。我不确定如何将 json 输入到 post 方法并将其读入变量。我在下面有一些我试过但没有用的#commented 代码。然而，我回到了基础，只是做了一个 self.write("Hello World") 来查看 post 是否正常工作。我也无法让 self.write 工作。当我转到 localhost:port#/flow_post 时，不断收到 500 错误消息。我的 get 方法中使用了变量 flow_status。
预期的结果是将 json {"setpoint":45.5} 纳入 post 方法。使用数字并插入到我的 flowMKS 方法中以更改传感器上的参数。
如何将 json 引入 post 方法并从 json 输入中获取数字并将其存储在变量中？
class Current(tornado.web.RequestHandler):
def get(self):
global flow_status
time = flow_status[0]
ip = flow_status[1]
rate = flow_status[2]
setp = flow_status[3]
tempc = flow_status[4]
status = {"flow_controller":{
"time":time,
"ip":ip,
"rate_sccm":rate,
"setpoint":setp,
"temperature_c":tempc,
}
}
self.write(status)
class Update(tornado.web.RequestHandler):
# def prepare(self):
# if self.request.haders["Content-Type"].startswith("application/json"):
# self.json_args = json.loads(self.request.body)
# else:
# self.json_args = None
def post(self):
# #expecting body data to contain JSON so we use json.loads to decrypt the JSON into a dict
# data = json.loads(self.request.body)
#
# #Getting what the setpoint should be
# setpoint = self.json_args["setpoint"]
#
# #making the input a float
# setpoint = float(setpoint)
#
# #setting up connection with sensor
# flowMKS = FlowController(flow_status[1])
#
# #sending setpoint to sensor
# flowMKS.set(setpoint)
self.write("Hello World")
if __name__ == '__main__':
# global flow_status
#Below is creating the Tornado based API for get and post methods
tornado.options.parse_command_line()
app = tornado.web.Application(
handlers=[(r'/',Current), (r'/flow_post', Update)])
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
#using PeriodicCallback to get info from the SQL database every 500 ms
PeriodicCallback(get_sql_status,500).start()
#starting the entire Tornado IOLoop
tornado.ioloop.IOLoop.current().start()
要使用 Tornado 上传文件，您可以使用此函数 tornado.httputil.parse_body_arguments，它将上传的文件内容拆分为字典 file_dict 和 args_dict 中 FormData 中的其他参数。
示例代码：
import tornado.httputil
import tornado.web
import tornado.escape
import json
import os
import sys
import traceback
class FileHandler(tornado.web.RequestHandler):
def _return_response(self, request, message_to_be_returned: dict, status_code):
"""
Returns formatted response back to client
"""
try:
request.set_header("Content-Type", "application/json; charset=UTF-8")
request.set_status(status_code)
#If dictionary is not empty then write the dictionary directly into
if(bool(message_to_be_returned)):
request.write(message_to_be_returned)
request.finish()
except Exception:
raise
def set_default_headers(self, *args, **kwargs):
self.set_header('Content-Type','text/csv')
self.set_header("Access-Control-Allow-Origin", "*")
self.set_header("Access-Control-Allow-Headers", "x-requested-with")
self.set_header("Access-Control-Allow-Methods", "*")
def post(self):
"""
This function reads an uploaded file
"""
try:
arg_dict = {}
file_dict = {}
tornado.httputil.parse_body_arguments(self.request.headers["Content-Type"], self.request.body, arg_dict, file_dict)
uploaded_file = file_dict['TestFile'][0]
if not uploaded_file:
return self._return_response(self, { 'message': 'No test file uploaded, please upload a test file' }, 400)
# File contents here
file_contents = str(uploaded_file['body'], "utf-8")
self.set_status(200)
self.finish()
except Exception as ex:
return self._return_response(self, { "message": 'Could not complete the request because of some error at the server!', "cause": ex.args[0], "stack_trace": traceback.format_exc(sys.exc_info()) }, 500)
您也可以使用 tornado.escape.json_decode 将请求正文反序列化为字典并对其进行处理。
示例代码：
import tornado.gen
import tornado.web
import tornado.escape
import json
import os
import sys
import traceback
class JSONHandler(tornado.web.RequestHandler):
def _return_response(self, request, message_to_be_returned: dict, status_code):
"""
Returns formatted response back to client
"""
try:
request.set_header("Content-Type", "application/json; charset=UTF-8")
request.set_status(status_code)
#If dictionary is not empty then write the dictionary directly into
if(bool(message_to_be_returned)):
request.write(message_to_be_returned)
request.finish()
except Exception:
raise
def set_default_headers(self, *args, **kwargs):
self.set_header("Content-Type", "application/json")
self.set_header("Access-Control-Allow-Origin", "*")
self.set_header("Access-Control-Allow-Headers", "x-requested-with")
self.set_header("Access-Control-Allow-Methods", "*")
def post(self):
"""
This function parses the request body and does something
"""
try:
# Do something with request body
request_payload = tornado.escape.json_decode(self.request.body)
return self._return_response(self, request_payload, 200)
except json.decoder.JSONDecodeError:
return self._return_response(self, { "message": 'Cannot decode request body!' }, 400)
except Exception as ex:
return self._return_response(self, { "message": 'Could not complete the request because of some error at the server!', "cause": ex.args[0], "stack_trace": traceback.format_exc(sys.exc_info()) }, 500)
我正在尝试使用龙卷风做一个简单的获取和 post 方法。龙卷风框架很新。对于 post 我想接受一个 json 作为输入，使用该输入输入另一个函数，我必须执行另一部分代码。但是，即使使用简单的 self.write().
，我也无法使用 tornado post 方法对于我的 get 方法，我从 SQL 数据库中读取以获取传感器的状态并将其写入 json 格式。 get 方法非常有效！当我转到 localhost:port# 时，它会读出我的 json 文件。对于我的 post 方法，我想采用一个简单的 json，其中只有一个 key:value，它是一个浮点数。我想获取用户在 json 中指定的浮点数，并在我的 flowMKS.set() 函数中使用它，这将更改传感器的设定点参数。我不确定如何将 json 输入到 post 方法并将其读入变量。我在下面有一些我试过但没有用的#commented 代码。然而，我回到了基础，只是做了一个 self.write("Hello World") 来查看 post 是否正常工作。我也无法让 self.write 工作。当我转到 localhost:port#/flow_post 时，不断收到 500 错误消息。我的 get 方法中使用了变量 flow_status。
预期的结果是将 json {"setpoint":45.5} 纳入 post 方法。使用数字并插入到我的 flowMKS 方法中以更改传感器上的参数。
如何将 json 引入 post 方法并从 json 输入中获取数字并将其存储在变量中？
class Current(tornado.web.RequestHandler):
def get(self):
global flow_status
time = flow_status[0]
ip = flow_status[1]
rate = flow_status[2]
setp = flow_status[3]
tempc = flow_status[4]
status = {"flow_controller":{
"time":time,
"ip":ip,
"rate_sccm":rate,
"setpoint":setp,
"temperature_c":tempc,
}
}
self.write(status)
class Update(tornado.web.RequestHandler):
# def prepare(self):
# if self.request.haders["Content-Type"].startswith("application/json"):
# self.json_args = json.loads(self.request.body)
# else:
# self.json_args = None
def post(self):
# #expecting body data to contain JSON so we use json.loads to decrypt the JSON into a dict
# data = json.loads(self.request.body)
#
# #Getting what the setpoint should be
# setpoint = self.json_args["setpoint"]
#
# #making the input a float
# setpoint = float(setpoint)
#
# #setting up connection with sensor
# flowMKS = FlowController(flow_status[1])
#
# #sending setpoint to sensor
# flowMKS.set(setpoint)
self.write("Hello World")
if __name__ == '__main__':
# global flow_status
#Below is creating the Tornado based API for get and post methods
tornado.options.parse_command_line()
app = tornado.web.Application(
handlers=[(r'/',Current), (r'/flow_post', Update)])
http_server = tornado.httpserver.HTTPServer(app)
http_server.listen(options.port)
#using PeriodicCallback to get info from the SQL database every 500 ms
PeriodicCallback(get_sql_status,500).start()
#starting the entire Tornado IOLoop
tornado.ioloop.IOLoop.current().start()
要使用 Tornado 上传文件，您可以使用此函数 tornado.httputil.parse_body_arguments，它将上传的文件内容拆分为字典 file_dict 和 args_dict 中 FormData 中的其他参数。
示例代码：
import tornado.httputil
import tornado.web
import tornado.escape
import json
import os
import sys
import traceback
class FileHandler(tornado.web.RequestHandler):
def _return_response(self, request, message_to_be_returned: dict, status_code):
"""
Returns formatted response back to client
"""
try:
request.set_header("Content-Type", "application/json; charset=UTF-8")
request.set_status(status_code)
#If dictionary is not empty then write the dictionary directly into
if(bool(message_to_be_returned)):
request.write(message_to_be_returned)
request.finish()
except Exception:
raise
def set_default_headers(self, *args, **kwargs):
self.set_header('Content-Type','text/csv')
self.set_header("Access-Control-Allow-Origin", "*")
self.set_header("Access-Control-Allow-Headers", "x-requested-with")
self.set_header("Access-Control-Allow-Methods", "*")
def post(self):
"""
This function reads an uploaded file
"""
try:
arg_dict = {}
file_dict = {}
tornado.httputil.parse_body_arguments(self.request.headers["Content-Type"], self.request.body, arg_dict, file_dict)
uploaded_file = file_dict['TestFile'][0]
if not uploaded_file:
return self._return_response(self, { 'message': 'No test file uploaded, please upload a test file' }, 400)
# File contents here
file_contents = str(uploaded_file['body'], "utf-8")
self.set_status(200)
self.finish()
except Exception as ex:
return self._return_response(self, { "message": 'Could not complete the request because of some error at the server!', "cause": ex.args[0], "stack_trace": traceback.format_exc(sys.exc_info()) }, 500)
您也可以使用 tornado.escape.json_decode 将请求正文反序列化为字典并对其进行处理。
示例代码：
import tornado.gen
import tornado.web
import tornado.escape
import json
import os
import sys
import traceback
class JSONHandler(tornado.web.RequestHandler):
def _return_response(self, request, message_to_be_returned: dict, status_code):
"""
Returns formatted response back to client
"""
try:
request.set_header("Content-Type", "application/json; charset=UTF-8")
request.set_status(status_code)
#If dictionary is not empty then write the dictionary directly into
if(bool(message_to_be_returned)):
request.write(message_to_be_returned)
request.finish()
except Exception:
raise
def set_default_headers(self, *args, **kwargs):
self.set_header("Content-Type", "application/json")
self.set_header("Access-Control-Allow-Origin", "*")
self.set_header("Access-Control-Allow-Headers", "x-requested-with")
self.set_header("Access-Control-Allow-Methods", "*")
def post(self):
"""
This function parses the request body and does something
"""
try:
# Do something with request body
request_payload = tornado.escape.json_decode(self.request.body)
return self._return_response(self, request_payload, 200)
except json.decoder.JSONDecodeError:
return self._return_response(self, { "message": 'Cannot decode request body!' }, 400)
except Exception as ex:
return self._return_response(self, { "message": 'Could not complete the request because of some error at the server!', "cause": ex.args[0], "stack_trace": traceback.format_exc(sys.exc_info()) }, 500)