如何在Airflow PostgresOperator中设置ON_ERROR_STOP=1？

嘿，这个问题我之前也碰到过，给你分享几种不用修改query.sql就能实现需求的方案：

方案1：用BashOperator调用psql命令行（完全匹配你bash里的用法） #

这种方法直接复用你熟悉的psql -v ON_ERROR_STOP=1逻辑，不需要改动原SQL文件，还能利用Airflow的Postgres连接配置，不用硬编码数据库信息：

from airflow.hooks.postgres_hook import PostgresHook
from airflow.operators.bash import BashOperator

def build_psql_cmd():
    # 从Airflow的Postgres连接中自动获取数据库地址、账号等信息
    pg_hook = PostgresHook(postgres_conn_id='my_server')
    conn_uri = pg_hook.get_uri()
    # 拼接psql命令，指定ON_ERROR_STOP参数并执行目标SQL文件
    # 注意：这里的SQL文件路径要根据你实际的存放位置调整，比如放在dags/sql下就用$AIRFLOW_HOME/dags/sql/query.sql
    return f'psql "{conn_uri}" -v ON_ERROR_STOP=1 -f $AIRFLOW_HOME/dags/sql/query.sql'

my_operator = BashOperator(
    task_id='my_operator',
    dag=dag,
    bash_command=build_psql_cmd()
)

方案2：用PostgresOperator + Jinja2模板（无需改原SQL） #

如果你更倾向于用PostgresOperator，其实它默认的事务行为已经和ON_ERROR_STOP=1的核心效果一致——只要SQL执行中出现错误，整个事务会回滚，后续语句也不会执行。你可以通过Jinja2模板把原SQL文件直接嵌入：

my_operator = PostgresOperator(
    task_id='my_operator',
    dag=dag,
    postgres_conn_id='my_server',
    sql="""
    {% include 'sql/query.sql' %}
    """,
    # 默认就是False，开启事务模式，出错自动回滚
    autocommit=False
)

这里要注意：PostgresOperator底层用的是psycopg2库，它不识别psql的\set这类客户端元命令，所以直接写\set ON_ERROR_STOP on是无效的，但它的事务机制已经能满足“出错即停止”的需求。

额外说明 #

为啥不能直接给PostgresOperator传ON_ERROR_STOP参数？因为ON_ERROR_STOP是psql命令行工具的专属参数，不是PostgreSQL服务器的配置项，也不是psycopg2库支持的连接参数，所以没法通过PostgresOperator的参数直接传递。如果一定要用psql的元命令，那只能选方案1的BashOperator方式。

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