事务处理完整代码如下：
from contextlib import contextmanager
import mysql.connector
from mysql.connector import Error
class DatabaseConfig:
"""数据库配置类，集中管理连接参数"""
HOST = '192.168.116.70' # 数据库服务器地址
PORT = 3306 # 数据库端口号
DATABASE = 'python_mysql_demo' # 数据库名
USER = 'root' # 数据库用户名
PASSWORD = 'aaaaaa' # 数据库密码
CHARSET = 'utf8mb4' # 字符集，该字符集可支持emoji表情
class Database:
"""数据库操作封装"""
def __init__(self, config=DatabaseConfig):
self.config = config
self.connection = None
def connect(self):
"""建立连接"""
self.connection = mysql.connector.connect(
host=self.config.HOST,
port=self.config.PORT,
database=self.config.DATABASE,
user=self.config.USER,
password=self.config.PASSWORD,
charset=self.config.CHARSET,
autocommit=False
)
return self.connection
def close(self):
"""关闭连接"""
if self.connection and self.connection.is_connected():
self.connection.close()
def __enter__(self):
"""
进入上下文管理器，建立数据库连接
在 with 语句入口处被调用，建立连接并返回实例本身
"""
self.connect()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""
退出上下文管理器，关闭数据库连接
"""
self.close()
def get_students_by_id(db, student_id):
"""根据ID查询学生信息"""
sql = "SELECT * FROM students WHERE id = %s"
cursor = db.connection.cursor(dictionary=True)
cursor.execute(sql, (student_id,))
return cursor.fetchone()
def transfer_age(db, from_id, to_id, age):
try:
cursor = db.connection.cursor()
# 操作 1: 扣减年龄
cursor.execute(
"UPDATE students SET age = age - %s WHERE id = %s AND age >= %s",
(age, from_id, age)
)
if cursor.rowcount == 0:
raise ValueError("年龄不足或用户不存在")
# 操作 2: 增加年龄
cursor.execute(
"UPDATE students SET age = age + %s WHERE id = %s",
(age, to_id)
)
if cursor.rowcount == 0:
raise ValueError("目标用户不存在")
# 提交事务 —— 所有操作生效
db.connection.commit()
print("年龄扣减成功")
except Exception as e:
# 回滚事务 —— 撤销所有操作
db.connection.rollback()
print("年龄扣减失败，已回滚:", e)
# raise # 抛出异常，让上层调用者知道事务失败
if __name__ == "__main__":
# 通过年龄字段来模拟数据
with Database() as db:
try:
# 扣减前
student1 = get_students_by_id(db, 21)
print("扣减前学生1信息:\n", student1)
student2 = get_students_by_id(db, 22)
print("扣减前学生2信息:\n", student2)
# 扣减
transfer_age(db, from_id=21, to_id=22, age=10)
student1 = get_students_by_id(db, 21)
print("扣减后学生1信息:\n", student1)
student2 = get_students_by_id(db, 22)
print("扣减后学生2信息:\n", student2)
except Exception as e:
print("操作失败:", e)
川公网安备51010802032098