0

我需要从 Ariflow dag 进行 API 调用。

下面是我在代码中使用的示例 json_string：

    def api_call_func(**context):
        source_file_name = context["dag_run"].conf["source_file_names"]
    json_data = { "filenames":[
                 {"FileName":[f'Source_credit_{{{{ ts_nodash }}}}']}
                  ]
                }

        json_string = json.dumps(json_data, 
                        skipkeys = True, 
                        allow_nan = True)
        requests.post(url = API_URL, data = json_string)

    api_task = PythonOperator(
    task_id='api_task',
    provide_context=True,
    python_callable=api_call_func,
    dag=dag,
    )

但它导致以下响应：

<filenames>
         <FileName>f'Source_credit_{{ ts_nodash }}</FileName>
</filenames>

以下是 currentDateTime 所需的响应：

<filenames>
         <FileName>f'Source_credit_20210408010223</FileName> 
</filenames>

如何在 json 中传递 ts_nodash 宏？或者如何在 json 中传递 dag 的执行日期时间？

4

1 回答 1

2

仅当包含模板的字符串作为参数传递给运算符的模板化参数时，才会处理模板。例如，op_args参数 PythonOperator。

在您的情况下， 的值作为参数ts_nodash传递给. 因此，您可以使用该参数来访问它：api_call_func()PythonOperatorcontext

json_data = {
    "ID": "A001-001",
    "SourceName": sourcefile,
    "filenames": [{"FileName": [f"Source_credit_{context['ts_nodash']}"]}],
}

请注意，可以以相同的方式访问其余默认变量的值。

于 2021-04-11T22:14:31.620 回答