Flask——基于python完整实现客户端和服务器后端流式请求及响应
创始人
2025-01-20 15:35:55
0

文章目录

    • 本地客户端
    • Flask服务器后端
    • 客户端/服务器端流式接收[打字机]效果

看了很多相关博客,但是都没有本地客户端和服务器后端的完整代码示例,有的也只说了如何流式获取后端结果,基本没有讲两端如何同时实现流式输入输出,特此整理总结,给大家交流学习和使用!

本地客户端

  • requests.post得到流式回复的重要参数:
    • stream:需要设置为True;
    • response.iter_content:使用该函数接收返回的流式数据。
import requests import time import json  def generate_stream_data():     # 假设这是要发送的文本列表     is_end = False     lines = ["Hello", "world", "this", "is", "a", "stream", "of", "text"]     for line in lines:         print(line)         if lines.index(line) == len(lines) - 1:             is_end = True         yield json.dumps({'line': line, 'is_end': is_end}) + '\n'         time.sleep(0.5)         # 模拟数据处理时间  def get_stream_response(response):     # 流式接收response     rec_data_list = []     temp_data = ''     for chunk in response.iter_content(chunk_size=1):         temp_data += chunk.decode('utf-8')         if temp_data.endswith('\n'):             temp_json = json.loads(temp_data)             rec_data_list.append(temp_json)             print(temp_data)             temp_data = ''             if temp_json['is_end']:                 break     print(rec_data_list)     print("----------------------------")     print(temp_data)     return rec_data_list  def stream_upload(url):          # 流式接收response     response = requests.post(url, data=generate_stream_data(), stream=True)          final_response = get_stream_response(response)          return final_response  url = 'http://127.0.0.1:5000/stream' response = stream_upload(url) 

Flask服务器后端

  • flask.request流式获取数据::
    • 使用request.stream.read读取数据,而不是get_data()等一次性函数。
from flask import Flask, Response, request import time import json import requests  app = Flask(__name__)  def process_stream_data(stream_data):     # 假设这是要发送的数据     print("开始生成新的数据流")     is_end = False     print(stream_data)     for idx, line in enumerate(stream_data):         if idx == len(stream_data)-1:             is_end = True         print(line)         yield json.dumps(line)+"\n"         time.sleep(0.5)         # 模拟数据处理时间  def get_stream_request(chunk_size=1):     req_data_list = []     temp_data = ''     while True:         chunk = request.stream.read(chunk_size)         temp_data += chunk.decode('utf-8')         if temp_data.endswith('\n'):             temp_json = json.loads(temp_data)             req_data_list.append(temp_json)             print(temp_data)             temp_data = ''             if temp_json['is_end']:                 return req_data_list  @app.route('/stream', methods=['POST']) def stream_text():          data = get_stream_request()      print("----------------------------")          return Response(process_stream_data(data))  if __name__ == "__main__":     app.run(host='0.0.0.0', port=5000, debug=True) 

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

请添加图片描述

相关内容

热门资讯

专业讨论!德扑之星真破解套路(... 专业讨论!德扑之星真破解套路(辅助挂)软件透明挂(有挂了解)-哔哩哔哩;人气非常高,ai更新快且高清...
每日必看!智星德州菠萝外挂检测... 每日必看!智星德州菠萝外挂检测(辅助挂)软件透明挂(有挂教学)-哔哩哔哩1、玩家可以在智星德州菠萝外...
透视透明挂!轰趴十三水有后台(... 轰趴十三水有后台赢率提升策略‌;透视透明挂!轰趴十三水有后台(辅助挂)软件透明挂(有挂详情)-哔哩哔...
发现玩家!德扑ai助手软件(辅... 发现玩家!德扑ai助手软件(辅助挂)透视辅助(有挂教学)-哔哩哔哩;玩家在德扑ai助手软件中需先进行...
一分钟了解!x-poker辅助... 一分钟了解!x-poker辅助软件(辅助挂)辅助透视(有挂攻略)-哔哩哔哩1、每一步都需要思考,不同...
一分钟揭秘!德州最新辅助器(辅... 一分钟揭秘!德州最新辅助器(辅助挂)透视辅助(有挂攻略)-哔哩哔哩;德州最新辅助器最新版本免费下载安...
玩家攻略推荐!德州辅助(辅助挂... 玩家攻略推荐!德州辅助(辅助挂)辅助透视(有挂了解)-哔哩哔哩是由北京得德州辅助黑科技有限公司精心研...
揭秘真相!pokernow德州... 《揭秘真相!pokernow德州(辅助挂)辅助透视(有挂介绍)-哔哩哔哩》 pokernow德州软件...
五分钟了解!德州之星辅助器(辅... 五分钟了解!德州之星辅助器(辅助挂)辅助透视(有挂透明)-哔哩哔哩1、很好的工具软件,可以解锁游戏的...
推荐一款!pokermaste... 1、推荐一款!pokermaster有外挂(辅助挂)透视辅助(有挂教学)-哔哩哔哩;详细教程。2、p...