python-3.x - 无法在云函数中使用 gcp 云调度程序的 json 主体作为参数值？

Question

我有一个云调度程序，我用它来触发我的云函数作为 http 调用，在我的云函数中，我想形成一个查询(应该是动态的)。为此，我从云调度程序(Json Body)传递了一些参数，但是当我触发我的云函数时，它不会将来自云调度程序的参数值作为 json 主体。谁能帮我解决这个问题。

来自云调度程序的 json 正文:

{ 
   "unit":"QA",
   "interval":"3"
}

云函数代码:

def main(request):

    request_json = request.get_json(silent=True)
    request_args = request.args

    if request_json and 'unit' in request_json:
        retail_unit = request_json['unit']
    elif request_args and 'unit' in request_args:
        retail_unit = request_args['unit']
    else:
        unit = 'UAT'

    if request_json and 'interval' in request_json:
        interval = request_json['interval']
    elif request_args and 'interval' in request_args:
        interval = request_args['interval']
    else:
        interval = 1

    query = "select * from `myproject.mydataset.mytable` where unit='{}' and interval ={}".format(                                                                                                    
    unit,interval)
    client = bigquery.Client()
    job_config = bigquery.QueryJobConfig()
    dest_dataset = client.dataset(destination_dataset, destination_project)
    dest_table = dest_dataset.table(destination_table)
    job_config.destination = dest_table
    job_config.create_disposition = 'CREATE_IF_NEEDED'
    job_config.write_disposition = 'WRITE_APPEND'
    job = client.query(query, location='US', job_config=job_config)
    job.result()

注意:当我从云调度程序传递相同的变量作为 http url ( https://my-region-test-project.cloudfunctions.net/mycloudfunction?unit=QA&interval=3 ) 中的参数值时，它会起作用

Answer 1

您可以覆盖默认值 Content-Type 来自 creating the cron job使用 gcloud与国旗 --headers Content-Type=application/json .
例如:

gcloud scheduler jobs create http my_cron_job \
  --schedule="every 5 hours" \
  --uri="https://${ZONE}-${PROJECT_ID}.cloudfunctions.net/${FUNCTION_NAME}" \
  --http-method=POST \
  --message-body='{"foo": "bar"}' \
  --headers Content-Type=application/json

这似乎还不能从 GCP Console 级别使用。 08/2021 更新:现在似乎已经在 UI 中实现了:


或者，使用 force=True如果您使用 Flask，似乎会有所帮助:

request.get_json(force=True)

这是因为 Cloud Scheduler 似乎设置了默认值 Content-Type标题到 application/octet-stream .见 doc