本文 AI 產出，尚未審核

Python 資料庫操作：交易（Transaction）

簡介

在日常開發中，資料庫交易是確保資料一致性與完整性的關鍵機制。無論是線上購物的訂單處理、銀行帳戶的餘額更新，或是多表格的批次匯入，都必須保證「要麼全部成功，要麼全部失敗」的原子性（Atomicity）。如果交易處理不當，最常見的問題包括資料遺失、重複寫入、甚至資料庫死鎖，這些都會直接影響系統的可靠度與使用者體驗。

Python 提供了多種與資料庫互動的套件（如 sqlite3、psycopg2、mysql‑connector-python、SQLAlchemy 等），而它們的交易控制方式大致相同：開始交易 → 執行 SQL → 提交（commit）或回滾（rollback）。本篇文章將從概念說明、實作範例、常見陷阱與最佳實踐，帶你一步步掌握 Python 中的交易操作，讓你的程式碼在面對複雜資料變更時依然安全、可控。

核心概念

1. 交易的四大特性（ACID）

  • Atomic（原子性）：交易內的所有操作要麼全部成功，要麼全部失敗。
  • Consistency（一致性）：交易結束後，資料庫必須從一個合法狀態轉換到另一個合法狀態。
  • Isolation（隔離性）：同時執行的交易不應互相干擾，常見的隔離層級有 READ UNCOMMITTED、READ COMMITTED、REPEATABLE READ、SERIALIZABLE。
  • Durability（永續性）：一旦交易提交，資料變更必須永久寫入磁碟，即使系統崩潰也不會遺失。

Tip：在大多數應用中，READ COMMITTED 已足以滿足需求；若需要防止幻讀（phantom read），可考慮 SERIALIZABLE。

2. 交易的基本流程

步驟 說明
開始交易 大多數 Python DB API 會在取得連線 (connection) 後自動開啟交易，或可手動呼叫 connection.begin()（視套件而定）。
執行 SQL 使用 cursor.execute() 執行增、刪、改等語句。
提交 (commit) 若所有語句皆成功，呼叫 connection.commit()，將變更寫入磁碟。
回滾 (rollback) 若發生例外或邏輯錯誤，呼叫 connection.rollback()，撤銷本次交易的所有變更。

3. 透過 Context Manager 簡化交易控制

Python 的 with 陳述式可以自動處理 commit 與 rollback，讓程式碼更乾淨且不易遺漏：

with connection:
    with connection.cursor() as cursor:
        cursor.execute("INSERT INTO orders ...")
        cursor.execute("UPDATE inventory ...")
# 離開 with 區塊時，若無例外則自動 commit，否則自動 rollback

注意：不同資料庫套件的 with 行為可能略有差異，請參考官方文件。

程式碼範例

以下示範三個常見的 Python DB 套件，從最簡單的 sqlite3 到較大型的 PostgreSQL，再到 ORM 的 SQLAlchemy。每個範例都包含 註解說明，方便初學者快速理解。

範例 1：SQLite 內建交易（sqlite3）

import sqlite3

# 建立連線（會自動開啟交易）
conn = sqlite3.connect('example.db')
try:
    cursor = conn.cursor()
    # 1. 新增訂單
    cursor.execute(
        "INSERT INTO orders (order_id, amount) VALUES (?, ?)",
        (101, 2500)
    )
    # 2. 更新庫存
    cursor.execute(
        "UPDATE products SET stock = stock - 1 WHERE product_id = ?",
        (5,)
    )
    # 若以上兩個語句都成功，提交交易
    conn.commit()
    print("交易成功，資料已寫入。")
except Exception as e:
    # 任一語句失敗，都回滾所有變更
    conn.rollback()
    print("交易失敗，已回滾。原因:", e)
finally:
    conn.close()

重點：sqlite3 的 commit() 必須手動呼叫；若忘記 rollback()，錯誤後的變更仍會暫留在記憶體中，最終關閉連線時會自動回滾。

範例 2：PostgreSQL 交易與隔離層級（psycopg2）

import psycopg2
from psycopg2 import sql, extensions

# 連線到 PostgreSQL
conn = psycopg2.connect(
    dbname="shop",
    user="postgres",
    password="secret",
    host="127.0.0.1",
    port=5432
)

# 設定交易的隔離層級為 SERIALIZABLE（最嚴格）
conn.set_isolation_level(extensions.ISOLATION_LEVEL_SERIALIZABLE)

try:
    with conn:
        with conn.cursor() as cur:
            # 1. 插入新用戶
            cur.execute(
                "INSERT INTO customers (name, email) VALUES (%s, %s)",
                ("Alice", "alice@example.com")
            )
            # 2. 同時扣除優惠券庫存（可能會觸發唯一鍵衝突）
            cur.execute(
                "UPDATE coupons SET remaining = remaining - 1 WHERE code = %s AND remaining > 0",
                ("WELCOME2025",)
            )
    # 離開 with 區塊自動 commit，若有例外則自動 rollback
    print("交易完成")
except psycopg2.Error as err:
    print("交易失敗，已回滾。錯誤資訊:", err)
finally:
    conn.close()

說明：psycopg2 內建支援 with 讓交易自動提交或回滾；此外，set_isolation_level 可調整 隔離層級，在高併發系統中尤為重要。

範例 3：MySQL 交易與自訂錯誤處理（mysql-connector-python）

import mysql.connector
from mysql.connector import errorcode

config = {
    'user': 'root',
    'password': 'pwd123',
    'host': 'localhost',
    'database': 'ecommerce',
    'autocommit': False   # 必須關閉自動提交，才能手動控制交易
}

cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()

try:
    # 1. 建立付款紀錄
    cursor.execute(
        "INSERT INTO payments (order_id, amount) VALUES (%s, %s)",
        (2001, 3999)
    )
    # 2. 更新訂單狀態
    cursor.execute(
        "UPDATE orders SET status = %s WHERE order_id = %s",
        ("PAID", 2001)
    )
    # 手動提交
    cnx.commit()
    print("付款交易成功")
except mysql.connector.Error as err:
    # 若是唯一鍵衝突，特別處理
    if err.errno == errorcode.ER_DUP_ENTRY:
        print("付款紀錄已存在，略過插入")
        cnx.rollback()
    else:
        cnx.rollback()
        print("交易失敗，已回滾。錯誤:", err)
finally:
    cursor.close()
    cnx.close()

技巧：在 MySQL 中若開啟 autocommit=False，必須自行呼叫 commit()；若忘記回滾，連線關閉時會自動回滾，但在長時間連線的服務中可能造成資源浪費。

範例 4：使用 SQLAlchemy ORM 進行交易（高階抽象）

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

Base = declarative_base()

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

# 建立 Engine（此例使用 SQLite，可換成 PostgreSQL/MySQL）
engine = create_engine('sqlite:///bank.db', echo=False)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)

def transfer(src_id: int, dst_id: int, amount: float):
    """將 amount 從 src 轉到 dst，使用交易保證原子性"""
    session = Session()
    try:
        src = session.get(Account, src_id)
        dst = session.get(Account, dst_id)

        if src.balance < amount:
            raise ValueError("餘額不足")

        src.balance -= amount
        dst.balance += amount

        # session.commit() 會自動提交整個交易
        session.commit()
        print("轉帳成功")
    except Exception as e:
        session.rollback()
        print("轉帳失敗，已回滾。原因:", e)
    finally:
        session.close()

# 範例呼叫
transfer(1, 2, 150.0)

說明：SQLAlchemy 的 Session 本身即是一個交易單位，commit() 前的所有變更都會被緩衝；發生例外時呼叫 rollback() 即可撤銷。這種寫法非常適合 大型專案，因為它把 SQL 隱蔽在 ORM 後，讓開發者專注於業務邏輯。

常見陷阱與最佳實踐

陷阱 可能產生的問題 建議的最佳實踐
忘記 commit() 變更只停留在記憶體，程式結束後自動回滾，資料永遠不會寫入。 養成使用 with 或統一的 commit() 函式，避免手動遺漏。
在迴圈內多次 commit() 交易粒度過小，效能下降，且在錯誤時只能回滾最後一次。 將相關操作合併為單一交易，只在所有操作完成後一次 commit()。
不當的隔離層級 讀取未提交資料（髒讀）或資料不一致（不可重複讀） 根據需求選擇 READ COMMITTED（大多數情況）或 SERIALIZABLE（需要防止幻讀）。
長時間持有交易 會鎖住資料表，導致其他連線等待或死鎖。 盡量縮短交易時間，只在需要的程式碼區段內保持開啟。
例外未捕獲導致自動提交 若程式在 commit() 前拋出未捕獲例外，部分 DB 會自動提交，造成不一致。 使用 try/except 包住整個交易，或交由 with 管理。
混用自動提交與手動提交 交易行為難以預測，可能導致資料遺失或重複寫入。 統一設定：若使用手動交易，務必在連線參數 autocommit=False；若使用自動提交，則不需要 commit()。

小技巧

  1. 自訂裝飾器：把交易管理抽成裝飾器，可讓業務函式保持簡潔。 

    from functools import wraps

def transactional(session_factory):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            session = session_factory()
            try:
                result = func(session, *args, **kwargs)
                session.commit()
                return result
            except Exception:
                session.rollback()
                raise
            finally:
                session.close()
        return wrapper
    return decorator

  2. 使用 SAVEPOINT（子交易）：在大型交易中，若某些步驟失敗只想回滾局部，SAVEPOINT 能提供更細緻的控制。 

    cur.execute("SAVEPOINT sp1")
try:
    cur.execute("INSERT ...")
    cur.execute("UPDATE ...")
except:
    cur.execute("ROLLBACK TO sp1")   # 只回到 sp1 前的狀態

實際應用場景

場景 為什麼需要交易 Python 實作要點
訂單結算（扣庫存 + 建立付款紀錄） 必須保證庫存與付款同步，避免出現「付款成功但庫存不足」的情況。 在同一交易內執行 INSERT payments、UPDATE inventory，失敗則全部 rollback()。
銀行轉帳（扣款 + 收款） 金錢的轉移必須是原子操作，否則會產生「錢不見」或「雙倍入帳」的問題。 使用 SERIALIZABLE 隔離層級，確保同時多筆轉帳不會互相干擾。
批次匯入（大量資料寫入） 若匯入途中發生錯誤，整批資料必須全部回滾，避免半成品資料污染系統。 先 BEGIN; → 多次 INSERT → 成功則 COMMIT;，失敗則 ROLLBACK;。可搭配 executemany() 提升效能。
分布式系統的兩階段提交（跨資料庫） 需要在多個資料庫間保持一致性（例如同時寫入 MySQL 與 PostgreSQL）。 Python 可使用 XA 協議或外部協調者（如 Celery + saga pattern），但仍必須在每個 DB 內部使用交易。
樂觀鎖（版本號） 多人同時編輯同筆資料時，避免「最後寫入者覆蓋」的情況。 在 UPDATE 時加入 WHERE version = :old_version，若受影響列數為 0，表示被其他交易修改，需重新讀取與重試。

總結

  • 交易是資料庫操作的基石：它保證了 原子性、一致性、隔離性、永續性（ACID），防止資料錯亂。
  • 在 Python 中，無論是使用原生 DB‑API（sqlite3、psycopg2、mysql‑connector）或是高階 ORM（SQLAlchemy），開始 → 執行 → 提交/回滾 的流程始終如一。
  • 利用 with、裝飾器、SAVEPOINT 等技巧，可讓交易管理更簡潔、錯誤處理更安全。
  • 謹記 避免長時間持有交易、正確設定隔離層級、統一交易粒度，即可在高併發環境中維持良好效能與資料完整性。

掌握了交易的概念與實作，你就能在 Python 專案中自信地處理任何需要多步驟資料變更的情境，讓系統更可靠，也更容易維護。祝開發順利！