Use XCOM Value In Operators

Question

I want to use XCOM values as a parameter of my Operator.

Firstly, was executed OracleReadOperator, which read table from db, and return values.

This is value in XCOM:

[{'SOURCE_HOST': 'TEST_HOST'}]

Using this function I want to get value from xcom

def print_xcom(**kwargs):
    ti = kwargs['ti']
    ti.xcom_pull(task_ids='task1')

Then use values as as parameter:

with DAG(
        schedule_interval='@daily',
        dagrun_timeout=timedelta(minutes=120),
        default_args=args,
        template_searchpath=tmpl_search_path,
        catchup=False,
        dag_id='test'
) as dag:
    
    test_l = OracleLoadOperator(
            task_id = "task1",
            oracle_conn_id="orcl_conn_id",
            object_name='table'
    )
    test_l
    
    def print_xcom(**kwargs):
        ti = kwargs['ti']
        ti.xcom_pull(task_ids='task1', value='TARGET_TABLE')


    
    
    load_from_db = MsSqlToOracleTransfer(
                task_id= 'task2',
                mssql_conn_id = "{task_instance.xcom_pull(task_ids='task1') }",
                oracle_conn_id = 'conn_def_orc',
                sql= 'test.sql',
                oracle_table = "oracle_table"
        tasks.append(load_from_db)

I don't know do I need print_xcom function. Or I can get value without it, if yes how? I got this error:

airflow.exceptions.AirflowNotFoundException: The conn_id `{ task_instance.xcom_pull(task_ids='task1') }` isn't defined

Answer 1

To resolve the immediate NameError exception, Jinja expressions are strings so the arg for oracle_table needs to be updated to:

oracle_table = "{{ task_instance.xcom_pull(task_ids='print_xcom', key='task1') }}"

EDIT

(Since the question and problem changed.)

Only template_fields declared for an operator can use Jinja expressions. It looks like MsSqlToOracleTransfer is a custom operator and if you want to use a Jinja template for the mssql_conn_id arg, it needs to be declared as part of template_fields otherwise the literal string is used as the arg value (which is what you're seeing). Also you need the expression in the "{{ ... }}" format as well.

Here is some guidance on Jinja templating with custom operators if you find it helpful.

However, it seems like there is more to this picture than what we have context for. What is task1? Are you simply trying to retrieve a connection ID? What is it exactly you are trying to accomplish accessing XComs in the DAG?

Answer 2

The Airflow tasks has implemented the output attribute that returns an intance of XComArs. For example:

def push_xcom(ti):
    return {"key": "value"}


def pull_xcom(input):
    print(f'XCom: {input}')


with DAG(...) as dag:
    start = PythonOperator(task_id='dp_start', python_callable=push_xcom)
    
    end = PythonOperator(task_id='dp_start', python_callable=pull_xcom,
                         op_kwargs={'input': start.output})

    start >> end

Maybe you could use test_l.output in load_from_db.mssql_conn_id, But I think in the case of whatever_conn_id parameters, the value should be the ID of an Airflow connection.