Python实现高效更新MySQL数据的方法详解

from dbutils.pooled_db import PooledDB
import pymysql

# 创建连接池
pool = PooledDB(
    creator=pymysql,
    host='localhost',
    user='root',
    password='password',
    database='test_db',
    charset='utf8mb4',
    mincached=5,  # 初始连接数
    maxcached=20, # 最大空闲连接数
    maxconnections=50,  # 最大连接数
    blocking=True  # 连接不足时是否阻塞等待
)

# 从连接池获取连接
conn = pool.connection()
try:
    with conn.cursor() as cursor:
        cursor.execute("UPDATE users SET balance = balance - 100 WHERE id = 1")
        conn.commit()
finally:
    conn.close()  # 归还连接到池中

import pymysql

conn = pymysql.connect(...)
try:
    with conn.cursor() as cursor:
        # 准备批量数据（列表的列表）
        data = [
            (100, 'Alice'),
            (200, 'Bob'),
            (300, 'Charlie')
        ]
        # 使用executemany批量更新
        cursor.executemany(
            "UPDATE accounts SET balance = %s WHERE username = %s",
            data
        )
        conn.commit()
        print(f"Updated {cursor.rowcount} records")
finally:
    conn.close()

def batch_update_with_case(user_ids, new_balances):
    conn = pymysql.connect(...)
    try:
        with conn.cursor() as cursor:
            # 构建动态SQL
            sql = """
            UPDATE users 
            SET balance = CASE id
            """
            for user_id, balance in zip(user_ids, new_balances):
                sql += f"WHEN {user_id} THEN {balance} "
            sql += "END WHERE id IN (" + ",".join(map(str, user_ids)) + ")"
            
            cursor.execute(sql)
            conn.commit()
    finally:
        conn.close()

# 步骤1：创建临时表并导入数据
cursor.execute("""
    CREATE TEMPORARY TABLE temp_updates (
        id INT PRIMARY KEY,
        new_balance DECIMAL(10,2)
    )
""")
# 使用executemany插入临时数据（此处省略具体代码）

# 步骤2：执行JOIN更新
cursor.execute("""
    UPDATE users u
    JOIN temp_updates t ON u.id = t.id
    SET u.balance = t.new_balance
""")

def transfer_money(from_id, to_id, amount):
    conn = pymysql.connect(autocommit=False)  # 显式关闭自动提交
    try:
        with conn.cursor() as cursor:
            # 开始事务（MySQL中可省略，DML语句会自动开启）
            cursor.execute("START TRANSACTION")
            
            # 执行扣款
            cursor.execute(
                "UPDATE accounts SET balance = balance - %s WHERE id = %s AND balance >= %s",
                (amount, from_id, amount)
            )
            if cursor.rowcount == 0:
                raise ValueError("Insufficient balance or user not found")
            
            # 执行加款
            cursor.execute(
                "UPDATE accounts SET balance = balance + %s WHERE id = %s",
                (amount, to_id)
            )
            
            conn.commit()  # 提交事务
            print("Transaction completed successfully")
    except Exception as e:
        conn.rollback()  # 回滚事务
        print(f"Transaction failed: {e}")
    finally:
        conn.close()

# 错误示例：对索引字段使用函数导致索引失效
cursor.execute("""
    SELECT * FROM users 
    WHERE DATE(create_time) = '2026-01-01'  # 索引失效
""")

# 正确写法：使用范围查询
cursor.execute("""
    SELECT * FROM users 
    WHERE create_time BETWEEN '2026-01-01 00:00:00' AND '2026-01-01 23:59:59'
""")

from celery import Celery
import pymysql

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def async_update_user_score(user_id, new_score):
    conn = pymysql.connect(...)
    try:
        with conn.cursor() as cursor:
            cursor.execute(
                "UPDATE users SET score = %s WHERE id = %s",
                (new_score, user_id)
            )
        conn.commit()
    finally:
        conn.close()

# 调用异步任务
async_update_user_score.delay(123, 95)

# 原始代码（存在问题）
for item_id in item_ids:
    cursor.execute(
        "UPDATE inventory SET stock = stock - 1 WHERE id = %s AND stock > 0",
        (item_id,)
    )
    conn.commit()  # 每次更新都提交，性能极差

def update_inventory_batch(item_updates):
    """
    item_updates: List[Tuple[item_id, quantity]]
    """
    conn = pymysql.connect(autocommit=False)
    try:
        with conn.cursor() as cursor:
            # 批量更新主逻辑
            for item_id, quantity in item_updates:
                cursor.execute("""
                    UPDATE inventory 
                    SET stock = stock - %s 
                    WHERE id = %s AND stock >= %s
                """, (quantity, item_id, quantity))
                if cursor.rowcount == 0:
                    raise ValueError(f"Inventory shortage for item {item_id}")
            
            # 提交事务（所有更新成功或全部回滚）
            conn.commit()
            
            # 可选：记录更新日志到异步队列
            # async_log_inventory_changes(item_updates)
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        conn.close()