import pymysql

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

# 轉(zhuǎn)賬：扣款
cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
# 假設(shè)這里程序崩潰 / 網(wǎng)絡(luò)超時(shí) / 拋出異常...
# 加款永遠(yuǎn)沒(méi)執(zhí)行
cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")

conn.commit()  # 第一條已經(jīng)生效，無(wú)法回滾

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()
    # 忘記關(guān)閉連接，直接返回
    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 自動(dòng)關(guān)閉，連接歸還池中

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)

# 定義事件：每次更新自動(dòng)維護(hù) updated_at
@event.listens_for(User, 'before_update')
def set_updated_at(mapper, connection, target):
    target.updated_at = datetime.now()

# 業(yè)務(wù)代碼：批量更新用戶(hù)狀態(tài)
session = Session()
stmt = update(User).where(User.status == 'pending').values(status='processed')
session.execute(stmt)  # ← 監(jiān)聽(tīng)器根本沒(méi)觸發(fā)
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()