使用 SimpleHttpOperator 传递数据以触发第二代云功能答案

【问题标题】：pass data with SimpleHttpOperator to trigger cloud function 2nd gen使用 SimpleHttpOperator 传递数据以触发第二代云功能
【发布时间】：2022-10-18 15:00:00
【问题描述】：

我有以下任务：

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"}

在云功能方面，我试图通过以下两种方式捕获参数：

# 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 变量，我如何才能将json 对象传递给data 变量？

【问题讨论】：

你可以试试这个test_data = request.get_json().get('data')吗？
我确实已经尝试过了，结果相同！
你能分享你遇到的详细错误/日志吗？

标签： google-cloud-functions google-cloud-composer

【解决方案1】：

我试图复制您的问题，并且基于此documentation，当您使用 json 数据调用POST 时，您需要添加json.dumps。然后提供 authentication credentials 作为存储在 Authorization 标头中的 Google 生成的 ID 令牌。

请参见下面的示例代码：

import datetime
import json

from airflow import models
from airflow.operators import bash
from airflow.providers.http.operators.http import SimpleHttpOperator

YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    'owner': 'Composer Example',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': YESTERDAY,
}

with models.DAG(
        'composer_quickstart',
        catchup=False,
        default_args=default_args,
        schedule_interval=datetime.timedelta(days=1)) as dag:

    # Print the dag_run id from the Airflow logs
    gen_auth = bash.BashOperator(
        task_id='gen_auth', bash_command='gcloud auth print-identity-token '
    )

    auth_token = "{{ task_instance.xcom_pull(task_ids='gen_auth') }}"
    
    this_is_a_task = SimpleHttpOperator(
        task_id='task_id',
        method='POST',
        http_conn_id='cf_conn1',
        data=json.dumps({"test": "foo"}),
        headers={"Content-Type": "application/json","Authorization": "Bearer " + auth_token}
    )
    
    gen_auth >> this_is_a_task

在云功能方面尝试使用以下示例代码：

test_data = request.get_json().get('test')
print(test_data)
    
return test_data

您还可以使用以下 curl 命令测试您的功能：

curl -i -X POST -H "Content-Type:application/json"  -d '{"test": "foo"}' "Authorization: bearer $(gcloud auth print-identity-token)" https://function-5-k6ssrsqwma-uc.a.run.app

【讨论】：