一、为什么选择PyMySQL？

在Python的MySQL生态中，PyMySQL因其纯Python实现、兼容MySQL-python且支持Python3的特性脱颖而出。相比mysql-connector，它在易用性和性能上取得了更好平衡，特别适合需要快速开发的Web应用和数据分析场景。

python

典型应用场景示例

import pymysql
from flask import Flask
app = Flask(name)

数据库配置封装

DBCONFIG = { 'host': '10.0.0.1', 'user': 'appuser',
'password': 'S3cr3t!',
'database': 'production_db',
'charset': 'utf8mb4',
'cursorclass': pymysql.cursors.DictCursor # 返回字典形式结果
}

二、建立高性能数据库连接

2.1 基础连接方式

python
def createconnection(): connection = pymysql.connect( host='localhost', user='devuser',
password='Dev123!',
database='testdb', connecttimeout=5 # 超时设置
)
return connection

使用with语句自动管理连接

try:
with create_connection() as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT VERSION()")
print(f"MySQL版本: {cursor.fetchone()[0]}")
except pymysql.Error as e:
print(f"数据库错误: {e}")

2.2 连接池最佳实践

对于Web应用，直接使用连接池可提升3-5倍性能：python
from DBUtils.PooledDB import PooledDB

pool = PooledDB(
creator=pymysql,
maxconnections=20,
mincached=5,
**DB_CONFIG
)

def querywithpool():
conn = pool.connection()
try:
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM users LIMIT 10")
return cursor.fetchall()
finally:
conn.close()

三、CRUD操作实战技巧

3.1 安全的参数化查询

python

正确示范（防SQL注入）

userid = '123 OR 1=1' with conn.cursor() as cursor: cursor.execute( "SELECT * FROM users WHERE id = %s", (userid,)
)
# 比直接拼接SQL安全100倍

3.2 批量插入优化

python data = [(f'user{i}', f'email{i}@test.com') for i in range(1000)] with conn.cursor() as cursor: cursor.executemany( "INSERT INTO users (username, email) VALUES (%s, %s)", data ) conn.commit() # 批量提交

四、高级特性应用

4.1 事务处理模式

python
def transferfunds(sender, receiver, amount): try: with conn.cursor() as cursor: # 检查余额 cursor.execute("SELECT balance FROM accounts WHERE userid=%s", sender)
if cursor.fetchone()[0] < amount:
raise ValueError("余额不足")

        # 执行转账
        cursor.execute(
            "UPDATE accounts SET balance=balance-%s WHERE user_id=%s",
            (amount, sender)
        )
        cursor.execute(
            "UPDATE accounts SET balance=balance+%s WHERE user_id=%s",
            (amount, receiver)
        )

    conn.commit()  # 成功则提交
except Exception as e:
    conn.rollback()  # 失败回滚
    raise e

4.2 流式读取大数据

python with conn.cursor(pymysql.cursors.SSCursor) as cursor: # 使用服务器端游标 cursor.execute("SELECT * FROM large_table WHERE create_time > %s", ('2023-01-01',)) for row in cursor: process_row(row) # 逐行处理避免内存爆炸

五、性能调优清单

1. 连接管理：始终使用with语句或连接池
2. 字段选择：避免SELECT *，只查询必要字段
3. 索引检查：EXPLAIN分析慢查询
4. 批处理：大量操作使用executemany
5. 类型匹配：Python类型与MySQL字段类型正确对应

python

数据类型映射示例

data = {
'name': '张三', # VARCHAR
'age': 25, # INT
'balance': 100.50, # DECIMAL
'isactive': True, # TINYINT(1) 'createdat': datetime.now() # DATETIME
}

六、常见问题解决方案

连接断开处理：
python def robust_query(sql, args=(), retry=3): for attempt in range(retry): try: with pool.connection() as conn: with conn.cursor() as cursor: cursor.execute(sql, args) return cursor.fetchall() except pymysql.OperationalError as e: if attempt == retry - 1: raise time.sleep(2 ** attempt) # 指数退避

编码问题：
确保连接时指定charset=utf8mb4以支持emoji等特殊字符