【问题标题】:Python Falcon - get POST dataPython Falcon - 获取 POST 数据
【发布时间】:2026-01-17 17:00:01
【问题描述】:

我尝试在我的项目中使用 falcon 包。问题是我没有找到从 HTTP 发布请求中获取正文数据的方法。

我使用了示例中的代码,但 req.stream.read() 没有按预期返回 JSON。

代码是:

raw_json = req.stream.read()
result.json(raw_json, encoding='utf-8')
resp.body = json.dumps(result_json, encoding='utf-8')

如何获取 POST 数据?

感谢您的帮助

【问题讨论】:

  • 我花了很长时间才找到这个答案:*.com/questions/34618619/…
  • 您要查找的字段是req.media。请参阅下面的答案。
  • req.stream.read() 只能使用一次。把它想象成从文件流中读取,除非你f.seek(0),否则你不能再做f.read()。一样,只是请求流没有.seek()

标签: python python-3.x http-post falconframework falcon


【解决方案1】:

falcon 2 中,如果您使用 json 类型,请使用 req.media

例如:

import falcon
from json import dumps

class Resource(object):
    def on_post(self, req, resp, **kwargs):
        result = req.media
        # do your job
        resp.body = dumps(result)


api = falcon.API()

api.add_route('/test', Resource())

【讨论】:

    【解决方案2】:

    对问题的深入研究导致了以下linked issue on github。它指出 falcon 框架至少在其版本 0.3 中并使用 Python 2 不会将“发布”的数据解析为字符串,如果它们被恰当地转义的话。我们可以使用有关您尝试通过 POST 请求发送哪些数据以及以何种格式发送的更多信息,例如作为简单文本发送,或者使用 Header Information Content-Type:application/json,或者如果它的通过 HTML 表单。

    虽然从问题中不清楚确切的问题,但我仍然建议尝试使用bounded_stream 而不是stream,如下所示:

    raw_json = req.bounded_stream.read()
    result.json(raw_json, encoding='utf-8')
    resp.body = json.dumps(result_json, encoding='utf-8')
    

    官方文档建议在 Content-Length 未定义或 0 等不确定条件下使用 bounded_stream,或者完全缺少标头信息。

    bounded_stream在官方falcon documentation中描述如下。

    围绕流的类文件包装器,以规范不同 WSGI 服务器使用的本机输入对象之间的某些差异。特别是,bounded_stream 知道正文的预期 Content-Length,并且永远不会阻止越界读取,假设客户端在将数据传输到服务器时不会停止。

    Falcon 接收 HTTP 请求数据作为缓冲区对象,由 WSGI 包装器传递,该包装器从客户端接收数据,并且出于性能原因,它可能没有在数据之上运行正确的解析以转换为更可用的数据结构.

    【讨论】:

      【解决方案3】:

      非常感谢 Ryan(和 Prateek Jain)的回答。

      解决方法是简单地输入app.req_options.auto_parse_form_urlencoded=True。例如:

      import falcon
      
      class ThingsResource(object):
          def on_post(self, req, resp):
              value = req.get_param("value", required=True)
              #do something with value
      
      app = falcon.API()
      app.req_options.auto_parse_form_urlencoded=True
      
      things = ThingsResource()
      
      app.add_route('/things', things)
      

      【讨论】:

        【解决方案4】:

        您要查找的字段名称有些混乱,但它是req.media

        返回请求流的反序列化形式。调用时,它将尝试使用 Content-Type 标头以及通过 falcon.RequestOptions 配置的媒体类型处理程序来反序列化请求流。

        如果请求是 JSON,req.media 已经包含一个 python dict。

        【讨论】:

          【解决方案5】:

          我在 falcon 框架的 request.py 中添加了更改,以解析 application/x-www-form-urlencoded 和 multipart/from-data。 我已经提出了拉取请求 - https://github.com/falconry/falcon/pull/1236 但它还没有合并到 master 中。 检查这个 - https://github.com/branelmoro/falcon

          我添加了新代码来解析 POST、PUT 和 DELETE application/x-www-form-urlencoded 和 multipart/form-data。 文本字段将在 req.form_data 字典中可用,上传文件缓冲流将在 req.files 字典中可用。

          我希望这将有助于分别访问 POST 和 GET 参数,我们也将能够上传文件。 更改的好处是它不会将整个上传的文件加载到内存中。

          下面是示例代码,展示如何使用 POST、PUT 和 DELETE application/x-www-form-urlencoded 和 multipart/form-data:

          import falcon
          
          class Resource(object):
          
              def on_post(self, req, resp):
          
                  # req.form_data will return dictionary of text field names and their values
                  print(req.form_data)
          
                  # req.form_data will return dictionary of file field names and
                  # their buffer class FileStream objects as values
                  print(req.files)
          
                  # support we are uploading a image.jpg in `pancard` file field then
                  # req.files["pancard"] will be FileStream buffer object
          
                  # We can use set_max_upload_size method to set maximum allowed
                  # file size let say 1Mb = 1*1024*1024 bytes for this file
          
                  req.files["pancard"].set_max_upload_size(1*1024*1024)
          
                  # We can use uploadto method to upload file on required path (Note: absolute filepath is required)
                  # This method returns boolean - `True` on successful upload
                  # and if upload is unsuccessful then it returns `False` and sets error on failure.
                  path = "/tmp/" + req.files["pancard"].name
          
                  response = req.files["pancard"].uploadto("/tmp/" + path)
          
                  print(response)
          
                  # Once file is uploaded sucessfully, we can check it's size
                  print(req.files["pancard"].size)
          
                  # If file is not uploaded sucessfully, we can check it's error
                  print(req.files["pancard"].error)
          
                  resp.body = "Done file upload"
          
                  resp.status = falcon.HTTP_200
          
          # falcon.API instances are callable WSGI apps
          app = falcon.API()
          
          things = Resource()
          
          # things will handle post requests to the '/post_path' URL path
          app.add_route('/post_path', things)
          

          如果您有任何疑问,请告诉我。

          【讨论】:

          • 猎鹰杀了我。为什么接收 Post 数据这么难。我收到AttributeError: 'Request' object has no attribute 'form_data'
          【解决方案6】:

          到目前为止...对我来说 bounded_stream.read() 和 stream.read() 都以 str 类型获取发布的数据。到目前为止,我只找到了解决此问题的一种方法:

          def on_post(self, req, resp):
              posted_data = json.loads(req.stream.read())
              print(str(type(posted_data)))
              print(posted_data)
          

          收到发布的数据后将字符串加载到 json 字典中是我唯一能想到的解决方案

          【讨论】:

            【解决方案7】:

            这是我在设计 API 时使用的东西。

            import falcon
            import json
            
            
            class VerifierResource():
                def on_post(self, req, resp):
                    credentials = json.loads(req.stream.read())
            
                    if credentials['username'] == USER \
                      and credentials['passwd'] == PASSWORD:
                        resp.body = json.dumps({"status": "verified"})
                    else:
                        resp.body = json.dumps({"status": "invalid"})
            
            api = falcon.API()
            api.add_route('/verify', VerifierResource())
            

            这会返回一个带有相应响应正文的序列化 JSON。

            【讨论】:

              【解决方案8】:

              有一种从正文获取媒体的示例方法。我用post方法获取body:

              def on_post(req,resp)
                  arguments = {}
                          # get body media on post method
                          body = req.get_media()
                          if 'something' in body:
                              arguments['something'] = body['something']
              

              发送正文内容类型 Media-Type 并打印响应或在代码中使用,但如果要发送 JSON 正文,您的代码应包含提供 JSON 参数。

              如果您有任何疑问,请告诉我。

              【讨论】: