使用SQLAlchemy调用PostgreSQL的PL/pgSQL函数插入数据无入库问题

嘿，我懂你这种新手踩坑的感觉——明明代码没报错，还拿到了Job ID，结果数据库里啥都没有，这确实挺挠头的。核心问题其实很简单：你漏掉了事务提交的步骤。

为什么旧代码能正常工作？ #

你的旧实现里明确调用了db.commit()，这会把之前执行的SQL操作（调用add_job函数）提交到数据库，让变更永久生效。而PostgreSQL的函数里的INSERT操作是在当前事务上下文里执行的，只有提交事务后，这些变更才会被持久化到数据库中。

新代码的问题出在哪？ #

你用engine.execute(query)的时候，SQLAlchemy默认会为这个操作开启一个事务，但它不会自动提交事务。所以虽然你从函数里拿到了返回的Job ID（这个结果是事务内可见的），但事务没有提交，当连接关闭后，这个事务就会被回滚，数据库里自然看不到对应的条目。

修复方案 #

方案1：手动管理连接并提交事务 #

如果你想继续用engine直接执行，可以显式获取连接，执行后提交事务：

def insert_job(user_id: int, param_id: str, proc_name: str, run_date: datetime) -> Job:
    query = select([func.add_job(user_id, param_id, proc_name, run_date)])
    try:
        # 显式获取连接并管理
        with engine.connect() as conn:
            result = conn.execute(query).fetchone()
            conn.commit()  # 关键：提交事务
    except RaiseException as e:
        raise
    return result

方案2：使用SQLAlchemy Session（更推荐） #

如果你的项目是基于SQLAlchemy ORM的，更推荐使用Session来管理事务，它会帮你处理连接和事务的生命周期：

from sqlalchemy.orm import sessionmaker

# 先创建Session工厂（通常在项目初始化时做一次）
Session = sessionmaker(bind=engine)

def insert_job(user_id: int, param_id: str, proc_name: str, run_date: datetime) -> Job:
    session = Session()
    try:
        result = session.execute(select([func.add_job(user_id, param_id, proc_name, run_date)])).fetchone()
        session.commit()  # 提交事务
        return result
    except RaiseException as e:
        session.rollback()  # 出错时回滚事务，避免脏数据
        raise
    finally:
        session.close()  # 确保Session关闭

额外提示 #

PostgreSQL的函数内的INSERT操作是依赖外层事务的，哪怕函数本身执行成功，只要外层事务没提交，所有变更都只是临时的。这也是为什么你能拿到返回的Job ID（事务内可见），但数据库里查不到的原因——事务还没落地呢！

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