Python 資料庫操作：MySQL 與 PostgreSQL 連線（pymysql、psycopg2）

簡介

在現代的 Python 應用程式中，資料庫是不可或缺的核心元件。無論是網站後端、資料分析平台，還是自動化工具，都需要透過程式與資料庫互動，以讀寫、查詢或更新資料。
MySQL 與 PostgreSQL 是兩大最常見的開源關聯式資料庫系統，分別在 Web 應用與企業級系統中佔有相當高的市佔率。Python 為了與這兩種資料庫溝通，提供了成熟且易上手的套件：pymysql（用於 MySQL）與 psycopg2（用於 PostgreSQL）。

本篇文章將從 安裝、連線、執行 CRUD（Create, Read, Update, Delete） 三個層面，逐步說明如何在 Python 中安全、有效率地操作 MySQL 與 PostgreSQL，並提供實務上常見的 陷阱與最佳實踐，幫助讀者從新手快速成長為中級開發者。

核心概念

1. 為何使用 DB‑API 2.0

Python 官方的資料庫介面規範（PEP 249）稱為 DB‑API 2.0。只要遵守此規範，程式碼就能在不同資料庫之間切換而不需要大幅重構。pymysql 與 psycopg2 都是符合 DB‑API 2.0 的實作，提供統一的：

connect() 取得連線物件
cursor() 建立執行指令的游標
execute()、executemany() 執行 SQL
commit() / rollback() 事務控制

了解此規範後，我們可以把 資料庫類型抽象化，未來若要改用 SQLite、SQL Server，只需要換掉套件與連線參數。

2. 安裝套件

# MySQL 用 pymysql
pip install pymysql

# PostgreSQL 用 psycopg2（建議安裝 binary 版）
pip install psycopg2-binary

Tip：在正式環境建議使用 psycopg2（非 binary）以獲得最佳效能與安全性，並在 requirements.txt 中明確指定版本。

3. 建立連線

3.1 pymysql（MySQL）

import pymysql

# 建立連線，使用字典型別的參數讓程式更具可讀性
conn = pymysql.connect(
    host='127.0.0.1',
    port=3306,
    user='my_user',
    password='my_password',
    database='my_db',
    charset='utf8mb4',   # 支援 Unicode
    cursorclass=pymysql.cursors.DictCursor  # 讓 cursor 回傳 dict 而非 tuple
)

3.2 psycopg2（PostgreSQL）

import psycopg2
import psycopg2.extras

conn = psycopg2.connect(
    host='127.0.0.1',
    port=5432,
    user='pg_user',
    password='pg_password',
    dbname='pg_db',
    # 讓 cursor 回傳 dict，類似 pymysql 的設定
    cursor_factory=psycopg2.extras.RealDictCursor
)

注意：務必將密碼等敏感資訊放在環境變數或設定檔，切勿硬寫在程式碼中。

4. CRUD 範例

以下示範 5 個實用範例，涵蓋從建立資料表到交易控制的完整流程。每個範例皆包含詳細註解，方便初學者快速上手。

4.1 建立資料表（Create）

# 以 pymysql 為例
with conn.cursor() as cur:
    create_sql = """
    CREATE TABLE IF NOT EXISTS users (
        id          INT AUTO_INCREMENT PRIMARY KEY,
        username    VARCHAR(50) NOT NULL UNIQUE,
        email       VARCHAR(100),
        created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    ) ENGINE=InnoDB;
    """
    cur.execute(create_sql)
    conn.commit()  # MySQL 必須手動 commit

PostgreSQL 的語法相同，只是 AUTO_INCREMENT 改為 SERIAL，ENGINE=InnoDB 可以省略。

4.2 插入多筆資料（Insert）

users = [
    ('alice', 'alice@example.com'),
    ('bob',   'bob@example.net'),
    ('carol', 'carol@example.org')
]

insert_sql = "INSERT INTO users (username, email) VALUES (%s, %s);"

with conn.cursor() as cur:
    cur.executemany(insert_sql, users)   # 一次執行多筆
    conn.commit()

executemany() 會自動使用 參數化查詢，避免 SQL Injection。

4.3 單筆查詢（Read）

search_username = 'bob'

select_sql = "SELECT id, username, email, created_at FROM users WHERE username = %s;"

with conn.cursor() as cur:
    cur.execute(select_sql, (search_username,))
    row = cur.fetchone()   # 只取第一筆
    print(row)             # {'id': 2, 'username': 'bob', ...}

4.4 更新資料（Update）與事務（Transaction）

new_email = 'bob_new@example.com'
user_id   = 2

update_sql = "UPDATE users SET email = %s WHERE id = %s;"

try:
    with conn.cursor() as cur:
        cur.execute(update_sql, (new_email, user_id))
    conn.commit()          # 成功才 commit
    print("更新成功")
except Exception as e:
    conn.rollback()        # 發生錯誤則 rollback
    print("更新失敗:", e)

4.5 刪除資料（Delete）與批次操作

delete_sql = "DELETE FROM users WHERE username LIKE %s;"

with conn.cursor() as cur:
    cur.execute(delete_sql, ('c%',))   # 刪除以 c 開頭的使用者
    affected = cur.rowcount
    conn.commit()
    print(f"共刪除 {affected} 筆資料")

5. 連線池（Connection Pool）

在高併發的 Web 應用中，每一次請求都重新建立連線會造成巨大的效能損耗。psycopg2 本身提供 psycopg2.pool 模組，而 MySQL 則可以透過第三方套件 DBUtils 或 SQLAlchemy 的 pool 功能。

5.1 psycopg2 連線池範例

from psycopg2 import pool

pg_pool = pool.SimpleConnectionPool(
    minconn=1,
    maxconn=10,
    host='127.0.0.1',
    port=5432,
    user='pg_user',
    password='pg_password',
    dbname='pg_db'
)

# 取得連線
conn = pg_pool.getconn()
try:
    with conn.cursor() as cur:
        cur.execute("SELECT version();")
        print(cur.fetchone())
finally:
    pg_pool.putconn(conn)   # 歸還至 pool

Tip：在 Flask/Django 等框架中，通常會把 pool 放在應用程式啟動時的全域變數，然後在每個 request 結束時釋放。

常見陷阱與最佳實踐

陷阱	說明	解決方案
硬編碼密碼	程式碼直接寫入帳號密碼，易於洩漏。	使用環境變數（`os.getenv()`）或 Vault 之類的密鑰管理服務。
未使用參數化查詢	直接在字串內拼接變數，會造成 SQL Injection。	永遠使用 `execute(sql, params)`，不要自行組合字串。
忘記 commit/rollback	在自動提交關閉的情況下，變更不會寫入資料庫。	明確呼叫 `conn.commit()`，若捕獲例外則 `conn.rollback()`。
Cursor 未關閉	長時間持有游標會佔用資料庫資源。	使用 with 語句讓游標自動關閉。
連線未釋放	造成連線泄漏，最終資料庫拒絕新連線。	在 finally 區塊或 with 語句中釋放連線，或使用連線池。
字元編碼不一致	中文或特殊符號寫入失敗，出現亂碼。	設定 `charset='utf8mb4'`（MySQL）或在 PostgreSQL 建立資料庫時使用 `ENCODING 'UTF8'`。
大量資料一次插入	使用 `executemany` 仍可能導致效能瓶頸。	考慮 COPY（PostgreSQL）或 LOAD DATA INFILE（MySQL），或分批 commit。

最佳實踐清單

使用參數化查詢，杜絕 SQL Injection。
將敏感資訊抽離，使用 .env、Docker secret 或雲端金鑰服務。
採用連線池，提升併發效能。
統一錯誤處理：捕獲 pymysql.err.* 或 psycopg2.Error，記錄日誌並回滾事務。
寫測試：利用 pytest 搭配測試資料庫（或 SQLite）驗證 CRUD 邏輯。
設定適當的超時（connect_timeout、read_timeout），防止程式卡住。
使用 ORM（如 SQLAlchemy）：若專案規模較大，ORM 能減少手寫 SQL 的錯誤。

實際應用場景

場景	使用套件	為何選擇此套件	相關程式碼片段
部落格網站	`pymysql` + Flask	MySQL 部署成本低、社群支援多；Flask 輕量，適合快速開發。	`app.config['MYSQL_DATABASE_USER'] = os.getenv('DB_USER')`
金融報表平台	`psycopg2` + Django	PostgreSQL 的事務完整性、複雜查詢效能優於 MySQL。	`django.db.backends.postgresql_psycopg2`
資料科學 ETL	`psycopg2` + Pandas	PostgreSQL 支援大量資料寫入與 COPY，配合 `pandas.read_sql` 可直接轉 DataFrame。	`df = pd.read_sql("SELECT * FROM trades", conn)`
IoT 裝置即時資料收集	`pymysql` + ThreadPool	MySQL 的寫入速度足以應付每秒上千筆小筆資料，且可透過分表降低鎖爭用。	`executor.submit(insert_sensor_data, payload)`
多租戶 SaaS	`psycopg2` + SQLAlchemy	PostgreSQL 的 schema 隔離特性，搭配 SQLAlchemy 的 ORM 可輕鬆切換租戶資料庫。	`engine = create_engine(f'postgresql+psycopg2://{user}:{pwd}@{host}/{tenant_db}')`

重點：選擇資料庫與套件時，應根據 資料量、事務需求、團隊熟悉度 以及 未來擴充性 來做判斷。

總結

pymysql 與 psycopg2 都是符合 DB‑API 2.0 的成熟套件，讓我們能以相同的程式結構操作 MySQL 與 PostgreSQL。
正確的 連線設定、參數化查詢、事務控制 以及 資源釋放 是避免資料遺失與安全漏洞的關鍵。
在高併發或大型系統中，連線池、批次寫入、以及 適當的編碼設定 能顯著提升效能與穩定性。
常見陷阱（硬編碼、忘記 commit、游標未關閉）只要養成 with‑statement、環境變數、統一錯誤處理 的好習慣，就能大幅降低風險。

掌握上述概念與實作範例後，你就能在 Python 專案中自信地與 MySQL、PostgreSQL 互動，無論是簡單的 CRUD 功能，還是需要高併發、事務完整性的企業級應用，都能游刃有餘。祝你寫程式愉快，資料庫永遠跑得順暢！