如何在Airflow Python Operator中用绑定变量执行带参数的Oracle查询

前置修正 #

先处理你代码里的笔误：你定义的变量名为last_snap_date，原拼接逻辑中误用为last_snap_date2，以下修改统一变量名。 #

核心修改逻辑 #

Oracle绑定变量使用:+变量名的格式作为占位符，不需要额外加单引号，直接作为TO_DATE的入参即可。调用get_records时传入parameters参数，类型为字典，key与SQL中的绑定变量名一一对应即可。 #

完整修改后代码 #

import cx_Oracle
from airflow.providers.oracle.hooks.oracle import OracleHook
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils import dates

def _process():
    print('*********** Oracle Conection ***************')   
    # 初始化Oracle连接
    hook = OracleHook(oracle_conn_id="Oracle")
    last_snap_date = '2021-03-23'
    # 使用绑定变量传递参数
    value = hook.get_records(
        sql="select distinct ven as name from Execution Metrics where date > TO_DATE(:last_snap_date, 'YYYY-MM-DD')",
        parameters={"last_snap_date": last_snap_date}
    )
    # 可按需添加后续处理逻辑
    print(value)

with DAG(
    dag_id='OracleQuery',
    start_date=dates.days_ago(2),
    schedule_interval='@monthly',
    catchup=False,
) as dag :
    task_a = PythonOperator(
        task_id="QueryIntegration",
        python_callable=_process
    )

补充说明 #

该写法底层直接调用cx_Oracle的参数绑定能力，完全避免SQL注入风险，也无需手动处理字符串转义、日期格式拼接的兼容性问题。

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