问气流Xcom未得到解决返回task_instance字符串

EN

from datetime import datetime

import simplejson as json
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator   
from airflow.operators.python_operator import PythonOperator
from airflow.providers.http.operators.http import SimpleHttpOperator
from google.auth.transport.requests import Request

default_airflow_args = {
    "owner": "divyaansh",
    "depends_on_past": False,
    "start_date": datetime(2022, 5, 18),        
    "retries": 0,
    "schedule_interval": "@hourly",
}

project_configs = {
    "project_id": "test",
    "conn_id": "google_cloud_storage_default",
    "bucket_name": "test-transfer",
    "folder_name": "processed-test-rdf",
}


def get_config_vals(**kwargs) -> dict:
    """
    Get config vals from airlfow variable and store it as xcoms

    """

    task_instance = kwargs["task_instance"]

    task_instance.xcom_push(key="http_con_id", value="gcp_cloud_function")


def generate_api_token(cf_name: str):
    """
    generate token for api request
    """
    import google.oauth2.id_token    
    
    request = Request()

    target_audience = f"https://us-central1-test-a2h.cloudfunctions.net/{cf_name}"

    return google.oauth2.id_token.fetch_id_token(
        request=request, audience=target_audience
    )


with DAG(
    dag_id="cf_test",
    default_args=default_airflow_args,
    catchup=False,
    render_template_as_native_obj=True,
) as dag:

    start = DummyOperator(task_id="start")

    config_vals = PythonOperator(
        task_id="get_config_val", python_callable=get_config_vals, provide_context=True
    )

    ip_data = json.dumps(
        {
            "bucket_name": project_configs["bucket_name"],
            "file_name": "dummy",
            "target_location": "/valid",
        }
    )

    conn_id = "{{ task_instance.xcom_pull(dag_id = 'cf_test',task_ids='get_config_val',key='http_con_id') }}"

    api_token = generate_api_token("new-cp")

    cf_task = SimpleHttpOperator(
        task_id="file_decrypt_and_validate_cf",
        http_conn_id=conn_id,
        method="POST",
        endpoint="new-cp",
        data=json.dumps(
            json.dumps(
                {
                    "bucket_name": "test-transfer",
                    "file_name": [
                        "processed-test-rdf/dummy_20220501.txt",
                        "processed-test-rdf/dummy_20220502.txt",                            
                    ],
                    "target_location": "/valid",
                }
            )
        ),
        headers={
            "Authorization": f"bearer {api_token}",
            "Content-Type": "application/json",                
        },
        do_xcom_push=True,
        log_response=True,
    )

    print("task new-cp", cf_task)   
    

    check_flow = DummyOperator(task_id="check_flow")

    end = DummyOperator(task_id="end")

start >> config_vals >> cf_task >> check_flow >> end