Python 課程 – 資料庫操作（Databases）

主題：查詢結果轉換

簡介

在日常的 Python 應用程式中，資料庫查詢是最常見的需求之一。無論是網站後端、資料分析或是自動化工具，都需要把資料庫回傳的結果轉換成程式可以直接操作的資料型別。
如果不熟悉查詢結果的結構與轉換方式，往往會出現「欄位名稱對不上」或「資料型別錯誤」的問題，甚至造成效能瓶頸。

本篇文章將以 SQLite、MySQL、PostgreSQL 為例，說明如何把查詢結果 轉成字典、資料類別 (dataclass)、Pandas DataFrame，以及在轉換過程中應注意的細節。文章以 淺顯易懂 為原則，適合剛入門的初學者，也能提供給中級開發者作為實務參考。

核心概念

1️⃣ 資料庫查詢回傳的原始型別

大多數 Python 的 DB‑API (PEP 249) 在執行 cursor.execute() 後，會回傳 tuple 的序列，每個 tuple 代表一筆資料，欄位的順序與 SQL 語句中 SELECT 的欄位順序相同。例如：

[(1, 'Alice', 25), (2, 'Bob', 30)]

這樣的結構雖然簡潔，但在程式碼中直接使用索引 (row[1]) 會降低可讀性，且在欄位變動時容易出錯。因此，我們需要轉換成更具意義的型別。

2️⃣ 轉成字典（`dict`）

2.1 使用 `cursor.description` 取得欄位名稱

import sqlite3

conn = sqlite3.connect(':memory:')
cur = conn.cursor()

# 建立測試資料表
cur.execute('CREATE TABLE users (id INTEGER, name TEXT, age INTEGER)')
cur.executemany('INSERT INTO users VALUES (?,?,?)',
                [(1, 'Alice', 25), (2, 'Bob', 30)])

# 執行查詢
cur.execute('SELECT id, name, age FROM users')
rows = cur.fetchall()

# 取得欄位名稱
col_names = [desc[0] for desc in cur.description]

# 轉成字典列表
result = [dict(zip(col_names, row)) for row in rows]
print(result)
# [{'id': 1, 'name': 'Alice', 'age': 25}, {'id': 2, 'name': 'Bob', 'age': 30}]

重點：cursor.description 是一個由 7 個元素組成的 tuple，第一個元素就是欄位名稱。

2.2 使用 `sqlite3.Row`（SQLite 專屬）

conn.row_factory = sqlite3.Row   # 設定回傳型別為 Row
cur = conn.cursor()
cur.execute('SELECT * FROM users')
rows = cur.fetchall()

# Row 物件本身就支援 dict 介面
result = [dict(row) for row in rows]
print(result)

此方式不需要手動取得欄位名稱，可讀性更佳，但僅限於 SQLite。

3️⃣ 轉成資料類別（`dataclass`）

Python 3.7+ 引入 dataclasses，讓結構化資料的宣告變得簡單。配合查詢結果，我們可以把每筆資料映射成 User 物件。

from dataclasses import dataclass
import mysql.connector

@dataclass
class User:
    id: int
    name: str
    age: int

# 以 MySQL 為例
conn = mysql.connector.connect(host='localhost', user='root',
                               password='pwd', database='test')
cur = conn.cursor()
cur.execute('SELECT id, name, age FROM users')
rows = cur.fetchall()

users = [User(*row) for row in rows]
print(users)   # [User(id=1, name='Alice', age=25), User(id=2, name='Bob', age=30)]

技巧：*row 會把 tuple 展開成位置參數，順序必須與 dataclass 欄位相同。若欄位名稱與資料庫不一致，可自行建立映射字典再傳入。

4️⃣ 轉成 Pandas DataFrame

在資料分析或報表產出時，Pandas 是首選工具。直接把查詢結果轉成 DataFrame，即可使用豐富的資料操作函式。

import pandas as pd
import psycopg2

# PostgreSQL 連線
conn = psycopg2.connect(host='localhost', dbname='test', user='postgres', password='pwd')
cur = conn.cursor()
cur.execute('SELECT id, name, age FROM users')
rows = cur.fetchall()

# 取得欄位名稱
col_names = [desc[0] for desc in cur.description]

df = pd.DataFrame(rows, columns=col_names)
print(df)
#    id   name  age
# 0   1  Alice   25
# 1   2    Bob   30

如果資料量非常大，建議使用 pandas.read_sql()，它會在底層使用資料庫的游標，直接把結果流式寫入 DataFrame，減少記憶體佔用。

df = pd.read_sql('SELECT id, name, age FROM users', conn)

5️⃣ 轉成 JSON（供 API 使用）

Web API 常以 JSON 格式回傳資料，以下示範把字典列表直接序列化：

import json

json_str = json.dumps(result, ensure_ascii=False, indent=2)
print(json_str)

注意：若欄位包含 datetime、Decimal 等非 JSON 原生型別，需要自行 自訂序列化，或使用 default=str 參數。

常見陷阱與最佳實踐

陷阱	說明	解決方式
欄位順序不一致	直接 `User(*row)` 依賴欄位順序，若 SQL 改變 SELECT 的欄位順序會出錯。	使用關鍵字參數 `User(**dict(zip(col_names, row)))`，或在 `dataclass` 中使用 `field(metadata={'db': 'column_name'})` 進行映射。
大量資料一次載入	`fetchall()` 會一次把所有資料讀入記憶體，資料量大時會 OOM。	使用 `fetchmany(size)` 或迭代游標 (`for row in cur:`) 逐批處理；或直接使用 `pandas.read_sql()` 的 `chunksize` 參數。
資料型別不匹配	DB‑API 會把 `DATE`、`TIMESTAMP` 轉成 `datetime.date`、`datetime.datetime`，但 JSON 無法直接序列化。	在轉 JSON 前先轉換為字串，或使用自訂的 `json.JSONEncoder`。
欄位名稱大小寫	部分資料庫（如 PostgreSQL）預設把欄位名稱轉成小寫，導致程式碼中 `User.id` 找不到。	在 SQL 中使用雙引號強制大小寫，或在程式碼中統一使用 `lower()` 處理欄位名稱。
連線未關閉	開啟大量查詢卻忘記 `conn.close()`，會耗盡資料庫連線池。	使用 context manager (`with sqlite3.connect(...) as conn:`) 或自行在 `finally` 區塊關閉。

最佳實踐：

統一欄位名稱：在資料庫層面盡量使用 snake_case，程式碼層面保持相同命名規則。
使用 context manager：確保連線與游標在使用完畢後自動釋放。
分批處理：對於超過 10,000 筆的查詢，建議每次取 1,000~5,000 筆。
型別安全：在 dataclass 或 Pydantic Model 中明確宣告型別，讓 IDE 能提供型別提示，減少錯誤。
測試轉換邏輯：寫單元測試驗證「欄位變動」或「資料型別變更」時的行為。

實際應用場景

場景	為什麼需要轉換	推薦的轉換方式
Web API 回傳使用者清單	前端期待 JSON 陣列，每筆資料包含 `id、name、email`。	`fetchall()` → `dict` → `json.dumps()`
資料分析報表	分析師需要使用 Pandas 進行統計、圖表。	`pandas.read_sql()` 或 `DataFrame(rows, columns=col_names)`
批次匯入至另一系統	目標系統接受 CSV 或自訂資料類別。	`dict` → `csv.DictWriter` 或 `dataclass` → 自訂序列化
即時監控儀表板	每秒鐘抓取最新指標，需要低延遲。	使用游標迭代 (`for row in cur:`) 直接寫入 Redis 或 InfluxDB。
資料遷移	從舊資料庫搬遷至新資料庫，需保留欄位映射。	先轉成 `dataclass`，再利用 ORM（如 SQLAlchemy）寫入新庫。

總結

查詢結果的轉換是 Python 與資料庫互動的關鍵一步。透過 cursor.description、sqlite3.Row、dataclass、pandas、以及 JSON 序列化，我們可以把原始的 tuple 陣列變成 易讀、易維護 的資料結構。
在實作過程中，務必要注意 欄位順序、資料型別、記憶體使用 與 連線資源釋放，並遵循「分批處理」與「型別安全」的最佳實踐。

掌握了這些技巧後，你就能在 Web 開發、資料分析、系統整合 等各種情境中，快速、可靠地把資料庫查詢結果轉換成程式所需的形式，提升開發效率與程式品質。祝你在 Python 資料庫操作的旅程中玩得開心、寫得順利！