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( hostlocalhost, userroot, password123456, databasess, port3306, charsetutf8mb4 ) 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( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 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( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) 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( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) 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( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) # 第一次连接插入并查询 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/passwordUnknown database ss库不存在或名字拼错执行SHOW DATABASES确认Cant connect to MySQL server服务没起或 host/port 错检查 MySQL 进程和端口插入后查不到数据漏了 commit在 execute 后加 conn.commit()中文乱码字符集不统一connect 加 charsetutf8mb4建库建表也用 utf8mb4Commands out of sync上一个结果集没读完用 fetchall 读完或换新 cursor还有一个隐蔽的坑cursor和conn忘记关闭长时间运行会耗尽连接数。建议用try/finally或上下文管理器。pymysql 的连接对象支持with语法但游标需要手动管理写法如下with pymysql.connect(hostlocalhost, userroot, password123456, databasess, charsetutf8mb4) as conn: with conn.cursor() as cursor: cursor.execute(SELECT 1) print(cursor.fetchone())with块结束时会自动 commit如果没异常或 rollback如果有异常并关闭连接。这个写法能省掉不少样板代码。如果你在排查连接或权限问题时需要更细的日志可以在 connect 里加cursorclasspymysql.cursors.DictCursor让结果以字典返回字段名一目了然调试时更直观。7. 把连接配置沉淀成可复用模块跑通一次闭环后建议把连接逻辑抽成函数或类避免每次重复写四要素。一个简单的做法是用字典存配置从环境变量读取import os import pymysql def get_connection(): return pymysql.connect( hostos.getenv(DB_HOST, localhost), useros.getenv(DB_USER, root), passwordos.getenv(DB_PASSWORD, 123456), databaseos.getenv(DB_NAME, ss), charsetutf8mb4, cursorclasspymysql.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 的读写链路嵌进你的编码工作流。 用 pymysql 连接数据库：从 connect 到 commit 的完整操作指南
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( hostlocalhost, userroot, password123456, databasess, port3306, charsetutf8mb4 ) 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( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) 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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 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( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) 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( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) 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( hostlocalhost, userroot, password123456, databasess, charsetutf8mb4 ) # 第一次连接插入并查询 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/passwordUnknown database ss库不存在或名字拼错执行SHOW DATABASES确认Cant connect to MySQL server服务没起或 host/port 错检查 MySQL 进程和端口插入后查不到数据漏了 commit在 execute 后加 conn.commit()中文乱码字符集不统一connect 加 charsetutf8mb4建库建表也用 utf8mb4Commands out of sync上一个结果集没读完用 fetchall 读完或换新 cursor还有一个隐蔽的坑cursor和conn忘记关闭长时间运行会耗尽连接数。建议用try/finally或上下文管理器。pymysql 的连接对象支持with语法但游标需要手动管理写法如下with pymysql.connect(hostlocalhost, userroot, password123456, databasess, charsetutf8mb4) as conn: with conn.cursor() as cursor: cursor.execute(SELECT 1) print(cursor.fetchone())with块结束时会自动 commit如果没异常或 rollback如果有异常并关闭连接。这个写法能省掉不少样板代码。如果你在排查连接或权限问题时需要更细的日志可以在 connect 里加cursorclasspymysql.cursors.DictCursor让结果以字典返回字段名一目了然调试时更直观。7. 把连接配置沉淀成可复用模块跑通一次闭环后建议把连接逻辑抽成函数或类避免每次重复写四要素。一个简单的做法是用字典存配置从环境变量读取import os import pymysql def get_connection(): return pymysql.connect( hostos.getenv(DB_HOST, localhost), useros.getenv(DB_USER, root), passwordos.getenv(DB_PASSWORD, 123456), databaseos.getenv(DB_NAME, ss), charsetutf8mb4, cursorclasspymysql.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 的读写链路嵌进你的编码工作流。