Python使用MySQL事务的三种主流方式

import pymysql

conn = pymysql.connect(
    host='localhost',
    user='root',
    password='your_password',
    database='test_db',
    charset='utf8mb4'
)

try:
    with conn.cursor() as cursor:
        # 开启事务（默认就是手动提交模式）
        sql1 = "UPDATE accounts SET balance = balance - 100 WHERE user_id = 1"
        sql2 = "UPDATE accounts SET balance = balance + 100 WHERE user_id = 2"
        
        cursor.execute(sql1)
        cursor.execute(sql2)
        
        # 全部成功，提交
        conn.commit()
        print("转账成功")
except Exception as e:
    # 任何一步出错，回滚
    conn.rollback()
    print(f"转账失败，已回滚：{e}")
finally:
    conn.close()

import mysql.connector

conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='your_password',
    database='test_db'
)

cursor = conn.cursor()

try:
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE user_id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE user_id = 2")
    conn.commit()
except mysql.connector.Error as err:
    conn.rollback()
    print(f"Error: {err}")
finally:
    cursor.close()
    conn.close()

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()

class Account(Base):
    __tablename__ = 'accounts'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    balance = Column(Integer)

engine = create_engine('mysql+pymysql://root:password@localhost/test_db')
Session = sessionmaker(bind=engine)

session = Session()

try:
    account1 = session.query(Account).filter_by(id=1).with_for_update().first()
    account2 = session.query(Account).filter_by(id=2).with_for_update().first()
    
    account1.balance -= 100
    account2.balance += 100
    
    session.commit()
    print("转账成功")
except Exception as e:
    session.rollback()
    print(f"转账失败：{e}")
finally:
    session.close()

# ❌ 错误示范
cursor.execute(sql1)
cursor.execute(sql2)  # 如果这里报错，sql1 已执行但没回滚
conn.commit()

conn.begin()  # 显式开启事务
cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")