从 xcom 气流任务实例数据创建 json

Question

我在气流 dag 中有两个任务。第一个任务点击 POST URL，它的结果成为第二个任务的参数。我可以点击第一个 URL 并将其发送到另一个任务。问题是，我无法从第二个创建正确的 json。

def create_dag(dag_id,
               schedule,
               default_args):
    def task1(**kwargs):
        res = requests.post('some url')
        return res

    def task2(**kwargs):
        ti = kwargs['ti']
        v1 = ti.xcom_pull(key=None, task_ids='taks1')
        for rec in v1:
            decode = rec.decode("utf-8")
            logger.error(decode)

第一个任务返回json：

{
    "data": [
        {
            "id": 36,
            "idPercent": 12.67605633802817,
            "idPerson": [
                "Washburn"
            ]
        },
        {
            "id": 37,
            "idPercent": 13.028169014084508,
            "idPerson": [
                "Nicole"
            ]
        }
    ]
}

rec.decode("utf-8") 打印出来的也是一样的。如果我在 rec.decode("utf-8") 上执行 json.loads，我就会开始出错。我想创建正确的 json，然后为数据的每个数组值发送一个 POST 请求。

我想要类似的东西：

for rec in v1['data']:
     requests.post(url, rec)

但我无法创建 json 并从中提取数据。

Answer 1

在 Airflow 中，xcom 作为字符串传递，在您的情况下，v1 将是一个字符串，即使它是从 xcom 对象加载的。解决此问题的一种方法是使用 ast 模块，该模块将字符串值转换为其“正确”类型。请注意，有时由于标点符号或类似的原因，它很难解释正确的类型。

import ast

v1 = ast.literal_eval(ti.xcom_pull(key=None, task_ids='taks1'))