提问人：Vladislav 提问时间：10/31/2023 最后编辑：Vladislav 更新时间：11/1/2023 访问量：19
在 BaseHTTPRequestHandler 中填写和返回列表
fill and return list in BaseHTTPRequestHandler
问：
我无法在BaseHTTPRequestHandler类中返回列表。 这是我的带有注释的代码 想法是将带有 json 的传入 post 请求转发到另一个函数，通过填写新列表/字典并返回它来处理它。 我也不明白，这样的对象是否根本无法通过 requests.post 返回
from http.server import BaseHTTPRequestHandler, HTTPServer
import os
import time
import threading
import requests
import json
import calendar
def calculate(Event_dictionary_list, cached_timestamp):
Event_dictionary_list.clear() #- empty Event_dictionary_list
cached_timestamp = 0
def Async_func(Event_dictionary_list, cached_timestamp):
calculate(Event_dictionary_list, cached_timestamp)
def handler(event, Event_dictionary_list, cached_timestamp):
if not Event_dictionary_list: #Checking if Event_dictionary_list is empty
cached_timestamp = event['time']
threading.Timer(60,Async_func, [Event_dictionary_list, cached_timestamp]).start()
# make decision as to when to calculate
if event['time'] - cached_timestamp < 60*1000: #- in milliseconds
Event_dictionary_list.append(event)
class server(BaseHTTPRequestHandler):
def __init__(self, *args, **kwargs):
self.Event_dictionary_list = []
self.cached_timestamp = 0
super().__init__(*args, **kwargs)
def do_POST(self):
print("post msg received");
self.data_string = self.rfile.read(int(self.headers['Content-Length']))
self.event_time = calendar.timegm(time.gmtime())
self.send_response(200) # 200 = success - thanks for message response
self.send_header("Content-type", "application/json")
self.end_headers()
data = json.loads(self.data_string)
data['time'] = self.event_time
# we assume json will be decoded as object, eg:
if type(data) is dict:
handler(data, self.Event_dictionary_list, self.cached_timestamp)
#Trying to return object, but no luck
return self.Event_dictionary_list
def run():
print('http server is starting...')
port = 8000
server_address = ('localhost', port)
httpd = HTTPServer(server_address, server)
print('http server is listening on port %d' % port)
httpd.serve_forever()
if __name__ == '__main__':
run()
我的请求如下所示：
requests.post(url = 'http://localhost:8000', json = {'key': 'value'})
答：
1赞
Andrej Kesely
11/1/2023
#1
首先：你需要返回一些字符串数据 - 而不是 Python 对象，例如 list、dict 等。例如，您可以将 python 对象编码为 Json 字符串并发送此字符串。
第二：我建议将 HTTPServer 子类化，并将“持久”数据放在那里。
例：
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
class MyServer(HTTPServer):
def __init__(self, *args, **kwargs):
HTTPServer.__init__(self, *args, **kwargs)
self.my_list = []
class MyCustomHandler(BaseHTTPRequestHandler):
def do_POST(self):
# read the data in chunks:
max_chunk_size = 10 * 1024 * 1024
size_remaining = int(self.headers["content-length"])
L = []
while size_remaining:
chunk_size = min(size_remaining, max_chunk_size)
chunk = self.rfile.read(chunk_size)
if not chunk:
break
L.append(chunk)
size_remaining -= len(L[-1])
try:
data = json.loads(b"".join(L).decode("utf-8"))
except:
self.send_response(500)
self.send_header("Content-length", "0")
self.end_headers()
else:
self.server.my_list.append(data)
response = json.dumps(self.server.my_list, indent=4).encode("utf-8")
self.send_response(200)
self.send_header("Content-type", "application/json")
self.send_header("Content-length", str(len(response)))
self.end_headers()
self.wfile.write(response)
def run():
print("http server is starting...")
httpd = MyServer(("localhost", 8000), MyCustomHandler)
print("http server is listening on port %d" % port)
httpd.serve_forever()
if __name__ == "__main__":
run()
运行代码将启动侦听服务器。localhost:8000
然后在其他终端中，您可以执行以下操作：
$ curl -X POST -H "Content-Type: application/json" -d '{"key": "value"}' http://localhost:8000
[
{
"key": "value"
}
]
执行后续的 curl 命令将添加到服务器端的列表中并返回它。
评论