Airflow DAG中prev_execution_date_success判空逻辑异常问题

问题现象 #

编写Airflow DAG时，任务为全新状态、从未运行过时，模板变量prev_execution_date_success会返回空值，该取值直接传入SQL会触发查询执行异常。最初通过user_defined_macros配置自定义处理宏做兜底，代码如下：

from datetime import datetime, timedelta

def __get_last_execution_time(execution_date: str) -> str:
    return (datetime.now() - timedelta(hours=1)).isoformat() if execution_date is None else execution_date

SQL脚本中按如下方式调用自定义宏：

WHERE created_at >= TIMESTAMP('{{ get_last_execution_time(prev_execution_date_success) }}')

上述逻辑运行时始终返回空值，即便prev_execution_date_success（即传入函数的execution_date参数）看起来取值为None，也无法触发判空分支。在自定义函数中添加调试日志确认函数执行状态，日志代码如下：

import logging

logging.info("prev_execution_date_success: %s", execution_date)
logging.info("test 1: %s", execution_date is None)
logging.info("test 2: %s", execution_date == 'None')
logging.info("Type of execution_date: %s", type(execution_date))

日志输出结果：

prev_execution_date_success: None
test 1: False
test 2: False
Type of execution_date: <class 'Proxy'>

核心疑问：Airflow DAG模板渲染上下文中，常规None判断逻辑失效的原因是什么，该Proxy类型对象应如何正确判空？

失效原因 #

Airflow模板渲染能力基于Jinja2实现，当模板变量无有效值（比如首次运行的任务无上次成功执行时间）时，Jinja2不会直接返回Python原生的None对象，而是返回一个Undefined类型的代理对象（即日志中打印的Proxy类型）。
该代理对象重写了字符串输出方法，因此日志打印时会显示为None，但它本质既不是Python原生None，也不是内容为'None'的字符串，因此execution_date is None、execution_date == 'None'两类判断都会返回False。

正确处理方案 #

• 布尔值判断法：Jinja2的Undefined代理对象在布尔上下文中会被判定为False，直接将判空逻辑修改为if not execution_date即可覆盖原生None、Undefined代理、空字符串等空值场景，修改后的宏代码如下：

from datetime import datetime, timedelta

def get_last_execution_time(execution_date) -> str:
    return (datetime.now() - timedelta(hours=1)).isoformat() if not execution_date else execution_date

注意：自定义宏不要使用双下划线开头的函数名，这类命名在Jinja2上下文存在访问限制，可能引发调用异常。

• 类型判断法：如果需要精准区分空值类型，避免0、空字符串等其他假值误触发兜底逻辑，可以引入Jinja2的Undefined类型做显式判断，同时覆盖原生None场景：

from datetime import datetime, timedelta
from jinja2 import Undefined

def get_last_execution_time(execution_date) -> str:
    if execution_date is None or isinstance(execution_date, Undefined):
        return (datetime.now() - timedelta(hours=1)).isoformat()
    return execution_date

• 模板内置过滤器法：无需编写自定义宏，直接在SQL模板中使用Jinja2内置的default过滤器设置空值兜底，写法更简洁：

WHERE created_at >= TIMESTAMP('{{ prev_execution_date_success | default((macros.datetime.now() - macros.timedelta(hours=1)).isoformat()) }}')

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