PostgreSQL数据库莫名回滚，SQLAlchemy连接处理是否合规？

核心问题：事务提交遗漏 + 连接管理不规范 #

你的代码存在两个关键缺陷，直接导致了数据丢失的风险：

1. 事务提交逻辑不完整 #

SQLAlchemy 1.4中，通过engine.connect()获取的连接默认处于自动事务模式：所有写操作（INSERT/UPDATE/DELETE）都会启动一个事务，只有显式调用commit()，修改才会持久化到数据库。

看你的execute_wrapper方法：

• 仅当vars_obj是列表时才调用self.connection.commit()
• 单个vars_obj或无vars_obj的写操作（比如单条INSERT、UPDATE语句）完全没有提交步骤
• 连接关闭时，未提交的事务会被自动回滚，所有未提交的修改都会丢失

如果你的程序长期持有同一个连接（比如调用start_connection后一直未执行end_connection），大量未提交的事务会堆积，一旦连接因异常断开、程序重启或连接池回收，这些数据就会全部回滚。

2. 异常时事务与连接处理不当 #

• 异常发生后直接调用restart_connection关闭旧连接，但未对旧连接中的事务做回滚或提交处理。处于异常状态的事务被直接关闭时，PostgreSQL会自动回滚该事务，导致之前未提交的修改丢失。
• 长期持有单个连接的模式违背了SQLAlchemy连接池的设计初衷，连接池应该按需获取、用完归还，长期持有容易引发事务泄漏、连接失效等问题。

修复方案 #

方案1：补全事务提交与异常回滚逻辑 #

修改execute_wrapper，统一处理写操作的提交，并在异常时回滚事务：

def execute_wrapper(self, qstring, vars_obj=None, fetchall=False, fetchone=False, is_write=False):
    if self.connection is None:
        self.start_connection()
    while True:
        try:
            if vars_obj is None:
                res = self.connection.execute(qstring)
            else:
                if isinstance(vars_obj, list):
                    res = self.connection.execute(qstring, vars_obj)
                else:
                    res = self.connection.execute(qstring, vars_obj)
            # 写操作强制提交
            if is_write:
                self.connection.commit()
            break
        except (
            sqlalchemy.exc.OperationalError,
            psycopg2.OperationalError,
            sqlalchemy.exc.DatabaseError,
        ) as e:
            database_logger.error(f"Database offline, retrying in 60s: {e}")
            # 异常时先回滚事务，避免脏数据残留
            if self.connection:
                self.connection.rollback()
            time.sleep(60)
            self.restart_connection()
        except Exception as e:
            database_logger.critical(f"Unknown DB Error: {e}\nQuery: {qstring}")
            # 异常时回滚事务
            if self.connection:
                self.connection.rollback()
            time.sleep(60)
            self.restart_connection()

    if fetchall is True:
        res = res.fetchall()
    elif fetchone is True:
        res = res.fetchone()
    return res

调用时，针对写操作传入is_write=True：

db.execute_wrapper("INSERT INTO table (...) VALUES (...)", vars_obj=params, is_write=True)

方案2：使用SQLAlchemy原生上下文管理器（推荐） #

放弃手动管理连接和事务，改用上下文管理器自动处理连接获取、事务提交/回滚：

class Database:
    def __init__(self, database_params=None):
        if database_params is None:
            database_params = DATABASE_PARAMS
        self.database_params = database_params
        self.engine = db.create_engine(
            f"{self.database_params['engine']}://{self.database_params['user']}:{quote(self.database_params['password'])}@{self.database_params['host']}:{self.database_params['port']}/{self.database_params['database_name']}"
        )

    def execute_wrapper(self, qstring, vars_obj=None, fetchall=False, fetchone=False, is_write=False):
        while True:
            try:
                with self.engine.connect() as conn:
                    if is_write:
                        # 写操作开启事务，上下文结束自动提交，异常自动回滚
                        with conn.begin():
                            if vars_obj is None:
                                res = conn.execute(qstring)
                            else:
                                if isinstance(vars_obj, list):
                                    res = conn.execute(qstring, vars_obj)
                                else:
                                    res = conn.execute(qstring, vars_obj)
                    else:
                        # 读操作无需事务，直接执行
                        if vars_obj is None:
                            res = conn.execute(qstring)
                        else:
                            if isinstance(vars_obj, list):
                                res = conn.execute(qstring, vars_obj)
                            else:
                                res = conn.execute(qstring, vars_obj)

                    if fetchall:
                        res = res.fetchall()
                    elif fetchone:
                        res = res.fetchone()
                return res
            except (
                sqlalchemy.exc.OperationalError,
                psycopg2.OperationalError,
                sqlalchemy.exc.DatabaseError,
            ) as e:
                database_logger.error(f"Database offline, retrying in 60s: {e}")
                time.sleep(60)
            except Exception as e:
                database_logger.critical(f"Unknown DB Error: {e}\nQuery: {qstring}")
                time.sleep(60)

这种方式完全由SQLAlchemy管理连接和事务：

• 上下文结束自动归还连接到池
• 写事务中发生异常自动回滚，无异常自动提交
• 彻底避免手动管理连接的风险

额外建议 #

1. 立即恢复备份：用前一日的备份恢复数据，减少损失
2. 开启PostgreSQL WAL日志：确保数据库层面有完整的事务日志，便于后续排查
3. 添加事务日志：记录每次事务的提交/回滚情况，方便定位问题

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