Python MySQL 查詢教學：fetchone、fetchmany、fetchall 完整範例

Python MySQL 查詢教學：fetchone、fetchmany、fetchall 完整範例

基於 mysql-connector-python 驅動，示範三種結果集獲取方法的適用場景與記憶體效能權衡。

環境準備

pip install mysql-connector-python
# 配置檔 python_mysql_dbconfig.py
def read_db_config():
    return {
        'host': 'localhost',
        'database': 'testdb',
        'user': 'root',
        'password': 'password'
    }

1. fetchone() - 逐行迭代，記憶體最省

from mysql.connector import MySQLConnection, Error
from python_mysql_dbconfig import read_db_config

def query_with_fetchone():
    try:
        dbconfig = read_db_config()
        conn = MySQLConnection(**dbconfig)
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM books")
        
        row = cursor.fetchone()
        while row is not None:
            print(row)
            row = cursor.fetchone()
            
    except Error as e:
        print(e)
    finally:
        cursor.close()
        conn.close()

if __name__ == '__main__':
    query_with_fetchone()

適用： 超大結果集、串流處理、記憶體受限環境

2. fetchall() - 一次性載入全部，適合小結果集

def query_with_fetchall():
    try:
        dbconfig = read_db_config()
        conn = MySQLConnection(**dbconfig)
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM books")
        
        rows = cursor.fetchall()
        print('Total Row(s):', cursor.rowcount)
        for row in rows:
            print(row)
            
    except Error as e:
        print(e)
    finally:
        cursor.close()
        conn.close()

適用： 結果集較小 (如 < 10 萬行)、需隨機存取、後續多次遍歷

3. fetchmany() - 分批獲取，平衡記憶體與效能

def iter_row(cursor, size=10):
    """生成器：分批 yield 行資料"""
    while True:
        rows = cursor.fetchmany(size)
        if not rows:
            break
        for row in rows:
            yield row

def query_with_fetchmany():
    try:
        dbconfig = read_db_config()
        conn = MySQLConnection(**dbconfig)
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM books")
        
        for row in iter_row(cursor, 10):  # 每次取 10 行
            print(row)
            
    except Error as e:
        print(e)
    finally:
        cursor.close()
        conn.close()

適用： 中大型結果集、避免一次性載入 OOM、需控制批次大小

選擇指南

進階提示

• 使用 cursor = conn.cursor(buffered=True) 可客戶端緩衝結果集，支援 cursor.rowcount 即時可用
• 大量寫入用 cursor.executemany() 批次執行
• 連線池：mysql.connector.pooling.MySQLConnectionPool
• 異步支援：aiomysql / asyncmy (Python 3.7+)