Airflow中datetime.strptime解析XCom变量报错原因咨询

问题场景 #

我有一个包含两个任务的简单DAG：一个任务返回转为字符串类型的datetime，另一个任务使用该值运行TimeSensorAsync延迟算子。代码如下：

from datetime import datetime, timedelta
from airflow.decorators import task
from airflow.sensors.time import TimeSensorAsync

@task
def get_scheduled_upgrade_time(data: dict) -> str:
    return str(datetime.utcnow() + timedelta(seconds=60))

wait = TimeSensorAsync(
    task_id="waiting_until_scheduled_time",
    target_time=(datetime.strptime("{{ task_instance.xcom_pull(task_ids='get_scheduled_upgrade_time', key='return_value') }}", "%Y-%m-%d %H:%M:%S.%f").time())
)

运行时Airflow UI抛出错误：

ValueError: time data "{{ task_instance.xcom_pull('get_scheduled_upgrade_time') }}" does not match format '%Y-%m-%d %H:%M:%S.%f'

错误原因 #

• DAG解析阶段提前执行了Python代码：datetime.strptime是纯Python代码，会在Airflow加载DAG文件的解析阶段直接运行，而此时Airflow的模板变量{{ task_instance.xcom_pull(...) }}还没被渲染成实际的XCom返回值，还是原始的模板字符串。strptime试图解析这个模板字符串，自然不符合指定的时间格式，直接报错。
• TimeSensorAsync的target_time不支持模板渲染：就算跳过strptime直接把模板字符串传给target_time，这个参数本身也不属于Airflow的模板化字段，运行时不会被替换成实际的时间值，依然无法正常工作。

修正方案 #

要让时间解析逻辑在任务运行阶段执行，而非DAG解析阶段，推荐两种解决方式：

方式1：用可调用函数动态获取XCom值 #

让target_time接收一个可调用函数，在函数内部获取XCom值并解析成时间对象：

from datetime import datetime, timedelta
from airflow.decorators import task, dag
from airflow.sensors.time import TimeSensorAsync
from airflow.operators.python import get_current_context

def get_target_time():
    # 获取当前任务上下文
    context = get_current_context()
    # 从XCom拉取前一个任务的返回值
    xcom_time_str = context["task_instance"].xcom_pull(task_ids='get_scheduled_upgrade_time', key='return_value')
    # 解析成datetime后提取time对象
    return datetime.strptime(xcom_time_str, "%Y-%m-%d %H:%M:%S.%f").time()

@dag(schedule_interval=None, start_date=datetime(2024, 1, 1))
def upgrade_dag():
    @task
    def get_scheduled_upgrade_time(data: dict) -> str:
        return str(datetime.utcnow() + timedelta(seconds=60))

    upgrade_time_task = get_scheduled_upgrade_time({})
    
    wait = TimeSensorAsync(
        task_id="waiting_until_scheduled_time",
        target_time=get_target_time,
        trigger_rule="all_success"
    )

    upgrade_time_task >> wait

upgrade_dag()

方式2：改用DateTimeSensorAsync（更推荐） #

如果需求是等待到具体的日期时间点（而非每天重复的固定时间），DateTimeSensorAsync更合适，它的target_datetime参数支持模板渲染，可以直接使用XCom值：

from datetime import datetime, timedelta
from airflow.decorators import task, dag
from airflow.sensors.date_time import DateTimeSensorAsync

@dag(schedule_interval=None, start_date=datetime(2024, 1, 1))
def upgrade_dag():
    @task
    def get_scheduled_upgrade_time(data: dict) -> str:
        return str(datetime.utcnow() + timedelta(seconds=60))

    upgrade_time_task = get_scheduled_upgrade_time({})
    
    wait = DateTimeSensorAsync(
        task_id="waiting_until_scheduled_time",
        target_datetime="{{ task_instance.xcom_pull(task_ids='get_scheduled_upgrade_time', key='return_value') }}"
    )

    upgrade_time_task >> wait

upgrade_dag()

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