如何确保Airflow DAG中Python代码抛异常时finally块仍执行

正常情况下finally块无论是否抛出异常都该执行，但在Airflow里没执行，大概率是Airflow在任务失败时直接强制终止了进程（比如发送SIGKILL信号），导致finally代码没来得及跑。另外，你在except里手动抛出ValueError的方式，可能触发了Airflow的快速终止逻辑。

给你几个靠谱的解决思路：

1. 用Airflow原生异常替代手动抛ValueError #

别自己抛ValueError，改用Airflow的AirflowException——它是Airflow识别任务失败的标准异常，不会触发强制进程终止，能保证finally正常执行：

from airflow.exceptions import AirflowException

try:
    dump_tables(mssql_engine=mssql_engine, tables_list=tables_for_db_connection)
except Exception as e:
    logger.error(f"Error during database operation: {str(e)}")
    raise AirflowException("Database operation failed, triggering DAG failure")
finally:
    mssql_engine.dispose()

2. 使用上下文管理器管理数据库连接 #

用contextlib.closing或者SQLAlchemy引擎自带的上下文管理，即使进程出问题，也能更可靠地释放资源：

from contextlib import closing

with closing(mssql_engine) as engine:
    try:
        dump_tables(mssql_engine=engine, tables_list=tables_for_db_connection)
    except Exception as e:
        logger.error(f"Error during database operation: {str(e)}")
        raise AirflowException("任务失败")
# 这里不需要finally，closing会自动调用dispose

3. 双重保障资源释放 #

如果一定要手动抛异常，可以在except里先手动执行资源释放，再抛异常，同时保留finally做双重保障：

try:
    dump_tables(mssql_engine=mssql_engine, tables_list=tables_for_db_connection)
except Exception as e:
    logger.error(f"Error during database operation: {str(e)}")
    # 先手动释放资源
    mssql_engine.dispose()
    raise AirflowException("触发DAG失败")
finally:
    # 再加一层保障，防止上面的释放没执行
    try:
        mssql_engine.dispose()
    except:
        pass

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