Python 使用 PyMySQL 操作 MySQL 数据库增删改查指南

4. 游标对象详解

游标（Cursor）用于遍历结果集和执行 SQL 命令。PyMySQL 提供多种游标类型：

• Cursor: 默认游标，返回元组列表。
• DictCursor: 返回字典列表，键为列名。
• SSCursor: 单行流式游标，适合大数据量查询，节省内存。

4.1 获取数据的方法

• fetchone(): 获取下一行记录。
• fetchmany(size): 获取指定数量的记录。
• fetchall(): 获取所有剩余记录。
• rowcount: 只读属性，表示最后一次 execute 影响的行数。

5. 数据库操作实战

5.1 创建表 (Create)

import pymysql

db = pymysql.connect(host="localhost", user="root", password="pwd", db="TESTDB")
cursor = db.cursor()

try:
    # 如果存在则删除旧表
    cursor.execute("DROP TABLE IF EXISTS EMPLOYEE")
    
    sql = """
    CREATE TABLE EMPLOYEE (
        FIRST_NAME  CHAR(20) NOT NULL,
        LAST_NAME  CHAR(20),
        AGE INT,
        SEX CHAR(1),
        INCOME FLOAT
    )
    """
    
    cursor.execute(sql)
    db.commit()
    print("表创建成功")
except Exception as e:
    db.rollback()
    print(f"创建失败：{e}")
finally:
    db.close()

5.2 插入数据 (Insert)

注意：为防止 SQL 注入，务必使用参数化查询（占位符 %s），而不是字符串拼接。

import pymysql

db = pymysql.connect(host="localhost", user="root", password="pwd", db="ck", port=3306)
cursor = db.cursor()

sql = "INSERT INTO store (id, company_id, title, address) VALUES (%s, %s, %s, %s)"
data = ('195', '61', '虹桥路店', '虹桥路 1027 号')

try:
    cursor.execute(sql, data)
    db.commit()
    print(f"插入成功，影响行数：{cursor.rowcount}")
except Exception as e:
    db.rollback()
    print(f"插入失败：{e}")
finally:
    db.close()

批量插入：使用 executemany 提高效率。

sql = "INSERT INTO stu (id, name, age) VALUES (%s, %s, %s)"
data_list = [
    ('201611001', 'xiaoqian', 20),
    ('201611002', 'smile', 21),
    ('201611003', 'wood', 23)
]
cursor.executemany(sql, data_list)
db.commit()

5.3 查询数据 (Select)

import pymysql

db = pymysql.connect(host="localhost", user="root", password="pwd", db="test")
cursor = db.cursor()

sql = "SELECT * FROM EMPLOYEE WHERE INCOME > %s"
income_threshold = 1000

try:
    cursor.execute(sql, (income_threshold,))
    results = cursor.fetchall()
    
    for row in results:
        print(f"Name: {row[0]}, Income: {row[4]}")
        # 如果使用 DictCursor，可直接用 row['FIRST_NAME']
except Exception as e:
    print(f"查询失败：{e}")
finally:
    db.close()

5.4 更新数据 (Update)

import pymysql

db = pymysql.connect(host="localhost", user="root", password="pwd", db="test")
cursor = db.cursor()

sql = "UPDATE EMPLOYEE SET AGE = AGE + 1 WHERE SEX = %s"
sex_value = 'M'

try:
    cursor.execute(sql, (sex_value,))
    db.commit()
    print(f"更新成功，影响行数：{cursor.rowcount}")
except Exception as e:
    db.rollback()
    print(f"更新失败：{e}")
finally:
    db.close()

5.5 删除数据 (Delete)

import pymysql

db = pymysql.connect(host="localhost", user="root", password="pwd", db="TESTDB")
cursor = db.cursor()

sql = "DELETE FROM EMPLOYEE WHERE AGE > %s"
age_limit = 20

try:
    cursor.execute(sql, (age_limit,))
    db.commit()
    print(f"删除成功，影响行数：{cursor.rowcount}")
except Exception as e:
    db.rollback()
    print(f"删除失败：{e}")
finally:
    db.close()

6. 事务管理与错误处理

MySQL 默认开启自动提交模式。对于涉及多步操作的场景，建议显式控制事务。

6.1 事务控制

• commit(): 提交当前事务，使更改永久生效。
• rollback(): 回滚当前事务，撤销自上次 commit/rollback 以来的所有更改。

6.2 异常捕获

始终使用 try...except...finally 结构确保连接关闭和数据一致性。

try:
    # 业务逻辑
    conn.commit()
except:
    conn.rollback()
finally:
    conn.close()

7. 最佳实践与注意事项

1. 关闭资源：每次操作后务必关闭游标和连接，或使用上下文管理器。
2. SQL 注入防护：永远不要将用户输入直接拼接到 SQL 字符串中，必须使用参数化查询。
3. 字符集设置：连接时指定 charset='utf8mb4'，避免中文乱码及 Emoji 存储问题。
4. 连接池：在高并发场景下，频繁创建连接开销大，建议使用连接池（如 DBUtils）复用连接。
5. 日志记录：生产环境中应记录异常堆栈信息，便于排查问题。

8. 常见问题排查

• Access Denied: 检查用户名、密码及权限配置。
• Unknown database: 确认 db 参数中的数据库是否存在。
• Lost connection: 网络波动或连接超时，可调整 connect_timeout 参数。
• Data too long: 字段长度不足，需修改表结构。

通过掌握上述内容，您可以高效地在 Python 应用中集成 MySQL 数据库功能，实现稳定可靠的数据持久化操作。