1. 接口测试里数据库校验为什么总写成一团乱麻做接口测试的同学大概率都遇到过这种场景接口返回{code: 0, msg: success}但数据库里那条订单状态到底改没改、库存扣没扣对、用户余额是不是少了一块钱光看响应体根本判断不了。这时候就得连上 MySQL 手动查一把或者写几行 Python 脚本临时跑一下。问题是临时脚本写多了代码里到处都是pymysql.connect(...)、cursor.execute(...)、conn.close()一个用例里重复三遍改个库地址要全局搜索替换连接忘了关还会把连接池占满。我试过在一个 200 多条的接口自动化项目里光数据库连接代码就散落在十几个文件里后来统一封装成工具类之后用例里只剩一行assert db.check_user(张三)清爽很多。这篇就围绕 Python 接口测试中的数据库操作这条主线从 PyMySQL 最基础的连接和游标讲起一步步封装出可复用的查询、断言工具类最后落到一个真实动作拿接口响应里的字段去和库表数据做比对。PyMySQL 是什么它是一个纯 Python 实现的 MySQL 客户端库不需要装 MySQL 的 C 扩展pip install pymysql就能用适合在接口测试、自动化脚本里做数据准备、环境检查和数据库断言。它适合谁适合正在写接口自动化、想做数据层校验但还没找到顺手封装方式的测试同学也适合想把散落 SQL 收拢成工具类的开发。核心检索词先摆出来Python 接口测试、数据库操作、PyMySQL、游标封装、数据库断言。下面所有代码都是可以直接复制运行的库表结构我会给出来你照着建表就能跑通。先说清楚 PyMySQL 里两个最容易搞混的概念。连接connection代表一次和数据库的会话游标cursor是从连接上开出来的操作句柄类似文件句柄你只能通过游标去执行 SQL 和拿结果。cur.execute(sql)执行查询后返回的是影响行数不是查询结果结果要用fetchone()、fetchmany(n)、fetchall()去取。而且取完的数据会从结果集里删掉第二次fetchone()拿到的就是None这点在写断言时特别容易踩坑后面会专门讲。写操作insert/update/delete执行完不会立即生效必须conn.commit()提交注意是连接提交不是游标提交。出错时用conn.rollback()回滚。理解了这几个点封装起来就顺了。2. TaoToken 前置准备把模型对话和 API Key 配好再动手在正式写数据库封装之前有个前置环节值得先花十分钟搞定就是模型对话和 API Key 的准备。为什么放在这里因为接口测试里经常需要让模型帮你生成测试数据、解释报错、或者把一段 SQL 改写成参数化写法有个顺手的模型入口能省不少事。TaoToken 这边提供了模型对话、API Key 管理和接入文档几个入口我按实际用到的顺序说。第一步打开模型对话页面地址是 https://taotoken.net/api 这个入口可以直接和模型对话用来问 PyMySQL 的用法、让它帮你写建表语句、或者把一段有 SQL 注入风险的字符串拼接改成参数化写法都挺方便。比如你可以直接把sql select * from user where name{}.format(name)丢进去让它改成cur.execute(select * from user where name%s, (name,))省得自己查文档。第二步去 API Keys 页面生成一个 Key地址是 https://taotoken.net/api-keys 。生成之后复制保存好这个 Key 后面在配置里要用到。注意 Key 只显示一次丢了就重新生成。第三步如果你打算把模型能力接进自己的测试脚本或者编辑器里可以看接入文档地址是 https://taotoken.net/doc 里面有 Base URL、Key、Model ID 三件套的说明。Base URL 统一用 https://taotoken.net/api Key 就是上一步生成的Model ID 按文档里列的填。这三件套在 Cline、Codex、Claude Code 这类工具里配置时都要写全缺一个就连不上。第四步如果你是要长期做编码或者跑 Agent 任务可以看下 Coding Plan地址是 https://taotoken.net/coding-plan 适合需要持续调用模型的场景。控制台入口在 https://taotoken.net/console 可以看用量和余额。这里要提醒一句TaoToken 是模型调用和 API 管理的入口不是数据库工具也不替代你的编辑器或 IDE。数据库连接还是走 PyMySQL 直连你自己的 MySQL两者是配合关系模型帮你写和改代码PyMySQL 负责实际的数据操作。配置的时候有个细节Base URL 不要带多余的路径就用 https://taotoken.net/api Key 放在 Authorization 头里格式是Bearer 你的Key。Model ID 按文档填别自己猜。这三样配好之后你在写数据库封装时遇到报错可以直接把错误贴给模型对话让它帮你定位比翻文档快。我实测下来把模型对话和 API Key 准备好之后写这类工具类的效率会高一些尤其是排错阶段。下面进入正题开始写 PyMySQL 的连接和游标封装。3. 可复制配置PyMySQL 连接、游标上下文管理器与工具类封装这一节是全文的核心所有代码都可以直接复制。先给建表语句保证你跑得通。CREATE DATABASE IF NOT EXISTS api_test DEFAULT CHARSET utf8mb4; USE api_test; CREATE TABLE user ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50) NOT NULL, passwd VARCHAR(100) NOT NULL, balance DECIMAL(10,2) DEFAULT 0.00 ); INSERT INTO user (name, passwd, balance) VALUES (张三, 123456, 100.00), (李四, 123456, 200.00), (王五, 123456, 300.00);建好表之后先看最基础的连接和游标操作理解每一步在干什么。import pymysql # 1. 建立连接 conn pymysql.connect( host127.0.0.1, port3306, userroot, password123456, dbapi_test, charsetutf8mb4 # 有中文必须指定 ) # 2. 从连接建立游标 cur conn.cursor() # 3. 执行查询 cur.execute(select * from user where name%s, (张三,)) # 4. 获取结果 result cur.fetchall() print(result) # ((张三, 123456, 100.00),) # 5. 关闭游标和连接 cur.close() conn.close()注意这里用了%s占位符而不是字符串拼接这是防 SQL 注入的关键。很多同学图省事写select * from user where name{}.format(name)一旦 name 里带引号就出问题接口测试里参数往往来自外部必须参数化。接下来是游标上下文管理器。每次手动cur.close()、conn.close()很容易漏用with语句能自动管理资源。import pymysql from contextlib import contextmanager DB_CONFIG { host: 127.0.0.1, port: 3306, user: root, password: 123456, db: api_test, charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor # 返回字典方便按字段名取值 } contextmanager def get_cursor(commitFalse): conn pymysql.connect(**DB_CONFIG) cur conn.cursor() try: yield cur if commit: conn.commit() except Exception as e: conn.rollback() raise e finally: cur.close() conn.close()这个get_cursor就是游标上下文管理器的核心。commitFalse时用于查询commitTrue时用于写操作异常自动回滚资源自动释放。DictCursor让结果返回字典而不是元组断言时可以用row[balance]而不是row[2]可读性高很多。基于它封装查询和写操作def query_one(sql, argsNone): with get_cursor() as cur: cur.execute(sql, args) return cur.fetchone() def query_all(sql, argsNone): with get_cursor() as cur: cur.execute(sql, args) return cur.fetchall() def execute(sql, argsNone): with get_cursor(commitTrue) as cur: return cur.execute(sql, args)再往上封装业务级的断言工具类这是接口测试里最常用的部分class DBAssert: def __init__(self, configNone): self.config config or DB_CONFIG def get_user_balance(self, name): row query_one(select balance from user where name%s, (name,)) return row[balance] if row else None def assert_balance(self, name, expected): actual self.get_user_balance(name) assert actual expected, f用户{name}余额不符期望{expected}实际{actual} return True def assert_user_exists(self, name): row query_one(select id from user where name%s, (name,)) assert row is not None, f用户{name}不存在 return True def assert_user_not_exists(self, name): row query_one(select id from user where name%s, (name,)) assert row is None, f用户{name}不应存在 return True如果你想把连接信息放到配置文件里推荐用 TOML路径放在项目根目录的config/db.toml[mysql] host 127.0.0.1 port 3306 user root password 123456 db api_test charset utf8mb4读取用 Python 3.11 自带的tomllibimport tomllib with open(config/db.toml, rb) as f: DB_CONFIG tomllib.load(f)[mysql] DB_CONFIG[cursorclass] pymysql.cursors.DictCursor这样库地址变了只改配置文件代码不用动。到这里可复制的配置和封装就齐了下一节做实际验证。4. 验证请求接口响应与库表数据比对跑通封装写完必须验证不然不知道能不能用。这一节做一个完整的动作模拟一个接口返回拿响应里的字段去和数据库比对。先准备一个模拟的接口响应假设是查询用户余额的接口# 模拟接口返回 api_response { code: 0, msg: success, data: { name: 张三, balance: 100.00 } }然后写验证逻辑把接口返回的 balance 和数据库里的 balance 做比对from db_assert import DBAssert def test_user_balance_api(): db DBAssert() # 1. 调接口拿到响应这里用模拟数据代替 resp api_response assert resp[code] 0, 接口返回码异常 # 2. 从响应里取字段 name resp[data][name] api_balance resp[data][balance] # 3. 从数据库取对应字段 db_balance db.get_user_balance(name) # 4. 比对 assert api_balance db_balance, ( f接口余额{api_balance}与数据库余额{db_balance}不一致 ) print(f校验通过{name} 接口余额{api_balance} 数据库余额{db_balance}) test_user_balance_api()跑一下输出应该是校验通过张三 接口余额100.0 数据库余额100.00注意这里有个坑接口返回的100.00经过 JSON 序列化后可能变成100.0而数据库DECIMAL取出来是Decimal(100.00)。直接比较可能失败。稳妥的做法是统一转成字符串或Decimal再比from decimal import Decimal assert Decimal(str(api_balance)) Decimal(str(db_balance))再验证一个写操作的场景接口调用后数据库里应该新增一条记录。假设接口是注册用户def test_register_api(): db DBAssert() new_name 赵六 # 调接口前确认不存在 db.assert_user_not_exists(new_name) # 模拟接口调用实际项目里这里是 requests.post execute(insert into user (name, passwd, balance) values (%s, %s, %s), (new_name, 123456, 0.00)) # 调接口后确认存在 db.assert_user_exists(new_name) print(f注册校验通过{new_name} 已写入数据库) # 清理数据 execute(delete from user where name%s, (new_name,)) test_register_api()这个模式在接口测试里非常通用调接口前用assert_user_not_exists确认前置状态调接口后用assert_user_exists确认结果最后清理数据保证用例可重复执行。再补一个游标结果集被消耗的验证帮你理解为什么不能重复 fetchwith get_cursor() as cur: cur.execute(select * from user where name%s, (张三,)) first cur.fetchone() second cur.fetchone() print(first) # {id: 1, name: 张三, ...} print(second) # None所以需要重复使用结果时一定要先fetchall()赋给变量再基于变量做多次断言。这个坑我在实际项目里踩过用例里先fetchone()判断存在又fetchall()取数据结果第二次拿到空排查了半天。到这里查询校验、写操作校验、结果集消耗三个场景都跑通了封装是可用的。5. 本篇常见错排查401、local proxy failed、reading choices 与 OAuth这一节把实际会撞到的报错列出来对照着排查。数据库侧和模型侧的问题分开说。数据库侧最常见的几个pymysql.err.OperationalError: (1045, Access denied for user rootlocalhost)这是账号密码不对。检查DB_CONFIG里的 user 和 password注意 PyMySQL 的参数名是password不是passwd虽然有些版本兼容passwd但统一用password更稳。pymysql.err.OperationalError: (2003, Cant connect to MySQL server on 127.0.0.1)连不上检查 MySQL 服务是否启动、端口是否 3306、防火墙是否放行。如果是 Docker 里的 MySQLhost 不能写 127.0.0.1要写容器名或宿主 IP。pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)SQL 语法错。建议先在 MySQL 客户端手工跑一遍确认没问题再放进代码。参数化时占位符用%s不要用?PyMySQL 不认?。UnicodeDecodeError或中文乱码检查连接参数charsetutf8mb4以及建表时的字符集。模型侧和工具侧的报错如果你在 Cline、Codex、Claude Code 里配置了 TaoToken可能会遇到这几个401 UnauthorizedKey 不对或没带上。检查 Authorization 头是不是Bearer 你的KeyKey 有没有复制完整有没有多余空格。去 https://taotoken.net/api-keys 重新生成一个试试。local proxy failed本地代理配置有问题。检查 Base URL 是不是写成了 https://taotoken.net/api 不要带多余路径也不要在系统里配了额外的代理导致请求被拦截。reading choices相关报错通常是响应结构解析失败多半是 Model ID 填错了。去 https://taotoken.net/doc 对照文档里的 Model ID 列表确认填的是支持的模型。OAuth相关报错出现在 Claude Code 这类工具的登录环节。如果你用的是 API Key 方式接入就不需要走 OAuth检查工具配置里是不是误开了 OAuth 模式改成 API Key 模式填全 Base URL、Key、Model ID 三件套。这里把三件套再强调一遍Cline、Codex 的auth.json、Claude Code 配置里都要写全Base URL: https://taotoken.net/api Key: 你的 API Key Model ID: 按文档填写Codex 的auth.json里字段名按官方要求填Key 放在对应字段Base URL 不要漏。Cline 的 MCP 配置里同样三件套齐全。少任何一个都会报错而且报错信息不一定直白所以配的时候逐项核对。数据库侧还有一个隐蔽的坑连接没关导致Too many connections。用get_cursor上下文管理器就不会有这个问题finally里一定会close()。如果你还在手写conn.close()检查一下异常分支有没有漏。最后一个Decimal和float比较失败。前面提过统一转Decimal(str(x))再比别直接。6. 把数据库校验接进你的接口测试流程代码写到这里工具类已经能用了。接下来是怎么把它接进日常的接口测试流程让它真正省事。第一把db_assert.py放到项目的utils目录配置文件放config/db.toml用例里from utils.db_assert import DBAssert直接用。不要在每个用例文件里重复写连接代码。第二用例结构建议是「前置数据准备 → 调接口 → 数据库断言 → 数据清理」四段。前置用execute插入测试数据断言用DBAssert的方法清理用execute删除。这样用例可重复跑不会因为脏数据失败。第三断言方法按业务扩展。assert_balance、assert_user_exists只是示例你的项目里可能是assert_order_status、assert_stock、assert_points。封装思路一样查一条记录取字段和期望值比不一致就抛带上下文的AssertionError。第四参数化 SQL 一定要坚持。所有外部传入的值都走%s占位符不要用 format 或 f-string 拼。接口测试的参数往往来自用例数据拼字符串既有注入风险遇到引号还会直接报语法错。第五连接信息不要硬编码。用 TOML 或环境变量读取本地、测试、预发环境各一份配置切换环境只改配置不改代码。如果你在写用例时需要模型帮忙生成测试数据或改写 SQL模型对话入口在 https://taotoken.net/api 接入文档在 https://taotoken.net/doc 长期跑编码任务可以看 https://taotoken.net/coding-plan 。API Key 在 https://taotoken.net/api-keys 管理控制台在 https://taotoken.net/console 。最后说个实用技巧把DBAssert的断言失败信息写详细带上期望值、实际值、SQL 和参数。接口测试失败时日志里一眼能看出是数据没写对还是接口返回错比只抛一个AssertionError强太多。我现在的做法是在assert_balance里把name、expected、actual都拼进错误信息排查时间能省一半。数据库操作有风险写操作记得加清理别在生产库上跑。测试库随便折腾生产库只读不写。