MySQL Python pymysql 连接与使用

pymysql 是 Python 中操作 MySQL 的纯Python驱动库，支持 MySQL 5.7 / 8.0，兼容 Python3。

本教程使用到的 Websites 表 SQL 文件：websites.sql。

旧的 MySQLdb 只支持 Python2，新项目统一使用 pymysql。

一、环境准备

1. 安装 pymysql

pip install pymysql

2. 准备测试数据表（沿用统一的 websites 表，和前面教程保持一致）

CREATE DATABASE IF NOT EXISTS test;
USE test;
DROP TABLE IF EXISTS websites;
CREATE TABLE websites (
  id INT PRIMARY KEY AUTO_INCREMENT,
  name VARCHAR(20),
  url VARCHAR(255),
  alexa INT,
  country VARCHAR(10)
);INSERT INTO websites(name,url,alexa,country) VALUES
('Google','https://www.google.com',1,'USA'),
('淘宝','https://www.taobao.com',13,'CN'),
('runoops教程','http://runoops.com',6988,'CN'),
('微博','http://weibo.com',22,'CN'),
('Facebook','https://www.facebook.com',3,'USA');

二、基础单次连接（脚本测试用，Web项目不推荐）

单次连接适合简单脚本；Web服务高并发场景推荐使用连接池 DBUtils.PooledDB。

import pymysql# 创建数据库连接
conn = pymysql.connect(
    host="localhost",
    port=3306,
    user="root",
    password="你的数据库密码",
    database="test",
    charset="utf8mb4"  # utf8mb4支持中文+emoji，不要写utf8
)
# 获取游标
cursor = conn.cursor()
# 执行SQL
sql = "SELECT 1+1 AS res"
cursor.execute(sql)
result = cursor.fetchone()
print(result)
# 关闭游标、连接
cursor.close()
conn.close()

关键说明：

charset="utf8mb4"，MySQL 的 utf8 不是完整utf-8，存在emoji乱码问题；

端口默认3306，如果修改过MySQL端口需要手动指定 port=xxx。

三、CRUD 增删改查

⚠️ 安全原则：使用 %s 占位符，禁止直接字符串拼接SQL，防止SQL注入！
pymysql占位符是 %s，不是Python字符串格式化，不要混淆。

3.1 查询 SELECT

fetchone() 获取单行，fetchall() 获取全部结果，fetchmany(n) 获取指定行数。

import pymysqlconn = pymysql.connect(
    host="localhost",
    user="root",
    password="你的数据库密码",
    database="test",
    charset="utf8mb4"
)
cursor = conn.cursor(pymysql.cursors.DictCursor)  # DictCursor 返回字典格式，方便读取
sql = "SELECT id,name,url,alexa,country FROM websites WHERE alexa < %s"
cursor.execute(sql, [50])  # 参数放在列表传入
rows = cursor.fetchall()
for row in rows:
    print(f"id:{row['id']}, name:{row['name']}, url:{row['url']}")
cursor.close()
conn.close()

3.2 插入 INSERT

增删改操作，执行完成后必须执行 conn.commit() 提交；出错用 conn.rollback() 回滚。

import pymysqlconn = pymysql.connect(
    host="localhost",
    user="root",
    password="你的数据库密码",
    database="test",
    charset="utf8mb4"
)
cursor = conn.cursor()
sql = "INSERT INTO websites(name,url,alexa,country) VALUES (%s,%s,%s,%s)"
params = ("Python教程", "https://python-demo.com", 23456, "CN")
try:
    cursor.execute(sql, params)
    conn.commit()
    print("新增ID：", cursor.lastrowid)
except Exception as e:
    conn.rollback()
    print("插入失败", e)
cursor.close()
conn.close()

3.3 更新 UPDATE

import pymysqlconn = pymysql.connect(
    host="localhost",
    user="root",
    password="你的数据库密码",
    database="test",
    charset="utf8mb4"
)
cursor = conn.cursor()
sql = "UPDATE websites SET name=%s, url=%s WHERE id=%s"
params = ("Python手册", "https://python-ref.com", 6)
try:
    rowcount = cursor.execute(sql, params)
    conn.commit()
    print(f"更新影响行数：{rowcount}")
except Exception as e:
    conn.rollback()
    print("更新失败", e)
cursor.close()
conn.close()

3.4 删除 DELETE

import pymysqlconn = pymysql.connect(
    host="localhost",
    user="root",
    password="你的数据库密码",
    database="test",
    charset="utf8mb4"
)
cursor = conn.cursor()
sql = "DELETE FROM websites WHERE id = %s"
params = [6]
try:
    rowcount = cursor.execute(sql, params)
    conn.commit()
    print(f"删除影响行数：{rowcount}")
except Exception as e:
    conn.rollback()
    print("删除失败", e)
cursor.close()
conn.close()

四、事务处理

MySQL InnoDB支持事务；pymysql默认自动提交关闭，DML操作需要手动commit。

import pymysqlconn = pymysql.connect(
    host="localhost",
    user="root",
    password="你的数据库密码",
    database="test",
    charset="utf8mb4"
)
cursor = conn.cursor()
try:
    # 多条关联SQL
    cursor.execute("UPDATE account SET balance = balance - 100 WHERE id=%s", [1])
    cursor.execute("UPDATE account SET balance = balance + 100 WHERE id=%s", [2])
    conn.commit()
    print("事务提交成功")
except Exception as e:
    conn.rollback()
    print("事务异常，已回滚：", e)
finally:
    cursor.close()
    conn.close()

五、生产推荐：连接池 DBUtils

频繁创建销毁连接性能差，Web项目推荐使用 DBUtils.PooledDB 连接池。

安装DBUtils

pip install DBUtils

连接池示例：

import pymysql
from DBUtils.PooledDB import PooledDB# 创建连接池
pool = PooledDB(
    creator=pymysql,
    maxconnections=10,    # 最大连接数
    mincached=2,          # 初始化空闲连接
    maxcached=5,
    host="localhost",
    user="root",
    password="你的数据库密码",
    database="test",
    charset="utf8mb4"
)
# 从池中获取连接
conn = pool.connection()
cursor = conn.cursor(pymysql.cursors.DictCursor)
sql = "SELECT * FROM websites"
cursor.execute(sql)
rows = cursor.fetchall()
print(rows)
# 关闭游标，连接自动归还连接池，不会真正断开
cursor.close()
conn.close()

六、错误处理与调试

规范捕获异常示例

import pymysqltry:
    conn = pymysql.connect(
        host="localhost",
        user="root",
        password="你的数据库密码",
        database="test",
        charset="utf8mb4"
    )
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM websites WHERE id=%s", [1])
    data = cursor.fetchone()
    print(data)
except pymysql.MySQLError as err:
    print(f"MySQL错误码：{err.args[0]}, 错误信息：{err.args[1]}")
finally:
    # 安全关闭资源
    if 'cursor' in locals():
        cursor.close()
    if 'conn' in locals():
        conn.close()

常见错误

Access denied for user：账号密码错误，数据库用户授权问题
Unknown database 'test'：数据库不存在
Table doesn't exist：数据表不存在
Lost connection to MySQL server：连接超时、防火墙、MySQL服务异常
中文乱码：连接时没有设置 charset="utf8mb4"

调试技巧：开发环境打印完整SQL与参数；生产环境禁止把数据库原始异常返回前端，只记录日志。

七、性能优化与安全建议

安全

永远使用占位符 %s，严禁字符串拼接SQL，防止SQL注入。
数据库账号遵循最小权限原则，业务账号禁止 DROP、ALTER。
密码不要硬编码，放到环境变量或配置文件，不要提交到代码仓库。

SQL优化

避免 SELECT *，只查询需要的字段。
大数据查询加 LIMIT，避免一次性加载大量数据造成内存占用高。
WHERE、ORDER BY 字段建立索引；%前缀模糊查询会导致索引失效。
批量插入使用 executemany()，不要循环单次execute。

# 批量插入示例
data_list = [("A站点","a.com",100,"CN"),("B站点","b.com",200,"CN")]
cursor.executemany("INSERT INTO websites(name,url,alexa,country) VALUES (%s,%s,%s,%s)", data_list)
conn.commit()

连接优化

Web项目使用连接池 PooledDB，不要每次请求新建连接。
事务代码尽量精简，缩短事务执行时间，减少锁等待和死锁。
用完游标和连接及时close；连接池调用close()只是归还连接，不是断开。

八、游标类型说明

游标类型	作用
默认游标	返回元组 tuple
`DictCursor`	返回字典dict，可读性更强，开发常用
`SSCursor`	流式游标，处理超大结果集，避免一次性加载全部数据到内存