拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

python(8):用pymysql连接mysql数据库并获取数据的完整配置指南

python(8):用pymysql连接mysql数据库并获取数据的完整配置指南

1. 为什么本地开发总在 pymysql 连接 MySQL 这一步卡住

如果你正在写 Python 脚本,想把 MySQL 里的数据读出来做接口测试、数据清洗或者后台任务,那pymysql大概率是你绕不开的一个库。它是什么?简单说,pymysql是一个纯 Python 实现的 MySQL 客户端驱动,能让你在 Python 代码里直接执行 SQL、拿回结果,不需要额外装 C 扩展。适合谁?适合本地开发、测试环境、小规模数据读取,以及那些不想引入重型 ORM 的轻量场景。

但实际动手时,很多人会卡在几个很具体的地方:pip install pymysql装完了,代码也照着抄了,一跑却报Access denied、Can't connect to MySQL server,或者查询结果是一堆元组看不懂字段名。更麻烦的是,连接参数写错一个字符,报错信息还不一定直白。我试过在本地反复调host、port、charset,最后发现是 MySQL 8 的认证插件和旧写法不兼容。

这篇就按“能直接复制跑通”的思路来:先给一套可复制的连接配置,再写建表和查询示例,接着用真实请求验证结果,最后把常见报错逐条对照排查。你跟着走一遍,本地读取数据这条链路基本就通了。核心检索词就是 python、mysql、pymysql、数据库连接,全文围绕“怎么连、怎么读、怎么排错”展开,不绕弯子。

2. 前置准备:pymysql 安装与 TaoToken 接入配置

2.1 安装 pymysql 与本地 MySQL 检查

第一步永远是装库。打开终端执行:

pip install pymysql

如果你用的是虚拟环境,先激活再装,避免装到全局。装完后可以快速验证:

python -c "import pymysql; print(pymysql.__version__)"

能打印出版本号就说明装好了。接着确认本地 MySQL 服务在跑。Windows 可以在服务列表里看 MySQL 服务状态,macOS 或 Linux 用:

mysql -u root -p -e "select version();"

输入密码后能看到版本号,说明服务正常。这里有个容易忽略的点:MySQL 8 默认用caching_sha2_password认证插件,而部分旧版 pymysql 对它支持不完整,后面排错章节会专门讲。

2.2 TaoToken 前置:把模型能力接进你的开发流

写数据库脚本时,很多人会顺手让模型帮忙生成 SQL、解释报错或者补全字段映射。这时候一个稳定的模型接入入口就很有用。TaoToken 提供统一的 API 入口,你可以把它理解成“一个 Key 走通多种模型调用”,适合本地开发时做辅助。

接入需要三件套:Base URL、API Key、Model ID。Base URL 用https://taotoken.net/api,API Key 在控制台创建,Model ID 按你实际要用的模型填。创建 Key 的入口在这里:

https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=api_keys

如果你更习惯在对话界面里先验证模型是否可用,可以用模型对话入口:

https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=model_chat

需要说明的是,TaoToken 在这里的角色是“模型调用入口”,不是数据库代理,也不替代你的 MySQL 客户端。数据库连接仍然走 pymysql 直连本地 MySQL,两者互不干扰。把模型辅助和数据库读取分开看,思路会清晰很多。

3. 可复制配置:pymysql 连接参数与建表查询示例

3.1 连接参数配置片段

先给一份可以直接改的配置。建议单独放一个config.py或.env,别把密码硬编码在业务代码里。下面用 JSON 形式列出关键参数,路径和字段名保持通用:

{ "mysql": { "host": "127.0.0.1", "port": 3306, "user": "root", "password": "your_password", "database": "interface_test_db", "charset": "utf8mb4" } }

注意charset建议用utf8mb4,而不是老的utf-8。因为 MySQL 里的utf8实际只支持 3 字节,遇到 emoji 或部分生僻字会报错,utf8mb4才是完整的 4 字节实现。这个坑我在存中文备注时踩过,字段直接截断。

3.2 建表与插入测试数据

连接之前,先在 MySQL 里建一张测试表,方便后面验证读取:

CREATE DATABASE IF NOT EXISTS interface_test_db DEFAULT CHARACTER SET utf8mb4; USE interface_test_db; CREATE TABLE IF NOT EXISTS case_info ( id INT PRIMARY KEY AUTO_INCREMENT, case_name VARCHAR(100) NOT NULL, api_path VARCHAR(200), status TINYINT DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); INSERT INTO case_info (case_name, api_path, status) VALUES ('登录接口', '/api/login', 1), ('查询用户', '/api/user/query', 1), ('下单接口', '/api/order/create', 0);

执行完可以用select * from case_info;确认三条数据在。

3.3 完整查询代码

下面这段代码可以直接复制运行,把密码换成你自己的:

import pymysql from pymysql.cursors import DictCursor class MysqlUtils: def __init__(self): self.conn_obj = pymysql.connect( host="127.0.0.1", port=3306, user="root", password="your_password", database="interface_test_db", charset="utf8mb4", cursorclass=DictCursor, autocommit=True, ) self.cursor = self.conn_obj.cursor() def get_mysql_data_by_sql(self, sql_str): try: self.cursor.execute(sql_str) case_list = self.cursor.fetchall() return case_list finally: self.cursor.close() self.conn_obj.close() if __name__ == "__main__": sql_str = "select * from case_info;" result = MysqlUtils().get_mysql_data_by_sql(sql_str) for row in result: print(row)

和网上很多示例相比,这里做了两处调整:一是把cursorclass=DictCursor直接写在连接参数里,省去单独cursor(cursor=...)的写法;二是用try/finally保证游标和连接一定关闭,避免脚本跑多次后连接数堆积。实测下来,字典游标返回的结果带字段名,调试时比元组友好太多。

4. 验证请求:跑通查询并确认成功结果

4.1 执行脚本与预期输出

保存为mysql_demo.py,运行:

python mysql_demo.py

如果一切正常,你会看到类似这样的输出:

{'id': 1, 'case_name': '登录接口', 'api_path': '/api/login', 'status': 1, 'created_at': datetime.datetime(2024, 5, 20, 10, 30, 0)} {'id': 2, 'case_name': '查询用户', 'api_path': '/api/user/query', 'status': 1, 'created_at': datetime.datetime(2024, 5, 20, 10, 30, 0)} {'id': 3, 'case_name': '下单接口', 'api_path': '/api/order/create', 'status': 0, 'created_at': datetime.datetime(2024, 5, 20, 10, 30, 0)}

每条记录是一个字典,key 就是字段名。这说明连接、执行、取数、关闭四个环节都通了。

4.2 带条件查询与参数化写法

实际业务里很少select *,更多是带条件。这里要注意别用字符串拼接,容易引发 SQL 注入。正确写法是参数化:

def get_cases_by_status(self, status): sql = "select id, case_name, api_path from case_info where status = %s" self.cursor.execute(sql, (status,)) return self.cursor.fetchall()

调用get_cases_by_status(1)会返回 status 为 1 的两条记录。注意占位符是%s,不是?,这是 pymysql 的约定。传参时用元组,哪怕只有一个参数也要写成(status,),否则会报参数数量不匹配。

4.3 用模型辅助验证 SQL 语义

如果你对某条 SQL 的语义不确定,可以把 SQL 贴到模型对话里让它解释执行计划或字段含义。入口用:

https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=model_chat

比如你写了个多表 join,不确定返回字段,让模型先帮你梳理一遍,再回本地跑,能省不少试错时间。这一步是辅助,不改变数据库连接本身。

5. 常见报错排查:从 401 到 local proxy failed 逐条对照

5.1 Access denied 与 401 类认证错误

最常见的报错长这样:

pymysql.err.OperationalError: (1045, "Access denied for user 'root'@'localhost' (using password: YES)")

这通常有三个原因:密码错了、用户名不对、或者该用户没有从localhost连接的权限。先确认密码,再检查用户权限:

SELECT user, host, plugin FROM mysql.user WHERE user = 'root';

如果host是%但你想从本地连,一般没问题;如果只有特定 host,需要补授权。MySQL 8 还要看plugin字段,如果是caching_sha2_password而 pymysql 版本较老,可能报认证失败。解决办法是升级 pymysql:

pip install --upgrade pymysql

或者临时改认证方式(仅测试环境):

ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'your_password'; FLUSH PRIVILEGES;

5.2 Can't connect 与 local proxy failed

报错形如:

pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '127.0.0.1' ([WinError 10061] ...)")

或者在某些环境里出现local proxy failed字样。这类基本是网络层问题:MySQL 服务没启动、端口不对、或者防火墙拦了。排查顺序是:先ping 127.0.0.1确认本机回环正常,再用telnet 127.0.0.1 3306看端口通不通。Windows 上如果 telnet 没装,可以用 PowerShell:

Test-NetConnection -ComputerName 127.0.0.1 -Port 3306

TcpTestSucceeded为 True 才说明端口可达。如果服务没起,去服务管理器启动 MySQL;如果端口被改过,检查my.ini或my.cnf里的port配置。

5.3 reading choices 与结果解析异常

有时候连接成功、SQL 也执行了,但取数时报:

pymysql.err.ProgrammingError: reading choices failed

或者返回结果里字段名乱码。前者多半是 SQL 语法问题,比如表名写错、字段不存在,建议先把 SQL 拿到 MySQL 客户端里单独跑一遍。后者通常是charset没设对,把连接参数里的charset改成utf8mb4,并确认建库建表时也是utf8mb4。

5.4 OAuth 与 Codex auth.json 相关配置

如果你在用 Codex 这类工具做辅助开发,可能会遇到OAuth相关的认证提示。这类问题不在 pymysql 本身,而在模型调用侧的凭证配置。以 Codex 为例,它的凭证通常放在auth.json里,路径一般在用户目录下的配置文件夹中。需要确认三件套齐全:Base URL 填https://taotoken.net/api,API Key 填控制台创建的 Key,Model ID 填你实际调用的模型。三者缺一,或者 Base URL 带了多余路径,都会导致认证失败。

如果你用的是 Cline 或 CC Switch 这类工具,配置逻辑类似,同样要保证 Base URL、Key、Model ID 三件套一致。CC Switch 里切换配置时,注意别把数据库连接参数和模型调用参数混在同一个配置文件里,两者职责不同,混在一起排查起来很痛苦。

5.5 连接未关闭导致的资源耗尽

脚本跑多次后报Too many connections,说明连接没关。前面代码里的try/finally就是解决这个的。如果你用的是连接池,记得在程序退出时统一关闭。本地测试环境可以把max_connections调小来复现问题,但生产环境一定要用连接池管理。

6. 长期编码与 Agent 场景下的接入选择

如果你只是偶尔跑个查询脚本,上面这套 pymysql 直连就够了。但如果你在长期做接口测试、数据同步或者 Agent 类任务,频繁手动管理连接和模型调用会很累。这时候可以考虑把模型调用部分交给 Coding Plan 来统一管理,减少每次配置 Key 的重复劳动。

Coding Plan 的入口在这里:

https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=coding_plan

它的定位是给长期编码和 Agent 场景提供稳定的模型调用额度,和数据库连接是两条独立的链路。你可以理解为:pymysql 负责把数据从 MySQL 取出来,Coding Plan 负责让模型在编码过程中随时可用,两者配合,本地开发流就完整了。

接入文档在这里,遇到配置细节可以对照查:

https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc

最后给一个实用建议:把数据库连接参数和模型调用参数分别放在两个配置文件里,用环境变量区分。本地测试时,先确保python mysql_demo.py能稳定打印出三条记录,再去接模型辅助。顺序反了,排错会变成两件事混在一起,很难定位。数据库这条链路跑通后,后面加什么辅助工具都是锦上添花。

返回列表