使用Python Flask流式传输数据时出现AssertionError字节写入错误

核心问题定位 #

这个错误的本质是WSGI服务器要求流式响应的输出必须是字节类型，当前你通过yield输出的内容要么是字符串，要么编码方式不正确，导致类型不匹配。

具体修复步骤 #

• 确保流式生成器的每一个yield输出都是bytes类型
  在generateStockStatus函数中，直接对要输出的字符串做UTF-8编码：

  def generateStockStatus():
      # 示例业务逻辑：遍历流式数据
      for item in stock_data_stream:
          # 若为普通字符串，转成bytes后yield
          yield f"当前库存状态：{item}\n".encode('utf-8')
          # 若为Jinja模板片段，先渲染再编码
          rendered_html = render_template_string('<div>{{ status }}</div>', status=item)
          yield rendered_html.encode('utf-8')
  

• 规范Flask流式响应的返回方式
  返回响应时，明确指定content_type，同时确保生成器输出严格为bytes：

  @app.route('/stream-stock')
  def stream_stock_status():
      return app.response_class(
          generateStockStatus(),
          content_type='text/html; charset=utf-8'
      )
  

• 排查错误的编码尝试
  如果你之前对GET请求返回值调用.encode()触发TypeError，大概率是因为该返回值已经是bytes类型。先判断类型再处理：

  api_response = requests.get('your-api-url')
  # 直接获取bytes类型的响应内容
  content_bytes = api_response.content
  # 若需要先处理字符串，再编码
  content_str = api_response.text
  content_bytes = content_str.encode('utf-8')
  

• 验证模板流式渲染逻辑
  若通过Jinja模板流式输出，必须保证渲染后的结果是字符串，编码为bytes后再yield，避免直接输出模板对象或未编码的渲染结果。

常见误区规避 #

• 不要混合yield字符串和bytes，所有输出必须统一为bytes类型
• 不要对已经是bytes的内容再次调用.encode()，这会触发TypeError
• 确保响应的content_type包含charset声明，避免浏览器解码异常

内容的提问来源于stack exchange，提问作者Joeyohanlin