Python 操作 MySQL 實戰：MySQLdb 安裝與 CRUD 完整示例

Python 操作 MySQL 實戰：MySQLdb 安裝與 CRUD 完整示例

MySQLdb（即 mysqlclient）是 Python 連接 MySQL 最成熟、高效的驅動之一。本文覆蓋 Windows 下安裝、連接池配置、參數化查詢防注入、事務控制等實戰要點。

一、安裝指南

Windows 預編譯輪子安裝（推薦）

1. 訪問 UCI 非官方輪子庫 下載對應版本，如 mysqlclient-2.1.1-cp310-cp310-win_amd64.whl
2. 命令行安裝：pip install mysqlclient-2.1.1-cp310-cp310-win_amd64.whl

Linux/macOS 系統包管理器

# Debian/Ubuntu
sudo apt-get install python3-dev default-libmysqlclient-dev build-essential
pip3 install mysqlclient

# CentOS/RHEL
sudo yum install python3-devel mysql-devel gcc
pip3 install mysqlclient

二、核心 API 與最佳實踐

import MySQLdb
import time
from contextlib import contextmanager

# 連接配置建議從環境讀取或配置文件
DB_CONFIG = {
    'host': 'localhost',
    'user': 'root',
    'passwd': 'root',
    'db': 'Python',
    'charset': 'utf8mb4',  # 支援 emoji
    'autocommit': False,   # 手動控制事務
}

@contextmanager
def get_cursor():
    conn = MySQLdb.connect(**DB_CONFIG)
    try:
        cursor = conn.cursor(MySQLdb.cursors.DictCursor)  # 返回字典游標
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

# === 示例業務邏輯 ===
with get_cursor() as cur:
    # 1. 建表
    cur.execute("""
        CREATE TABLE IF NOT EXISTS thinkgamer (
            name VARCHAR(128) PRIMARY KEY,
            created INT(10) NOT NULL
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """)

    # 2. 單條插入（參數化，防 SQL 注入）
    sql = "INSERT INTO thinkgamer (name, created) VALUES (%s, %s)"
    cur.execute(sql, ("aaa", int(time.time())))
    print(f"insert: {cur.rowcount}")

    # 3. 批量插入
    data = [("bbb", int(time.time())), ("ccc", 33), ("ddd", 44)]
    cur.executemany(sql, data)
    print(f"insertmany: {cur.rowcount}")

    # 4. 更新
    cur.execute("UPDATE thinkgamer SET name=%s WHERE name=%s", ("zzz", "aaa"))
    print(f"update: {cur.rowcount}")

    # 5. 查詢
    cur.execute("SELECT * FROM thinkgamer")
    for row in cur.fetchall():
        print(row)  # DictCursor 返回 {'name': 'zzz', 'created': 1700000000}

    # 6. 刪除
    cur.execute("DELETE FROM thinkgamer WHERE name=%s", ("bbb",))
    print(f"delete: {cur.rowcount}")

    # 7. 再次查詢驗證
    cur.execute("SELECT * FROM thinkgamer")
    print(cur.fetchall())

三、進階技巧

• 連接池：生產環境請使用 DBUtils.PooledDB 或 SQLAlchemy Engine 管理連接，避免頻繁建連開銷。
• 異步支援：高併發場景可考慮 aiomysql（基於 asyncio）或 asyncmy。
• 類型轉換：DictCursor 返回 datetime、Decimal 等原生類型，便於業務層直接使用。
• SSL 連接：ssl={'ca': '/path/ca.pem'} 參數啟用加密傳輸。

四、常見報錯排查