如何用SimpleHttpOperator传JSON至Cloud Function 2nd Gen并正确接收

我的任务代码如下：

this_is_a_task = SimpleHttpOperator(
    task_id= 'task_id',
    method='POST',
    http_conn_id='conn_id',
    endpoint='/?test=foo',
    # data={"test": "foo"},
    headers={"Content-Type": "application/json"}
)

在Cloud Function端，我尝试两种方式获取参数：

# catching data
# test_data = request.get_json().get('test')
# print('test: {}'.format(test))

# catching end point
test_endpoint = request.args.get('test')
print('test: {}'.format(test))

第二种方式（request.args.get('test')）可以正常工作，但使用第一种方式（request.get_json().get('test')）时会出现400请求错误。

请问：如果不使用SimpleHttpOperator中的endpoint变量传递参数，如何正确接收data变量传递的JSON对象？

解决办法 #

1. 修正SimpleHttpOperator的参数配置

问题核心是data参数的默认处理逻辑：当你传入字典类型的data时，SimpleHttpOperator会自动把它编码成x-www-form-urlencoded格式的请求体，但你设置了Content-Type: application/json，服务器收到后按JSON解析自然失败，返回400错误。

有两种正确配置方式：

• 推荐：使用json参数（Airflow 1.10.12及以上版本支持）
  这个参数会自动把字典序列化为JSON字符串，同时自动设置Content-Type: application/json，无需手动添加headers：

  this_is_a_task = SimpleHttpOperator(
      task_id='task_id',
      method='POST',
      http_conn_id='conn_id',
      endpoint='/',  # 移除query参数
      json={"test": "foo"}  # 直接传字典，自动处理序列化
  )
  

• 如果必须用data参数
  需要手动将字典序列化为JSON字符串，同时保留Content-Type头：

  import json
  
  this_is_a_task = SimpleHttpOperator(
      task_id='task_id',
      method='POST',
      http_conn_id='conn_id',
      endpoint='/',
      data=json.dumps({"test": "foo"}),  # 手动转成JSON字符串
      headers={"Content-Type": "application/json"}
  )
  

2. 修正Cloud Function端的代码

原代码存在变量名错误（print里的test未定义），同时要处理JSON解析失败的情况：

# catching data
test_data = request.get_json()
if test_data is not None:
    test_value = test_data.get('test')
    print('test: {}'.format(test_value))
else:
    print('请求体不是合法的JSON格式')

按以上配置调整后，Cloud Function就能正确接收通过json或序列化后的data传递的JSON对象了。

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