Python操作MySQL数据库时常见的三个陷阱及解决方法

import pymysql

conn = pymysql.connect(host='localhost', user='root', password='pwd', database='test')
cursor = conn.cursor()

# 转账：扣款
cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
# 假设这里程序崩溃 / 网络超时 / 抛出异常...
# 加款永远没执行
cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")

conn.commit()  # 第一条已经生效，无法回滚

try:
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except Exception as e:
    conn.rollback()  # ← 必须显式回滚
    raise  # 或记录日志后重新抛出
finally:
    cursor.close()
    conn.close()

import mysql.connector
from mysql.connector import pooling

dbconfig = {
    "host": "localhost", "user": "root", "password": "pwd", "database": "test"
}
connection_pool = pooling.MySQLConnectionPool(pool_name="mypool", pool_size=5, **dbconfig)

def get_user(user_id):
    conn = connection_pool.get_connection()  # 从池中取连接
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    result = cursor.fetchall()
    # 忘记关闭连接，直接返回
    return result

def get_user(user_id):
    conn = connection_pool.get_connection()
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
        return cursor.fetchall()
    finally:
        cursor.close()
        conn.close()  # ← 必须归还连接到池中

from contextlib import closing

def get_user(user_id):
    with closing(connection_pool.get_connection()) as conn:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
            return cursor.fetchall()
    # 退出 with 自动关闭，连接归还池中

from sqlalchemy import event, update
from sqlalchemy.orm import Session

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    updated_at = Column(DateTime)

# 定义事件：每次更新自动维护 updated_at
@event.listens_for(User, 'before_update')
def set_updated_at(mapper, connection, target):
    target.updated_at = datetime.now()

# 业务代码：批量更新用户状态
session = Session()
stmt = update(User).where(User.status == 'pending').values(status='processed')
session.execute(stmt)  # ← 监听器根本没触发
session.commit()

from sqlalchemy import func

stmt = (
    update(User)
    .where(User.status == 'pending')
    .values(status='processed', updated_at=func.now())
)
session.execute(stmt)
session.commit()