設為首頁 加入收藏
首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
您现在的位置： 程式碼之家 >> 編程語言 >  >> Java 程式設計 >> Java 綜合教學

Python 操作 MySQL 完整指南：從 PyMySQL 到 SQLAlchemy ORM 實戰

     編輯：Java 綜合教學

Python 操作 MySQL 完整指南：從 PyMySQL 到 SQLAlchemy ORM 實戰

在 Python 開發中，操作 MySQL 資料庫主要有兩種主流方式：使用原生驅動模組 PyMySQL 進行底層 SQL 操作，或採用 SQLAlchemy ORM 框架進行物件關係映射開發。本文將系統性介紹這兩種方案的安裝、配置與核心用法，並提供完整程式碼範例供參考。

一、PyMySQL：輕量級原生驅動

PyMySQL 是純 Python 實現的 MySQL 客戶端庫，API 設計與 MySQLdb 高度相容，適合需要精細控制 SQL 執行的場景。

1. 環境安裝

pip3 install pymysql

2. 基礎連線與 SQL 執行

建立連線、創建游標、執行 SQL 並提交事務的標準流程如下：

import pymysql

# 建立資料庫連線
conn = pymysql.connect(
    host='127.0.0.1',
    port=3306,
    user='root',
    passwd='123',
    db='t1',
    charset='utf8mb4'
)

# 創建游標
cursor = conn.cursor()

# 執行單條更新語句
effect_row = cursor.execute("UPDATE hosts SET host = '1.1.1.2'")

# 參數化查詢防止 SQL 注入
# effect_row = cursor.execute("UPDATE hosts SET host = %s WHERE nid > %s", ('1.1.1.2', 1))

# 批量插入
# effect_row = cursor.executemany(
#     "INSERT INTO hosts(host, color_id) VALUES(%s, %s)",
#     [("1.1.1.11", 1), ("1.1.1.11", 2)]
# )

# 提交事務（必須執行，否則修改不生效）
conn.commit()

# 關閉資源
cursor.close()
conn.close()

3. 獲取自增主鍵 ID

執行插入操作後，可透過 cursor.lastrowid 獲取最新生成的自增 ID：

cursor.executemany(
    "INSERT INTO hosts(host, color_id) VALUES(%s, %s)",
    [("1.1.1.11", 1), ("1.1.1.11", 2)]
)
conn.commit()
new_id = cursor.lastrowid  # 獲取最後一條插入記錄的自增 ID
print(f"新記錄 ID: {new_id}")

4. 查詢數據獲取方式

PyMySQL 提供三種游標獲取數據的方法：

  • fetchone()：獲取單條記錄
  • fetchmany(size)：獲取指定數量記錄
  • fetchall()：獲取所有記錄
cursor.execute("SELECT * FROM hosts")
row_1 = cursor.fetchone()      # 第一行
# row_2 = cursor.fetchmany(3)  # 前三行
# row_3 = cursor.fetchall()    # 所有行

5. 游標位置控制

使用 cursor.scroll(offset, mode) 移動游標位置：

  • cursor.scroll(1, mode='relative')：相對當前位置移動
  • cursor.scroll(2, mode='absolute')：相對絕對位置移動

6. 字典類型游標

預設游標返回元組，指定 DictCursor 可返回字典格式數據：

cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
cursor.execute("SELECT * FROM hosts")
row = cursor.fetchone()
# row 格式：{'id': 1, 'host': '1.1.1.1', 'color_id': 1}

二、SQLAlchemy：強大的 ORM 框架

SQLAlchemy 是 Python 生態中最成熟的 ORM 框架，支援 Core（SQL 表達式語言）與 ORM 兩種操作模式，底層依賴 Dialect 機制適配不同資料庫驅動。

1. 安裝與連線字串

pip3 install SQLAlchemy

常見連線字串格式：

# PyMySQL 驅動
mysql+pymysql://username:password@host/dbname[?options]

# MySQLdb 驅動
mysql+mysqldb://user:password@host:port/dbname

# MySQL Connector
mysql+mysqlconnector://user:password@host:port/dbname

2. Core 模式：Engine 直接操作

使用 create_engine 建立引擎，直接執行 SQL：

from sqlalchemy import create_engine

engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/t1",
    max_overflow=5,  # 連線池溢出數量
    echo=True        # 開啟 SQL 日誌（開發環境建議開啟）
)

# 執行插入並獲取自增 ID
# result = engine.execute("INSERT INTO hosts (host, color_id) VALUES ('1.1.1.22', 3)")
# print(result.lastrowid)

# 批量插入（參數化）
# engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",
#     [('1.1.1.22', 3), ('1.1.1.221', 3)]
# )

# 命名參數風格
# engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
#     host='1.1.1.99', color_id=3
# )

# 查詢操作
# result = engine.execute('SELECT * FROM hosts')
# result.fetchone()   # 單條
# result.fetchmany(3) # 多條
# result.fetchall()   # 全部

3. ORM 模式：宣告式映射

定義模型類，自動建表並進行物件操作：

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

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
Base = declarative_base()

# 單表模型
class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))
    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_id_name', '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, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    favor = relationship("Favor", backref='persons')  # 關聯屬性

# 多對多關聯
class ServerToGroup(Base):
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
    group = relationship("Group", backref='server_links')
    server = relationship("Server", backref='group_links')

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)

# 建表 / 刪表
def init_db():
    Base.metadata.create_all(engine)

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

4. Session 會話操作

透過 sessionmaker 創建會話，進行 CRUD 操作：

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

# === 增 ===
obj = Users(name="alex0", extra='sb')
session.add(obj)
session.add_all([
    Users(name="alex1", extra='sb'),
    Users(name="alex2", extra='sb'),
])
session.commit()

# === 刪 ===
session.query(Users).filter(Users.id > 2).delete()
session.commit()

# === 改 ===
session.query(Users).filter(Users.id > 2).update({"name": "099"})
# 使用列運算式更新
session.query(Users).filter(Users.id > 2).update(
    {Users.name: Users.name + "099"}, 
    synchronize_session=False
)
session.commit()

# === 查 ===
users = session.query(Users).all()
names = session.query(Users.name, Users.extra).all()
alex = session.query(Users).filter_by(name='alex').first()

# === 進階查詢 ===
from sqlalchemy import and_, or_, func

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

# 通配符
session.query(Users).filter(Users.name.like('e%')).all()

# 分頁與排序
session.query(Users)[1:2]  # LIMIT 1 OFFSET 1
session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

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

# 連表查詢
session.query(Users, Favor).filter(Users.id == Favor.nid).all()
session.query(Person).join(Favor).all()           # 內連接
session.query(Person).join(Favor, isouter=True).all()  # 左連接

# 集合運算
q1 = session.query(Users.name).filter(Users.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、或不想引入 ORM 開銷的場景。
  • SQLAlchemy Core：適合需要 SQL 靈活性但又想要連線池、方言適配等基礎設施的場景。
  • SQLAlchemy ORM：適合複雜業務邏輯、多表關聯、需要領域模型驅動開發的中大型專案。

實際開發中，可根據專案規模與團隊習慣選擇合適方案，甚至在同一專案中混合使用：核心業務用 ORM，複雜報表查詢用 Core 或原生 SQL。

  1. 上一篇文章：
  2. 下一篇文章：
Java 綜合教學
HibernateTemplate.load 參數類型不匹配：int vs Serializable 解決方案 HibernateTemplate.load 參數類型不匹配：int vs Serializable 解決方案

錯誤現象 The method load(Class, Se
Java 多態下成員訪問完整語義：字段、實例方法、靜態方法三大差異深度解析 Java 多態下成員訪問完整語義：字段、實例方法、靜態方法三大差異深度解析

核心結論速查 成員類型訪問規則綁定時機關鍵字 實例字段按
MyBatis Generator 實戰指南：從配置到生成的完整最佳實踐 MyBatis Generator 實戰指南：從配置到生成的完整最佳實踐

MyBatis Generator 實戰指南：從配置到生成的
Spring Boot 開發「Hello World」Web 應用完整指南 Spring Boot 開發「Hello World」Web 應用完整指南

Spring Boot 開發「Hello World」Web
VMware Ubuntu 忘記密碼單人模式重置全流程 Java 三大框架之 Hibernate 核心概念、工作流程與實戰代碼詳解 Java 集合框架 Map 接口詳解：常用方法、HashTable、HashMap、TreeMap、Properties 區別 Linux 內核網路協議棧：二層（鏈路層）數據包發送流程深度解析 Oracle 位圖索引導致 DML 鎖競爭死鎖：原理重現、AWR 分析與解決方案 Eclipse 常用快捷鍵完整指南：提升開發效率的必備技巧 Java 簡單工廠模式：從 KFC 點餐理解設計模式核心原理
欄目導航
  • Java 程式設計入門
  • Java 相關知識
  • J2EE
  • J2SE
  • J2ME
  • Java 綜合教學
    • Copyright © 程式碼之家 All Rights Reserved