您现在的位置：程式碼之家 >> 數據庫知識 > >> MySQL 資料庫 >> MySQL 綜合教學

Python 操作 MySQL 完全指南：從 pymysql 到 SQLAlchemy ORM 實戰

編輯：MySQL 綜合教學

Python 操作 MySQL 完全指南：從 pymysql 到 SQLAlchemy ORM 實戰

在 Python 開發中，操作 MySQL 資料庫主要有兩種主流方式：使用原生驅動 pymysql 直接執行 SQL，或是透過 SQLAlchemy ORM 進行物件關聯映射操作。本文將系統性介紹這兩種方式的核心用法、進階技巧與最佳實踐，幫助你快速上手並避開常見坑點。

一、pymysql：原生 SQL 操作的基石

pymysql 是純 Python 實現的 MySQL 客戶端庫，API 設計貼近 DB-API 2.0 標準，適合需要精細控制 SQL 或效能敏感的場景。

1. 安裝與連線

pip3 install pymysql

建立連線時建議使用字典游標（DictCursor），回傳結果為字典格式，欄位名可直接作為鍵存取，程式碼更易讀：

import pymysql
conn = pymysql.connect(
    host='127.0.0.1', port=3306,
    user='root', passwd='123', db='t1',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

2. 執行增刪改查與事務控制

所有操作皆透過游標進行。修改資料後必須 conn.commit() 提交事務，否則不會持久化。

cursor = conn.cursor()
# 參數化查詢，防止 SQL 注入
effect_row = cursor.execute(
    "UPDATE hosts SET host = %s WHERE nid > %s",
    ('1.1.1.2', 1)
)
# 批次新增
cursor.executemany(
    "INSERT INTO hosts(host, color_id) VALUES (%s, %s)",
    [("1.1.1.11", 1), ("1.1.1.12", 2)]
)
conn.commit()
cursor.close()
conn.close()

3. 取得自增 ID 與查詢結果

插入資料後，cursor.lastrowid 可取得最新自增主鍵值。查詢結果提供三種擷取方式：

fetchone()：單筆
fetchmany(n)：前 n 筆
fetchall()：全部

4. 游標控制與資料類型轉換

使用 cursor.scroll(offset, mode='relative|absolute') 移動游標位置。若需字典型結果，建立連線時指定 cursorclass=pymysql.cursors.DictCursor 即可。

5. 使用 `contextlib` 簡化資源管理

利用上下文管理器自動處理連線開關與事務提交/回滾：

import contextlib

@contextlib.contextmanager
def mysql_conn(**cfg):
    conn = pymysql.connect(**cfg, cursorclass=pymysql.cursors.DictCursor)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

# 使用範例
with mysql_conn(host='127.0.0.1', user='root', passwd='123', db='t1') as cur:
    cur.execute("SELECT * FROM hosts")
    print(cur.fetchall())

二、SQLAlchemy ORM：物件導向的資料庫操作

SQLAlchemy 是 Python 生態系中最強大的 ORM 框架，核心理念是「將類別映射為資料表，將物件實例映射為資料列」。它建立在 DBAPI（如 pymysql）之上，支援多種資料庫方言。

1. 安裝與引擎建立

pip3 install SQLAlchemy pymysql

連線字串格式：mysql+pymysql://user:pass@host:port/dbname?charset=utf8mb4。create_engine 內建連線池，max_overflow 控制額外連線數。

from sqlalchemy import create_engine
engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8mb4",
    max_overflow=5, pool_pre_ping=True
)

2. 核心表達式語言

若不想定義 ORM 模型，可直接用 engine.execute() 執行原生 SQL，回傳 ResultProxy 支援 fetchone()/fetchall() 等方法。

3. 宣告式 ORM 模型定義

繼承 declarative_base() 產生的基類，定義類別屬性對應欄位：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import relationship, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(32), nullable=False)
    extra = Column(String(16))
    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_name_extra', 'name', 'extra'),
    )
    def __repr__(self):
        return f""

# 一對多關聯
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)

class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    favor_id = Column(Integer, ForeignKey('favor.nid'))
    favor = relationship("Favor", backref='persons')

# 多對多關聯（需中間表）
class ServerToGroup(Base):
    __tablename__ = 'server_to_group'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
    server = relationship("Server", backref='s2g')
    group = relationship("Group", backref='s2g')

class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)

class Server(Base):
    __tablename__ = 'server'
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)
    groups = relationship("Group", secondary="server_to_group", backref='servers')

4. 建立/刪除資料表

def init_db():
    Base.metadata.create_all(engine)

def drop_db():
    Base.metadata.drop_all(engine)

5. Session 與 CRUD 操作

sessionmaker(bind=engine) 建立 Session 工廠，每次操作實例化一個 Session 物件：

Session = sessionmaker(bind=engine)
session = Session()

# 新增
user = User(name="張三", extra="三兒")
session.add(user)
session.add_all([User(name="李四", extra="四兒"), User(name="王五", extra="五兒")])
session.commit()

# 刪除
session.query(User).filter(User.id > 2).delete()
session.commit()

# 修改
session.query(User).filter(User.id > 2).update({"name": "nick", "extra": "updated"})
session.commit()

# 查詢
users = session.query(User).all()
user = session.query(User).filter_by(name='nick').first()
names = session.query(User.name, User.extra).all()

6. 進階查詢技巧

from sqlalchemy import and_, or_, func
from sqlalchemy.sql import func

# 條件組合
session.query(User).filter(and_(User.id > 3, User.name == 'nick')).all()
session.query(User).filter(or_(User.id < 2, User.name == 'nick')).all()

# 模糊查詢
session.query(User).filter(User.name.like('n%')).all()

# 排序、分頁
session.query(User).order_by(User.name.desc(), User.id.asc()).limit(10).offset(20).all()

# 聚合與分組
session.query(func.max(User.id), func.count(User.id)).group_by(User.name).having(func.min(User.id) > 2).all()

# 連表查詢
session.query(User, Favor).filter(User.id == Favor.nid).all()
session.query(Person).join(Favor).all()           # INNER JOIN
session.query(Person).join(Favor, isouter=True).all()  # LEFT JOIN

# 集合運算
q1 = session.query(User.name).filter(User.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
q1.union(q2).all()        # UNION 去重
q1.union_all(q2).all()    # UNION ALL 保留重複

總結：pymysql 適合輕量、高效能、需手寫 SQL 的場景；SQLAlchemy ORM 則在複雜業務邏輯、關聯查詢、遷移維護上優勢明顯。實際專案常混合使用：核心高頻查詢用 Core/原生 SQL，業務邏輯用 ORM，取長補短。