1. Python 脚本连 MySQL 总踩坑?先理清连接、插入、更新这条链路
Python 操作 MySQL 这件事,说简单也简单,connect一下就能跑;说坑也多,编码、占位符、事务提交、连接泄漏,随便一个都能让你在本地调试半小时。这篇聚焦三件最基础也最高频的动作:连接、插入、更新,面向本地开发与数据入库场景,给你能直接复制的参数模板和示例代码。
同时我会把脚本里可能出现的 API 调用统一走 TaoToken,用一个 Key 管住模型调用,避免本地脚本里散落一堆 Key 和 Base URL。TaoToken 是一个 API 聚合平台,能做什么?简单说就是把模型调用收敛到一个入口,适合谁?适合本地写脚本、做数据入库、顺手要调模型的开发者。你不需要在多个平台之间来回切换配置,一个 Key 就能打通脚本调试链路。
先明确本文的检索核心:Python 连接 MySQL、Python 插入数据、Python 更新数据,以及用统一 Key 管理脚本里的 API 调用。下面从环境准备讲到完整读写验证,每一步都能跟做。
我试过在本地用MySQLdb和pymysql两套驱动来回切,最后发现新手最容易卡在三个地方:一是连接参数写错端口或字符集,二是插入时用字符串拼接导致 SQL 注入或引号报错,三是更新后忘了commit以为没生效。这篇会把这三个坑都填上。
环境上你只需要:本地或局域网有一台 MySQL(5.7/8.0 都行),Python 3.8+,装好驱动。驱动选择上,MySQLdb是老牌驱动,很多老项目在用;pymysql是纯 Python 实现,安装省事,兼容性也好。本文示例以pymysql为主,同时给出MySQLdb的等价写法,方便你对照老代码。
先建一张测试表,后面插入和更新都用它:
CREATE TABLE user_process_info ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(64) NOT NULL, step1_num INT DEFAULT 0, step2_num INT DEFAULT 0, step3_num INT DEFAULT 0, email VARCHAR(128) DEFAULT NULL, login_state TINYINT DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;这张表覆盖了插入需要的数值字段和更新需要的状态字段,够用了。注意字符集用utf8mb4,别用utf8,否则遇到 emoji 或生僻字会报错,这是很多人踩过的坑。
连接参数模板先给你,直接改 host、user、password、db 四个值就能用:
DB_CONFIG = { "host": "127.0.0.1", "port": 3306, "user": "root", "password": "your_password", "db": "test_db", "charset": "utf8mb4", "cursorclass": "DictCursor", # pymysql 用 }这里charset一定要和建表时一致,port默认 3306,如果你本地改过端口要同步。cursorclass用字典游标,查询结果直接是 dict,比元组好读。
2. TaoToken 前置准备:一个 Key 管住脚本里的模型调用
为什么 Python 操作 MySQL 的脚本里会涉及模型调用?因为很多数据入库场景不是纯搬运,比如你要对入库的文本做摘要、分类、字段抽取,或者调试阶段让模型帮你生成测试数据。这些调用如果每个平台一套 Key,脚本里就会散落各种配置,换环境时特别容易漏。
TaoToken 在这里的角色是统一入口。你只需要在平台拿到一个 API Key,配置好 Base URL,脚本里所有模型调用都指向它。这样本地调试、换机器、交接给别人,都只改一个 Key 的事。
前置准备分三步。第一步,注册并登录 TaoToken 控制台,地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,进去后在控制台里创建 API Key。第二步,记下你的 Base URL,API 入口是 https://taotoken.net/api ,注意这个地址不带查询参数,配置时直接用它。第三步,确认你要用的模型 ID,比如做文本处理常用的对话模型,具体 ID 在模型对话页面能看到。
这里有个关键点:Base URL 和 Key 是两件事,别混。Base URL 是请求地址前缀,Key 是身份凭证。很多新手把完整请求地址当成 Base URL 填进去,结果报 404。正确做法是 Base URL 只填到/api,具体路径由 SDK 或请求库拼接。
如果你用的是 OpenAI 兼容的 SDK,配置大概长这样:
from openai import OpenAI client = OpenAI( api_key="你的_TaoToken_Key", base_url="https://taotoken.net/api", ) resp = client.chat.completions.create( model="你的模型ID", messages=[{"role": "user", "content": "把这句话转成 JSON:张三 28 岁"}], ) print(resp.choices[0].message.content)这段代码里三个要素齐全:Base URL、Key、Model ID。缺一个都跑不通。Model ID 填错会报模型不存在,Key 填错会报 401,Base URL 填错会报连接失败或 404。后面排障章节会逐个对照。
如果你更习惯用环境变量管理 Key,可以这样:
export TAOTOKEN_API_KEY="你的_TaoToken_Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"然后脚本里用os.environ读取。这样 Key 不会硬编码进代码,提交到仓库也安全。本地开发强烈建议这么做,尤其是多人协作的项目。
拿到 Key 之后,先别急着写业务逻辑,用一次最简单的模型对话验证连通性。打开模型对话页面 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,发一句测试消息,能正常返回就说明 Key 和 Base URL 没问题。这一步花不了一分钟,但能帮你排除掉后面一半的配置类报错。
3. 可复制配置:连接、插入、更新三段代码直接跑
这一节给你三段能直接复制的代码,分别对应连接、插入、更新。每段都标了关键参数和容易改错的地方。
先看连接。用pymysql的写法:
import pymysql def get_conn(): return pymysql.connect( host="127.0.0.1", port=3306, user="root", password="your_password", database="test_db", charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=False, )注意database这个参数,MySQLdb里叫db,pymysql里两个都认,但建议统一用database。autocommit=False是默认值,意味着你需要手动commit,这是插入和更新必须记住的点。
如果你用的是老项目里的MySQLdb,等价写法是:
import MySQLdb conn = MySQLdb.connect( host="127.0.0.1", user="root", passwd="your_password", db="test_db", charset="utf8mb4", ) cur = conn.cursor() cur.execute("SET NAMES utf8mb4") conn.commit()MySQLdb里密码参数叫passwd不是password,这是老代码里最常见的改错点。另外SET NAMES那行是显式设置连接字符集,配合charset参数双保险。
再看插入。推荐用参数化占位符,别用字符串拼接:
def insert_process_info(username, step1, step2, step3): conn = get_conn() try: with conn.cursor() as cur: sql = ( "INSERT INTO user_process_info " "(username, step1_num, step2_num, step3_num) " "VALUES (%s, %s, %s, %s)" ) cur.execute(sql, (username, step1, step2, step3)) conn.commit() return cur.lastrowid except Exception as e: conn.rollback() raise e finally: conn.close()这里%s是占位符,不管字段是字符串还是数字都用%s,驱动会帮你处理类型。参数用元组传入,顺序和 SQL 里的字段一一对应。lastrowid能拿到刚插入的自增 ID,入库场景经常需要。
对比一下老代码里的字符串拼接写法:
sql = "insert into usercount_processinfo (time,username,step1num) values('%s','%s',%d)" % (time, username, step1num)这种写法有两个问题:一是字符串字段要手动加引号,漏了或多了都报错;二是如果 username 里带单引号,SQL 直接崩,还有注入风险。参数化写法把这两个问题都解决了,强烈建议换掉。
最后看更新。更新最容易忘commit:
def update_login_state(email, password): conn = get_conn() try: with conn.cursor() as cur: sql = ( "UPDATE user_process_info " "SET login_state = %s, email = %s " "WHERE email = %s" ) affected = cur.execute(sql, (-1, password, email)) conn.commit() return affected except Exception as e: conn.rollback() raise e finally: conn.close()execute的返回值是受影响行数,可以用来判断更新是否命中。如果返回 0,说明WHERE条件没匹配到记录,不是代码错了,是数据不存在。这个区分很重要,很多人看到 0 以为更新失败,其实是没这行数据。
三段代码的共同点是:用try/except/finally包住,异常时rollback,最后close。连接泄漏在本地调试时不容易发现,但脚本跑久了会耗尽连接数,报Too many connections。
4. 验证请求:一次完整读写跑通并看到结果
配置写好了,得跑一次完整链路验证。这一节给你一个端到端脚本,插入一条记录、更新它、再查出来确认。
import pymysql def get_conn(): return pymysql.connect( host="127.0.0.1", port=3306, user="root", password="your_password", database="test_db", charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) def main(): conn = get_conn() try: with conn.cursor() as cur: # 1. 插入 insert_sql = ( "INSERT INTO user_process_info " "(username, step1_num, step2_num, step3_num, email) " "VALUES (%s, %s, %s, %s, %s)" ) cur.execute(insert_sql, ("zhangsan", 1, 2, 3, "zhangsan@test.com")) new_id = cur.lastrowid conn.commit() print(f"插入成功,id={new_id}") # 2. 更新 update_sql = ( "UPDATE user_process_info " "SET login_state = %s, step3_num = %s " "WHERE id = %s" ) affected = cur.execute(update_sql, (1, 99, new_id)) conn.commit() print(f"更新影响行数={affected}") # 3. 查询确认 cur.execute( "SELECT id, username, step3_num, login_state, email " "FROM user_process_info WHERE id = %s", (new_id,), ) row = cur.fetchone() print("查询结果:", row) except Exception as e: conn.rollback() print("出错:", e) finally: conn.close() if __name__ == "__main__": main()跑起来预期输出:
插入成功,id=1 更新影响行数=1 查询结果: {'id': 1, 'username': 'zhangsan', 'step3_num': 99, 'login_state': 1, 'email': 'zhangsan@test.com'}看到step3_num从 3 变成 99、login_state变成 1,说明插入和更新都生效了。如果查询结果里step3_num还是 3,八成是更新后没commit,或者WHERE条件没匹配上。
再验证一下模型调用链路。在同一个脚本里加一段,把查询结果丢给模型做格式化:
import os from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url="https://taotoken.net/api", ) def summarize(row): resp = client.chat.completions.create( model="你的模型ID", messages=[ {"role": "system", "content": "把用户数据整理成一句话"}, {"role": "user", "content": str(row)}, ], ) return resp.choices[0].message.content print(summarize(row))能正常返回文本,说明 MySQL 读写和模型调用两条链路都通了。这一步的意义在于:你的脚本调试链路现在是统一的,数据库操作和 API 调用都在一个环境里,换机器只需要改数据库连接和 TaoToken Key 两处配置。
如果你要长期跑编码类任务或 Agent 场景,可以考虑 Coding Plan,地址是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,适合需要持续调用模型的场景。只是偶尔验证模型的话,用模型对话页面就够了。
5. 本篇常见错排查:401、连接失败、更新不生效逐个对照
这一节把最常见的几类报错列出来,对照你的实际报错找原因。
第一类,认证类报错。典型信息是401 Unauthorized或invalid api key。原因通常是 Key 填错、Key 过期、或者 Key 前后带了空格。排查方法:把 Key 复制到模型对话页面测一下,能通说明 Key 没问题,问题在脚本里的读取方式。如果你用环境变量,确认export之后新开的终端才生效,老终端读不到。
第二类,连接类报错。典型信息是Can't connect to MySQL server或local proxy failed。前者检查 host、port、MySQL 服务是否启动;后者如果出现在 API 调用里,通常是 Base URL 填错,比如填成了带具体路径的完整地址。正确做法是 Base URL 只填https://taotoken.net/api,不要在后面加/v1/chat/completions之类。
第三类,模型相关报错。典型信息是model not found或读取choices时报KeyError。前者是 Model ID 填错,去模型对话页面确认准确的 ID;后者通常是请求没成功,返回体里没有choices字段,先打印完整响应看看结构。别直接resp.choices[0],先确认resp里有东西。
第四类,更新不生效。现象是执行了UPDATE但查询结果没变。三个可能:一是没commit,二是WHERE条件没匹配到记录(affected返回 0),三是连到了不同的数据库或表。排查顺序:先看affected返回值,再看commit有没有执行,最后确认连接参数里的database是不是你查的那个库。
第五类,编码报错。典型信息是Incorrect string value或中文变问号。原因是连接字符集和表字符集不一致。解决方法是连接参数里charset="utf8mb4",建表时也用utf8mb4,两边对齐。老代码里的SET NAMES UTF8建议改成SET NAMES utf8mb4。
第六类,连接泄漏。现象是脚本跑一段时间后报Too many connections。原因是连接没关。检查每个connect是否都有对应的close,推荐用try/finally或上下文管理器。with conn.cursor()只关游标,不关连接,连接还得单独close。
对照表方便你快速定位:
| 报错关键词 | 大概率原因 | 排查动作 |
|---|---|---|
| 401 / invalid api key | Key 错误或过期 | 去模型对话页面验证 Key |
| Can't connect to MySQL | host/port/服务问题 | 检查 MySQL 是否启动、端口是否对 |
| local proxy failed | Base URL 填错 | 只填 https://taotoken.net/api |
| model not found | Model ID 错误 | 去模型对话页面确认 ID |
| KeyError: 'choices' | 请求失败无 choices | 打印完整响应体 |
| 更新影响行数=0 | WHERE 没匹配 | 检查条件字段值 |
| Incorrect string value | 字符集不一致 | 连接和表都用 utf8mb4 |
| Too many connections | 连接未关闭 | 补 close,用 try/finally |
排障时如果涉及接入配置,可以去接入文档页面 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 对照参数说明;需要重新生成 Key 就去 API Keys 页面 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。这两个页面配合用,基本能覆盖配置类问题。
6. 把 Key 配置和数据库操作收进一个调试脚本
最后给你一个把两件事收在一起的实用技巧:写一个debug_env.py,启动时自检数据库连接和 TaoToken 连通性,任何一项不通就直接报出来,省得业务逻辑跑到一半才崩。
import os import pymysql from openai import OpenAI def check_mysql(): try: conn = pymysql.connect( host="127.0.0.1", port=3306, user="root", password="your_password", database="test_db", charset="utf8mb4", ) conn.close() print("[OK] MySQL 连接正常") return True except Exception as e: print(f"[FAIL] MySQL: {e}") return False def check_taotoken(): try: client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url="https://taotoken.net/api", ) resp = client.chat.completions.create( model="你的模型ID", messages=[{"role": "user", "content": "ping"}], max_tokens=5, ) print("[OK] TaoToken 连通,返回:", resp.choices[0].message.content) return True except Exception as e: print(f"[FAIL] TaoToken: {e}") return False if __name__ == "__main__": ok1 = check_mysql() ok2 = check_taotoken() print("环境自检:", "通过" if ok1 and ok2 else "有问题,先修上面报错")这个脚本的好处是把两类配置问题前置暴露。数据库连不上和 Key 配错是本地调试最常见的两类阻塞,自检一遍能省很多来回。跑通之后再去写业务逻辑,心里有底。
日常使用上,数据库连接建议抽成一个get_conn()函数,别在每个函数里重复写连接参数,改端口或密码时只改一处。TaoToken 的 Key 用环境变量管理,别硬编码。模型 ID 也抽成常量,换模型时只改一个地方。
如果你后续要做更复杂的入库流程,比如批量插入用executemany、更新用ON DUPLICATE KEY UPDATE,思路是一样的:参数化、显式 commit、异常回滚、连接关闭。把这四件事做扎实,Python 操作 MySQL 的脚本就稳了。