ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flask——基于python完整实现客户端和服务器后端流式哀求及响应 [打印本页]

---

作者: 缠丝猫    时间: 2024-9-11 22:22
标题: Flask——基于python完整实现客户端和服务器后端流式哀求及响应
看了很多相关博客，但是都没有本地客户端和服务器后端的完整代码示例，有的也只说了怎样流式获取后端结果，基本没有讲两头怎样同时实现流式输入输出，特此整理总结，给各人交流学习和使用！
  本地客户端

• requests.post得到流式复兴的紧张参数：
  
  
  
  • stream：需要设置为True；
    
  • response.iter_content：使用该函数接收返回的流式数据。
    
  
  

1. import requests
2. import time
3. import json
5. def generate_stream_data():
6.     # 假设这是要发送的文本列表
7.     is_end = False
8.     lines = ["Hello", "world", "this", "is", "a", "stream", "of", "text"]
9.     for line in lines:
10.         print(line)
11.         if lines.index(line) == len(lines) - 1:
12.             is_end = True
13.         yield json.dumps({'line': line, 'is_end': is_end}) + '\n'
14.         time.sleep(0.5)
15.         # 模拟数据处理时间
17. def get_stream_response(response):
18.     # 流式接收response
19.     rec_data_list = []
20.     temp_data = ''
21.     for chunk in response.iter_content(chunk_size=1):
22.         temp_data += chunk.decode('utf-8')
23.         if temp_data.endswith('\n'):
24.             temp_json = json.loads(temp_data)
25.             rec_data_list.append(temp_json)
26.             print(temp_data)
27.             temp_data = ''
28.             if temp_json['is_end']:
29.                 break
30.     print(rec_data_list)
31.     print("----------------------------")
32.     print(temp_data)
33.     return rec_data_list
35. def stream_upload(url):
36.    
37.     # 流式接收response
38.     response = requests.post(url, data=generate_stream_data(), stream=True)
39.    
40.     final_response = get_stream_response(response)
41.    
42.     return final_response
44. url = 'http://127.0.0.1:5000/stream'
45. response = stream_upload(url)

复制代码

Flask服务器后端

• flask.request流式获取数据：：
  
  
  
  • 使用request.stream.read读取数据，而不是get_data()等一次性函数。
    
  
  

1. from flask import Flask, Response, request
2. import time
3. import json
4. import requests
6. app = Flask(__name__)
8. def process_stream_data(stream_data):
9.     # 假设这是要发送的数据
10.     print("开始生成新的数据流")
11.     is_end = False
12.     print(stream_data)
13.     for idx, line in enumerate(stream_data):
14.         if idx == len(stream_data)-1:
15.             is_end = True
16.         print(line)
17.         yield json.dumps(line)+"\n"
18.         time.sleep(0.5)
19.         # 模拟数据处理时间
21. def get_stream_request(chunk_size=1):
22.     req_data_list = []
23.     temp_data = ''
24.     while True:
25.         chunk = request.stream.read(chunk_size)
26.         temp_data += chunk.decode('utf-8')
27.         if temp_data.endswith('\n'):
28.             temp_json = json.loads(temp_data)
29.             req_data_list.append(temp_json)
30.             print(temp_data)
31.             temp_data = ''
32.             if temp_json['is_end']:
33.                 return req_data_list
35. @app.route('/stream', methods=['POST'])
36. def stream_text():
37.    
38.     data = get_stream_request()
40.     print("----------------------------")
41.    
42.     return Response(process_stream_data(data))
44. if __name__ == "__main__":
45.     app.run(host='0.0.0.0', port=5000, debug=True)

复制代码

客户端/服务器端流式接收[打字机]结果

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4