psycopg2执行插入报错InFailedSqlTransaction，求解决方案

问题场景 #

使用psycopg2向PostgreSQL数据库插入数据时触发错误：

psycopg2.errors.InFailedSqlTransaction: current transaction is aborted, commands ignored until end of transaction block

用户代码如下：

def insert_transaction():
global username
now = datetime.now()
date_checkout = datetime.today().strftime('%d-%m-%Y')
time_checkout = now.strftime("%H:%M:%S")

username = "Peter1"

connection_string = "host='localhost' dbname='Los Pollos Hermanos' user='postgres' password='******'"
conn = psycopg2.connect(connection_string)
cursor = conn.cursor()
try:
    query_check_1 = """(SELECT employeeid FROM employee WHERE username = %s);"""
    cursor.execute(query_check_1, (username,))
    employeeid = cursor.fetchone()[0]
    conn.commit()
except:
    print("Employee error")

try:
    query_check_2 = """SELECT MAX(transactionnumber) FROM Transaction"""
    cursor.execute(query_check_2)
    transactionnumber = cursor.fetchone()[0] + 1
    conn.commit()
except:
    transactionnumber = 1

""""---------INSERT INTO TRANSACTION------------"""


query_insert_transaction = """INSERT INTO transactie (transactionnumber, date, time, employeeemployeeid)
                    VALUES (%s, %s, %s, %s);"""
data = (transactionnumber, date_checkout, time_checkout, employeeid)
cursor.execute(query_insert_transaction, data)
conn.commit()
conn.close()

错误原因 #

1. 第一个try块执行失败（如查询不到对应username的employeeid）时，仅打印错误但未回滚事务，PostgreSQL事务会进入失败状态，后续所有数据库操作都会被拒绝。
2. 若employeeid未被成功赋值，后续插入操作会引发新错误，进一步导致事务异常。
3. 频繁提交单个查询的事务，破坏了事务原子性，也增加了出错概率。

解决方案 #

修复后的代码 #

def insert_transaction():
    from datetime import datetime
    import psycopg2

    username = "Peter1"
    now = datetime.now()
    date_checkout = datetime.today().strftime('%d-%m-%Y')
    time_checkout = now.strftime("%H:%M:%S")

    connection_string = "host='localhost' dbname='Los Pollos Hermanos' user='postgres' password='******'"
    conn = psycopg2.connect(connection_string)
    cursor = conn.cursor()
    
    try:
        # 查询员工ID
        query_check_1 = """SELECT employeeid FROM employee WHERE username = %s;"""
        cursor.execute(query_check_1, (username,))
        employeeid = cursor.fetchone()[0]
        if not employeeid:
            raise ValueError(f"未找到用户 {username} 对应的员工ID")
        
        # 获取最大事务编号，处理表为空的情况
        query_check_2 = """SELECT MAX(transactionnumber) FROM "Transaction";"""
        cursor.execute(query_check_2)
        max_trans_num = cursor.fetchone()[0]
        transactionnumber = max_trans_num + 1 if max_trans_num is not None else 1
        
        # 插入事务数据
        query_insert_transaction = """INSERT INTO transactie (transactionnumber, date, time, employeeemployeeid)
                                     VALUES (%s, %s, %s, %s);"""
        data = (transactionnumber, date_checkout, time_checkout, employeeid)
        cursor.execute(query_insert_transaction, data)
        
        # 所有操作成功后统一提交事务
        conn.commit()
        print("事务数据插入成功")
    except psycopg2.Error as db_err:
        # 数据库操作错误时回滚事务
        conn.rollback()
        print(f"数据库操作失败: {str(db_err)}")
        raise  # 可根据需求决定是否抛出异常终止流程
    except Exception as general_err:
        # 其他意外错误时回滚事务
        conn.rollback()
        print(f"操作失败: {str(general_err)}")
        raise
    finally:
        # 确保资源被释放
        cursor.close()
        conn.close()

关键修复点 #

• 将所有数据库操作放在同一个try块中，保证事务原子性，要么全部成功提交，要么全部失败回滚。
• 异常块中添加conn.rollback()，重置失败的事务状态，避免后续操作被阻塞。
• 处理MAX(transactionnumber)返回None的情况（表为空时），防止类型错误。
• 增加对employeeid的有效性检查，提前发现数据问题。
• 使用finally块确保数据库连接和游标被关闭，避免资源泄漏。

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