1. 为什么你的 pymysql 一跑就报错：先看清 connect 到 commit 的链路
很多 Python 开发者第一次用 pymysql 连数据库，代码看着没问题，运行起来却报Access denied、Unknown database，或者更隐蔽的——数据明明插进去了，重新查询却什么都没有。问题往往不在 SQL 本身，而在于没搞清楚 pymysql 的四个核心环节：connect 建立连接、cursor 创建游标、execute 执行语句、commit 提交事务。
pymysql 是纯 Python 实现的 MySQL 客户端库，不需要编译 C 扩展，pip install pymysql就能用。它适合谁？适合做数据采集落库、后台脚本、Flask/Django 之外的轻量数据操作，以及需要快速验证 SQL 逻辑的场景。你可以把它理解成一座桥：Python 程序在桥这头，MySQL 服务在桥那头，connect 是搭桥，cursor 是桥上的传送带，execute 是把货物放上传送带，commit 则是按下“确认收货”——不按，货就一直在半路。
我试过在本地起一个 MySQL 8.0，用 pymysql 走一遍建表、插入、查询、更新、删除的完整闭环，中间踩过字符集和事务的坑。下面把可复制的配置骨架和验证动作拆开讲，你跟着敲一遍就能独立跑通。
2. 前置准备：装库、建库、确认连接四要素
在写代码之前，先把环境理清楚。pymysql 的安装很简单：
pip install pymysql如果你用的是虚拟环境，先激活再装。装完可以用pip show pymysql确认版本，一般 1.1.x 就够用。
接着确认 MySQL 服务在跑，并且你有一个可用的库。这里假设你已经建好一个名为ss的数据库。如果没有，用命令行或客户端执行：
CREATE DATABASE ss DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;字符集建议统一用utf8mb4，否则中文和 emoji 容易出问题。连接四要素必须写对：host、user、password、database。密码字段在 pymysql 里既可以写password也可以写passwd，两者等价，但别拼错。端口默认 3306，如果改过要显式指定port。
注意：连接信息不要硬编码在业务代码里，正式项目建议走环境变量或配置文件。本文为了演示直观，直接写在代码中。
3. 可复制配置：connect、cursor、execute、commit 四步骨架
先给一个最小可运行骨架，把四个环节串起来。这段代码的作用是连接数据库、查询 MySQL 版本，验证链路是否通：
import pymysql # 第一步：connect 建立连接 conn = pymysql.connect( host='localhost', user='root', password='123456', database='ss', port=3306, charset='utf8mb4' ) try: # 第二步：cursor 创建游标 cursor = conn.cursor() # 第三步：execute 执行 SQL cursor.execute('SELECT VERSION()') # 取回单条结果 data = cursor.fetchone() print('MySQL 版本:', data) finally: # 第四步：关闭游标和连接 cursor.close() conn.close()运行后如果打印出类似MySQL 版本: ('8.0.36',)，说明 connect、cursor、execute 三个环节都通了。这里没有写 commit，因为查询不改变数据状态，不需要提交。
接下来建一张表，把 commit 加进来。表名和字段名建议用反引号包裹，避免和关键字冲突：
import pymysql conn = pymysql.connect( host='localhost', user='root', password='123456', database='ss', charset='utf8mb4' ) try: cursor = conn.cursor() sql = ''' CREATE TABLE IF NOT EXISTS `tast` ( `first_name` VARCHAR(10) PRIMARY KEY, `last_name` VARCHAR(10), `age` INT(5), `sex` VARCHAR(6), `income` INT(30) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ''' cursor.execute(sql) # 关键：DDL 之后也要 commit，否则部分场景下不生效 conn.commit() print('建表成功') except Exception as e: conn.rollback() print('建表失败:', e) finally: cursor.close() conn.close()这里有个容易忽略的点：CREATE TABLE属于 DDL，在 MySQL 里默认是自动提交的，但显式写conn.commit()更稳妥，也养成习惯。异常捕获里加conn.rollback()，出错时回滚，避免留下半成品状态。
4. 增删改查全流程：executemany 批量插入与事务提交验证
建好表后，先插入数据。单条插入用execute，多条插入用executemany，后者效率更高：
import pymysql conn = pymysql.connect( host='localhost', user='root', password='123456', database='ss', charset='utf8mb4' ) try: cursor = conn.cursor() sql = ''' INSERT INTO `tast` (`first_name`, `last_name`, `age`, `sex`, `income`) VALUES (%s, %s, %s, %s, %s) ''' params = ( ('a', 'joe', 21, 'm', 5000), ('b', 'jae', 22, 'f', 5000), ('c', 'jbe', 23, 'f', 5000), ('d', 'jvce', 24, 'm', 5000), ) rows = cursor.executemany(sql, params) conn.commit() print('影响行数:', rows) except Exception as e: conn.rollback() print('插入失败:', e) finally: cursor.close() conn.close()executemany返回的是受影响的行数，这里应该是 4。注意参数用%s占位，不要用字符串拼接，否则会有 SQL 注入风险。
插入完成后，验证查询：
import pymysql conn = pymysql.connect( host='localhost', user='root', password='123456', database='ss', charset='utf8mb4' ) try: cursor = conn.cursor() cursor.execute('SELECT `first_name`, `age`, `income` FROM `tast` WHERE `sex` = %s', ('f',)) results = cursor.fetchall() for row in results: print(row) finally: cursor.close() conn.close()fetchall()返回所有结果，fetchone()返回单条。如果查询结果为空，fetchone()返回None，别直接下标取值。
更新和删除同样遵循 execute + commit 的模式：
# 更新 cursor.execute('UPDATE `tast` SET `income` = %s WHERE `first_name` = %s', (8000, 'a')) conn.commit() # 删除 cursor.execute('DELETE FROM `tast` WHERE `first_name` = %s', ('d',)) conn.commit()每次改动数据后都要 commit，否则当前连接关闭时改动会丢失。你可以做个实验：把 commit 注释掉，插入数据后关闭连接，再重新查询，会发现数据没进去——这就是事务未提交的典型表现。
5. 验证请求与成功结果：用一条查询确认闭环
完整的验证动作是这样：插入数据后，不关闭连接，直接在同一连接内查询，确认数据可见；然后关闭连接，重新建立连接再查一次，确认数据已持久化。
import pymysql def get_conn(): return pymysql.connect( host='localhost', user='root', password='123456', database='ss', charset='utf8mb4' ) # 第一次连接：插入并查询 conn = get_conn() cursor = conn.cursor() cursor.execute("INSERT INTO `tast` VALUES ('e', 'test', 30, 'm', 9000)") conn.commit() cursor.execute("SELECT COUNT(*) FROM `tast`") print('当前连接内行数:', cursor.fetchone()) cursor.close() conn.close() # 第二次连接：重新查询，验证持久化 conn2 = get_conn() cursor2 = conn2.cursor() cursor2.execute("SELECT COUNT(*) FROM `tast`") print('新连接行数:', cursor2.fetchone()) cursor2.close() conn2.close()两次打印的行数一致，说明 commit 生效、数据已落盘。如果第二次查不到，八成是 commit 漏了，或者连接用了自动提交关闭的配置。
6. 本篇常见错排查：从报错信息定位到具体环节
下面这张表把高频报错和对应环节列出来，方便你对号入座：
| 报错信息 | 可能原因 | 排查动作 |
|---|---|---|
Access denied for user | 用户名或密码错 | 核对 connect 的 user/password |
Unknown database 'ss' | 库不存在或名字拼错 | 执行SHOW DATABASES确认 |
Can't connect to MySQL server | 服务没起或 host/port 错 | 检查 MySQL 进程和端口 |
| 插入后查不到数据 | 漏了 commit | 在 execute 后加 conn.commit() |
| 中文乱码 | 字符集不统一 | connect 加 charset='utf8mb4'，建库建表也用 utf8mb4 |
Commands out of sync | 上一个结果集没读完 | 用 fetchall 读完或换新 cursor |
还有一个隐蔽的坑：cursor和conn忘记关闭，长时间运行会耗尽连接数。建议用try/finally或上下文管理器。pymysql 的连接对象支持with语法，但游标需要手动管理，写法如下：
with pymysql.connect(host='localhost', user='root', password='123456', database='ss', charset='utf8mb4') as conn: with conn.cursor() as cursor: cursor.execute('SELECT 1') print(cursor.fetchone())with块结束时会自动 commit（如果没异常）或 rollback（如果有异常），并关闭连接。这个写法能省掉不少样板代码。
如果你在排查连接或权限问题时需要更细的日志，可以在 connect 里加cursorclass=pymysql.cursors.DictCursor，让结果以字典返回，字段名一目了然，调试时更直观。
7. 把连接配置沉淀成可复用模块
跑通一次闭环后，建议把连接逻辑抽成函数或类，避免每次重复写四要素。一个简单的做法是用字典存配置，从环境变量读取：
import os import pymysql def get_connection(): return pymysql.connect( host=os.getenv('DB_HOST', 'localhost'), user=os.getenv('DB_USER', 'root'), password=os.getenv('DB_PASSWORD', '123456'), database=os.getenv('DB_NAME', 'ss'), charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor )这样本地开发和线上部署只需改环境变量，代码不用动。配合with语法，一个完整的读写操作可以压缩到几行：
with get_connection() as conn: with conn.cursor() as cursor: cursor.execute('SELECT * FROM `tast` WHERE `age` > %s', (20,)) for row in cursor.fetchall(): print(row['first_name'], row['income'])到这里，connect、cursor、execute、commit 四个环节你已经完整走了一遍。接下来可以试着把 SQL 换成自己的业务表，或者接入定时任务做批量落库。如果后续要接大模型做数据问答、或者把数据库操作封装成 Agent 工具，可以到 TaoToken 的模型对话页面先验证模型对 SQL 的理解能力，再决定怎么把 pymysql 的读写链路嵌进你的编码工作流。
