Python 操作 MySQL 主流庫：mysqlclient (C 綁定、性能最佳)、PyMySQL (純 Python、兼容性好)、asyncmy (異步)。本文以 mysqlclient 為例，構建生產級 HeroDB 封裝類，支援連接池、參數化查詢、事務上下文、批量操作、資源自動釋放，附完整測試用例。
pip install mysqlclient
# 或
pip install PyMySQL
# 異步
pip install asyncmy#!/usr/bin/env python3
import MySQLdb
from MySQLdb.cursors import DictCursor
from contextlib import contextmanager
from typing import List, Tuple, Any, Optional, Iterator
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HeroDB:
def __init__(self, host: str, user: str, password: str, database: str,
port: int = 3306, charset: str = 'utf8mb4',
pool_size: int = 10):
self.pool = [] # 簡易連接池演示，生產建議用 DBUtils.PooledDB
self.conn_params = dict(host=host, user=user, passwd=password,
db=database, port=port, charset=charset,
cursorclass=DictCursor)
# 預熱連接
for _ in range(min(2, pool_size)):
self.pool.append(MySQLdb.connect(**self.conn_params))
def _get_conn(self):
if self.pool:
conn = self.pool.pop()
if not self._is_alive(conn):
conn = MySQLdb.connect(**self.conn_params)
return conn
return MySQLdb.connect(**self.conn_params)
def _put_conn(self, conn):
if len(self.pool) < 10:
self.pool.append(conn)
else:
conn.close()
def _is_alive(self, conn):
try:
conn.ping(reconnect=True)
return True
except MySQLdb.Error:
return False
@contextmanager
def connection(self) -> Iterator[MySQLdb.Connection]:
conn = self._get_conn()
try:
yield conn
finally:
self._put_conn(conn)
@contextmanager
def transaction(self) -> Iterator[MySQLdb.Connection]:
conn = self._get_conn()
try:
conn.autocommit(False)
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.autocommit(True)
self._put_conn(conn)
def execute(self, sql: str, params: Tuple = ()) -> int:
with self.connection() as conn:
with conn.cursor() as cur:
return cur.execute(sql, params)
def executemany(self, sql: str, params_list: List[Tuple]) -> int:
with self.connection() as conn:
with conn.cursor() as cur:
return cur.executemany(sql, params_list)
def fetchone(self, sql: str, params: Tuple = ()) -> Optional[dict]:
with self.connection() as conn:
with conn.cursor() as cur:
cur.execute(sql, params)
return cur.fetchone()
def fetchall(self, sql: str, params: Tuple = ()) -> List[dict]:
with self.connection() as conn:
with conn.cursor() as cur:
cur.execute(sql, params)
return cur.fetchall()
def fetchmany(self, sql: str, params: Tuple = (), size: int = 10) -> List[dict]:
with self.connection() as conn:
with conn.cursor() as cur:
cur.execute(sql, params)
return cur.fetchmany(size)
# 便捷 CRUD
def insert(self, table: str, data: dict) -> int:
cols = ', '.join(f'`{k}`' for k in data.keys())
placeholders = ', '.join(['%s'] * len(data))
sql = f'INSERT INTO `{table}` ({cols}) VALUES ({placeholders})'
with self.connection() as conn:
with conn.cursor() as cur:
cur.execute(sql, tuple(data.values()))
return cur.lastrowid
def update(self, table: str, data: dict, where: str, where_params: Tuple) -> int:
sets = ', '.join(f'`{k}`=%s' for k in data.keys())
sql = f'UPDATE `{table}` SET {sets} WHERE {where}'
with self.connection() as conn:
with conn.cursor() as cur:
return cur.execute(sql, tuple(data.values()) + where_params)
def delete(self, table: str, where: str, where_params: Tuple) -> int:
sql = f'DELETE FROM `{table}` WHERE {where}'
with self.connection() as conn:
with conn.cursor() as cur:
return cur.execute(sql, where_params)
def close_all(self):
for conn in self.pool:
conn.close()
self.pool.clear()#!/usr/bin/env python3
from hero_db import HeroDB
db = HeroDB(host='localhost', user='root', password='your_pass',
database='hero', pool_size=5)
# 建表示例 (一次性)
db.execute('''
CREATE TABLE IF NOT EXISTS heros (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(32) NOT NULL,
sex TINYINT NOT NULL DEFAULT 0,
age INT NOT NULL DEFAULT 0,
info VARCHAR(255)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
''')
# 插入單條
db.insert('heros', {'name': 'Zeus', 'sex': 1, 'age': 5000, 'info': 'King of Gods'})
# 批量插入
heros = [
{'name': 'Athena', 'sex': 0, 'age': 3000, 'info': 'Goddess of Wisdom'},
{'name': 'Thor', 'sex': 1, 'age': 1500, 'info': 'God of Thunder'},
]
db.executemany(
'INSERT INTO heros (name, sex, age, info) VALUES (%s, %s, %s, %s)',
[(h['name'], h['sex'], h['age'], h['info']) for h in heros]
)
# 查詢
print('All:', db.fetchall('SELECT * FROM heros'))
print('First:', db.fetchone('SELECT * FROM heros ORDER BY id LIMIT 1'))
print('Page 1:', db.fetchmany('SELECT * FROM heros ORDER BY id', size=2))
# 事務演示：轉移年齡 (原子操作)
try:
with db.transaction() as conn:
with conn.cursor() as cur:
cur.execute('UPDATE heros SET age = age - 100 WHERE name = %s', ('Zeus',))
cur.execute('UPDATE heros SET age = age + 100 WHERE name = %s', ('Thor',))
# 故意報錯測試回滾
# raise Exception('模擬失敗')
except Exception as e:
logger.error(f'事務回滾: {e}')
# 刪除
db.delete('heros', 'name = %s', ('Athena',))
db.close_all()| 主題 | 方案 |
|---|---|
| 生產級連接池 | DBUtils.PooledDB 或 sqlalchemy.pool.QueuePool |
| ORM | SQLAlchemy (Core/ORM)、Tortoise ORM (異步)、Peewee |
| 異步操作 | asyncmy + asyncio / aiomysql |
| 參數化防注入 | 永遠用 %s 佔位符，絕不用 f-string 拼接 SQL |
| 大結果集流式讀取 | SSCursor (服務端游標) 避免一次性載入內存 |
| 連接保活 | 連接池 ping(reconnect=True) 或 recycle=3600 |
MySQLdb 參數佔位符是 %s (非 ?)，所有類型統一 %s。cursorclass=DictCursor 返回字典更易用。commit/rollback。executemany，單條循環極慢。ping(reconnect=True) 防止 MySQL server has gone away。Python 操作 MySQL 核心三板斧：參數化查詢防注入、連接池復用性能、上下文管理器保證資源釋放。生產環境請直接上SQLAlchemy+PyMySQL/mysqlclient+DBUtils.PooledDB，專注業務而非造輪子。
