如何将XCOM值传入PythonVirtualenvOperator的op_args参数？

你当前代码中直接在op_args里使用ti.xcom_pull()会触发错误——因为DAG解析阶段（代码加载时）ti（任务实例对象）还未创建，只有任务运行阶段才能访问该对象。下面提供两种可行的解决方案：

方法1：使用Jinja模板语法拉取XCom值 #

PythonVirtualenvOperator的op_args支持Jinja模板渲染，可直接通过模板语法引用XCom中的值：

from airflow import DAG
from airflow.decorators import task
from airflow.operators.python import PythonVirtualenvOperator
from airflow.utils.dates import days_ago
from airflow.hooks.base import BaseHook
from datetime import timedelta

from test_process import process_test_data


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=1),
    'catchup':False    
}

connection = BaseHook.get_connection('test_flow')
db_user = connection.login
db_pass = connection.password


with DAG(
    'test',
    default_args=default_args,
    schedule_interval=None,
    start_date=days_ago(1)  
) as dag:
    # 示例：上游任务推送数据库信息到XCom
    @task
    def push_db_info():
        return {"db_host": "your_db_host", "db_database": "your_db_name"}
    
    push_db_info_task = push_db_info()
    
    process_test_data_task = PythonVirtualenvOperator(
        task_id='process_test_data',
        python_callable=process_test_data,
        # 通过Jinja模板从XCom拉取指定值
        op_args=[
            db_user,
            db_pass,
            "{{ ti.xcom_pull(task_ids='push_db_info', key='return_value')['db_host'] }}",
            "{{ ti.xcom_pull(task_ids='push_db_info', key='return_value')['db_database'] }}"
        ],
        python_version=3.8,
        system_site_packages=True,
        requirements=['psycopg2', 'configparser'],
    )

    push_db_info_task >> process_test_data_task

方法2：使用TaskFlow API（Airflow 2.x+推荐） #

如果使用Airflow 2.x及以上版本，用@task.virtualenv装饰器替代PythonVirtualenvOperator，可直接通过参数传递上游任务的返回值，无需手动处理XCom：

from airflow import DAG
from airflow.decorators import task
from airflow.hooks.base import BaseHook
from datetime import timedelta, days_ago


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=1),
    'catchup':False    
}

connection = BaseHook.get_connection('test_flow')
db_user = connection.login
db_pass = connection.password


with DAG(
    'test',
    default_args=default_args,
    schedule_interval=None,
    start_date=days_ago(1)  
) as dag:
    @task
    def push_db_info():
        return {"db_host": "your_db_host", "db_database": "your_db_name"}
    
    # 用@task.virtualenv定义虚拟环境任务
    @task.virtualenv(
        python_version=3.8,
        system_site_packages=True,
        requirements=['psycopg2', 'configparser']
    )
    def process_test_data_task(user, password, host, db):
        # 在虚拟环境中导入并执行目标函数
        from test_process import process_test_data
        process_test_data(user, password, host, db)
    
    db_info = push_db_info()
    # 直接传递参数，Airflow自动处理XCom传递
    process_test_data_task(db_user, db_pass, db_info["db_host"], db_info["db_database"])

注意事项 #

• 方法1中，拉取XCom时需指定task_ids对应上游任务；如果上游任务通过return返回数据，默认XCom的key为return_value。
• 方法2的TaskFlow API更简洁直观，是Airflow 2.x的推荐写法，无需手动管理XCom的存储与拉取。

内容的提问来源于stack exchange，提问作者Vlad Vlad