影刀RPA 数据库事务处理：原子操作与回滚

影刀RPA 数据库事务处理原子操作与回滚署名林焱什么情况用RPA流程要给100个客户转账——转到第57个的时候断电了。前56个已转后44个没转。怎么办手动对账事务就是解决这个问题的要么全部成功要么全部回滚。中间任何一步失败已经执行的全部撤销数据库回到操作前的状态。典型场景批量转账/扣款必须全部成功或全部失败订单处理减库存生成订单扣余额三步必须原子完成数据迁移旧表数据搬到新表中间不能中断多表关联操作主表和子表必须同时写入怎么做拼多多店群自动化上架方案1. 基础事务importsqlite3 db_pathrC:\RPAData\business.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 建表cursor.execute(CREATE TABLE IF NOT EXISTS accounts (id INTEGER PRIMARY KEY, name TEXT, balance REAL))cursor.execute(CREATE TABLE IF NOT EXISTS transfer_log (id INTEGER PRIMARY KEY, from_id INTEGER, to_id INTEGER, amount REAL, transfer_time TEXT))cursor.executemany(INSERT OR IGNORE INTO accounts VALUES (?,?,?),[(1,张三,10000),(2,李四,5000),(3,王五,3000)])conn.commit()# 转账操作事务保证原子性try:# 开启事务sqlite3默认已开启cursor.execute(UPDATE accounts SET balance balance - 1000 WHERE id 1)# 张三扣1000cursor.execute(UPDATE accounts SET balance balance 1000 WHERE id 2)# 李四加1000cursor.execute(INSERT INTO transfer_log VALUES (NULL, 1, 2, 1000, datetime(now)))conn.commit()# 提交事务三步同时生效print(转账成功张三 → 李四 ¥1000)exceptExceptionase:conn.rollback()# 回滚事务三步全部撤销print(f转账失败已回滚{e})finally:conn.close()2. 批量事务处理importsqlite3fromdatetimeimportdatetime db_pathrC:\RPAData\business.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 批量转账列表transfers[(1,2,500),# 张三→李四 500(1,3,300),# 张三→王五 300(2,3,200),# 李四→王五 200(3,1,100),# 王五→张三 100]try:forfrom_id,to_id,amountintransfers:# 检查余额是否充足cursor.execute(SELECT balance FROM accounts WHERE id ?,(from_id,))balancecursor.fetchone()[0]ifbalanceamount:raiseValueError(f账户{from_id}余额不足{balance}{amount})# 执行转账cursor.execute(UPDATE accounts SET balance balance - ? WHERE id ?,(amount,from_id))cursor.execute(UPDATE accounts SET balance balance ? WHERE id ?,(amount,to_id))cursor.execute(INSERT INTO transfer_log VALUES (NULL, ?, ?, ?, ?),(from_id,to_id,amount,datetime.now().strftime(%Y-%m-%d %H:%M:%S)))conn.commit()# 所有转账一次性提交print(f批量转账完成{len(transfers)}笔)exceptExceptionase:conn.rollback()print(f批量转账失败已全部回滚{e})finally:conn.close()3. 事务封装类importsqlite3fromcontextlibimportcontextmanager db_pathrC:\RPAData\business.dbcontextmanagerdeftransaction(db_path):事务上下文管理器自动提交/回滚connsqlite3.connect(db_path)try:yieldconn conn.commit()# 无异常则提交exceptExceptionase:conn.rollback()# 有异常则回滚raise# 重新抛出异常finally:conn.close()# 使用示例处理订单减库存生成订单扣余额withtransaction(db_path)asconn:cursorconn.cursor()order_items[(SKU-001,2,99.9),(SKU-002,1,199.0),]customer_id1total_amountsum(qty*pricefor_,qty,priceinorder_items)# 1. 检查并扣减库存forsku,qty,priceinorder_items:cursor.execute(SELECT stock FROM inventory WHERE sku ?,(sku,))resultcursor.fetchone()ifnotresult:raiseValueError(f商品{sku}不存在)stockresult[0]ifstockqty:raiseValueError(f商品{sku}库存不足{stock}{qty})cursor.execute(UPDATE inventory SET stock stock - ? WHERE sku ?,(qty,sku))# 2. 扣减客户余额cursor.execute(SELECT balance FROM accounts WHERE id ?,(customer_id,))balancecursor.fetchone()[0]ifbalancetotal_amount:raiseValueError(f余额不足{balance}{total_amount})cursor.execute(UPDATE accounts SET balance balance - ? WHERE id ?,(total_amount,customer_id))# 3. 生成订单cursor.execute(INSERT INTO orders VALUES (NULL, ?, ?, ?, datetime(now)),(customer_id,str(order_items),total_amount))print(f订单创建成功总额¥{total_amount})4. 保存点部分回滚importsqlite3 db_pathrC:\RPAData\business.dbconnsqlite3.connect(db_path)cursorconn.cursor()try:# 主事务cursor.execute(UPDATE accounts SET balance balance - 500 WHERE id 1)# 设置保存点cursor.execute(SAVEPOINT sp1)# 这段操作可能失败try:cursor.execute(UPDATE accounts SET balance balance 500 WHERE id 2)cursor.execute(INSERT INTO transfer_log VALUES (NULL, 1, 2, 500, datetime(now)))cursor.execute(RELEASE SAVEPOINT sp1)# 释放保存点确认这段操作exceptExceptionase:cursor.execute(ROLLBACK TO SAVEPOINT sp1)# 只回滚到保存点print(f第二步失败已回滚到保存点{e})# 主事务的第一步不受影响conn.commit()# 提交主事务exceptExceptionase:conn.rollback()print(f主事务失败全部回滚{e})finally:conn.close()有什么坑坑1sqlite3的自动提交行为Python的sqlite3模块默认不是自动提交模式——每条DML语句INSERT/UPDATE/DELETE执行后不会立即生效需要手动commit。但DDL语句CREATE/ALTER/DROP会自动触发隐式提交。# 这段代码看起来没问题但如果不commit数据不会持久化cursor.execute(INSERT INTO orders VALUES (1, test))# conn.close() ← 直接关闭数据丢失# 正确关闭前commitcursor.execute(INSERT INTO orders VALUES (1, test))conn.commit()conn.close()更安全的做法用isolation_levelNone开启自动提交模式或者在finally中确保commit/rollback。坑2事务中连接断开导致隐式回滚事务执行到一半网络断了MySQL或进程被杀SQLite数据库会自动回滚整个事务。但你的RPA流程可能不知道继续往下跑导致数据不一致。TEMU店群如何管理运营# 解决方案关键事务加幂等性检查deftransfer_with_check(conn,from_id,to_id,amount,transfer_key):带幂等性的转账cursorconn.cursor()# 检查是否已执行过通过唯一键cursor.execute(SELECT * FROM transfer_log WHERE transfer_key ?,(transfer_key,))ifcursor.fetchone():print(f转账{transfer_key}已执行过跳过)return# 执行转账cursor.execute(UPDATE accounts SET balance balance - ? WHERE id ?,(amount,from_id))cursor.execute(UPDATE accounts SET balance balance ? WHERE id ?,(amount,to_id))cursor.execute(INSERT INTO transfer_log VALUES (NULL, ?, ?, ?, ?, ?, datetime(now)),(from_id,to_id,amount,transfer_key))坑3事务时间过长导致锁表一个事务里做了太多操作执行了5分钟这期间其他连接无法写入SQLite或等待锁超时MySQL。# 错误在一个大事务里处理所有数据cursor.execute(BEGIN)foriteminhuge_list:# 10万条cursor.execute(INSERT INTO big_table VALUES (?),(item,))cursor.execute(COMMIT)# 10分钟后才提交期间数据库被锁# 正确分批提交BATCH_SIZE1000foriinrange(0,len(huge_list),BATCH_SIZE):batchhuge_list[i:iBATCH_SIZE]cursor.execute(BEGIN)cursor.executemany(INSERT INTO big_table VALUES (?),batch)cursor.execute(COMMIT)坑4嵌套事务不是真嵌套SQLite不支持真正的嵌套事务。在一个事务中再开一个事务第二个BEGIN会被忽略。MySQL的嵌套事务也是用保存点模拟的。# 这段代码看起来是嵌套事务实际不是cursor.execute(BEGIN)cursor.execute(INSERT...)# 在外层事务中cursor.execute(BEGIN)# 被忽略没有新事务cursor.execute(INSERT...)# 还是在外层事务中cursor.execute(COMMIT)# 提交了所有操作cursor.execute(INSERT...)# 自动提交模式如果驱动支持要用保存点SAVEPOINT实现部分回滚的效果如上面第4节所示。坑5异常处理中忘记rollback最常见的bugcatch了异常但没rollback连接还挂着事务后续操作全部阻塞。# 错误try:cursor.execute(UPDATE accounts SET balance balance - 1000 WHERE id 1)resultsome_risky_operation()# 这里抛异常cursor.execute(UPDATE accounts SET balance balance 1000 WHERE id 2)conn.commit()exceptExceptionase:print(f失败{e})# 忘了rollback事务还开着第一步的UPDATE已执行但未提交# 下次连接看到的数据还是旧的但锁还占着# 正确try:cursor.execute(UPDATE accounts SET balance balance - 1000 WHERE id 1)resultsome_risky_operation()cursor.execute(UPDATE accounts SET balance balance 1000 WHERE id 2)conn.commit()exceptExceptionase:conn.rollback()# 必须rollbackprint(f失败已回滚{e})finally:conn.close()# 确保连接关闭