1. 为什么 Connection 和 cursor 总让人绕晕如果你写过 Python 数据库代码大概率见过这样的结构先connect()拿到一个 Connection再从它身上cursor()出一个游标然后execute()、fetchall()最后commit()和close()。看起来就四五个动作但真到项目里问题往往出在“谁负责关、谁负责提交、谁负责重连”这三件事上。Connection 可以理解成一条到数据库的“会话通道”它管连接、事务、字符集、超时cursor 则是这条通道上跑 SQL 的“执行手柄”它管语句、参数、结果集。两者职责不同但生命周期是绑定的cursor 由 Connection 创建Connection 一关cursor 全部失效。很多人把 cursor 当全局变量传来传去或者在 with 块里提前把 Connection 关了结果就是InterfaceError: cursor closed或者OperationalError: (0, )。这篇聚焦 Python 数据库编程里 Connection 与 cursor 的核心用法同时结合 TaoToken 的统一 Key/API 通道给出一套可复制的config.toml与settings.json配置骨架再演示连接建立、游标执行、结果获取、关闭的完整验证动作。适合正在搭数据库访问层、或者想把散落的连接参数收拢到统一配置里的开发者。下面所有代码都可以直接跑改掉主机和库名即可。2. TaoToken 统一 Key 的前置准备在写数据库访问层之前先把“凭证从哪来”这件事定下来。我习惯把模型调用和数据库访问的密钥都收拢到 TaoToken 的统一 Key 体系里这样本地开发、CI、容器环境用的是同一套引用方式不用在代码里硬编码。你需要先拿到一个可用的 API Key。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台创建 Key控制台入口是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite Key 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。创建完复制那串sk-开头的字符串后面配置里会用到。这里要区分两件事TaoToken 的 Key 用于访问模型/API 通道数据库的账号密码用于访问你的 MySQL。两者不要混在一个字段里。我的做法是让config.toml管数据库连接参数settings.json管 TaoToken 的 Key 和 API 地址代码里通过环境变量覆盖敏感值。API 基础地址用 https://taotoken.net/api 注意这个地址不带查询参数直接作为 base_url 使用。如果你后面要做长期编码或 Agent 类任务可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入细节看文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite Claude Code 相关配置在 https://taotoken.net/ClaudeCodeAnthropic?utm_sourcetaotoken_aicg_blog_endutm_contentClaudeCodeAnthropicutm_campaignrewrite 。这些先放着本篇主线还是数据库。3. config.toml 与 settings.json 配置骨架先建目录结构保持配置和代码分离project/ ├── config/ │ ├── config.toml │ └── settings.json ├── db/ │ ├── __init__.py │ └── pool.py └── main.pyconfig/config.toml负责数据库连接参数。用 TOML 是因为它支持注释、层级清晰比 ini 好读# config/config.toml [mysql] host 127.0.0.1 port 3306 user app_user password change_me database demo_db charset utf8mb4 [mysql.pool] min_size 2 max_size 10 connect_timeout 10 read_timeout 30 write_timeout 30 [mysql.cursor] # 默认游标类可选 DictCursor 让结果变成字典 cursorclass DictCursorconfig/settings.json负责 TaoToken 通道参数。JSON 适合被各种语言和工具读取CI 里注入也方便{ taotoken: { base_url: https://taotoken.net/api, api_key: sk-your-key-here, timeout: 60 }, app: { env: dev, log_level: INFO } }敏感值不要写死在文件里。用环境变量覆盖代码里读os.environ.get(MYSQL_PASSWORD, cfg[mysql][password])这种形式。这样本地用文件、线上用环境变量一套代码两种环境都能跑。读取配置的辅助模块# db/config_loader.py import json import os import tomllib from pathlib import Path BASE Path(__file__).resolve().parent.parent / config def load_toml() - dict: with open(BASE / config.toml, rb) as f: cfg tomllib.load(f) # 环境变量优先 cfg[mysql][password] os.environ.get( MYSQL_PASSWORD, cfg[mysql][password] ) cfg[mysql][host] os.environ.get( MYSQL_HOST, cfg[mysql][host] ) return cfg def load_settings() - dict: with open(BASE / settings.json, r, encodingutf-8) as f: s json.load(f) s[taotoken][api_key] os.environ.get( TAOTOKEN_API_KEY, s[taotoken][api_key] ) return s注意tomllib是 Python 3.11 起内置的3.10 及以下用tomli并pip install tomli。这一步别踩坑否则import tomllib直接报 ModuleNotFoundError。4. Connection 与 cursor 的可复制实现下面用 PyMySQL 演示因为它对 Connection 和 cursor 的暴露最直白和 PEP 249 规范对得上。先装依赖pip install pymysql dbutilsdbutils用来做连接池避免每次请求都新建 Connection。核心封装# db/pool.py import pymysql from dbutils.pooled_db import PooledDB from pymysql.cursors import DictCursor from db.config_loader import load_toml _cfg load_toml()[mysql] _pool PooledDB( creatorpymysql, mincached_cfg[pool][min_size], maxcached_cfg[pool][max_size], host_cfg[host], port_cfg[port], user_cfg[user], password_cfg[password], database_cfg[database], charset_cfg[charset], cursorclassDictCursor, connect_timeout_cfg[pool][connect_timeout], read_timeout_cfg[pool][read_timeout], write_timeout_cfg[pool][write_timeout], autocommitFalse, ) def get_conn(): return _pool.connection()这里有几个关键点。autocommitFalse意味着你必须显式commit()否则数据不落库这是新手最常见的“插入了但查不到”的原因。cursorclassDictCursor让fetchall()返回字典列表而不是元组列表字段名可读性高很多。mincached和maxcached控制池内空闲连接数别设太大否则数据库连接数会被占满。接下来是 Connection 和 cursor 的完整使用骨架# db/repo.py from db.pool import get_conn def create_user(name: str, email: str) - int: conn get_conn() cursor conn.cursor() try: sql INSERT INTO users (name, email) VALUES (%s, %s) cursor.execute(sql, (name, email)) conn.commit() return cursor.lastrowid except Exception: conn.rollback() raise finally: cursor.close() conn.close() def list_users(limit: int 10) - list: conn get_conn() cursor conn.cursor() try: cursor.execute(SELECT id, name, email FROM users LIMIT %s, (limit,)) return cursor.fetchall() finally: cursor.close() conn.close()cursor.execute(sql, args)的参数化写法能防 SQL 注入别用字符串拼接。lastrowid拿自增主键rowcount拿影响行数。查询用fetchall()数据量大时改用fetchmany(size)分批取或者用SSCursor无缓冲游标边取边处理内存占用低很多代价是不能回滚游标、拿不到总行数。事务边界要清楚一个业务操作一个 Connection操作内多个 cursor 共享同一事务全部成功再commit()任一失败rollback()。不要把 Connection 跨请求缓存连接池已经帮你复用了。5. 验证请求与成功结果写个main.py跑一遍完整链路确认配置、连接、游标、结果、关闭都正常# main.py from db.repo import create_user, list_users from db.config_loader import load_settings def main(): settings load_settings() print(TaoToken base_url:, settings[taotoken][base_url]) print(TaoToken key prefix:, settings[taotoken][api_key][:6] ...) uid create_user(alice, aliceexample.com) print(inserted id:, uid) rows list_users(5) for r in rows: print(r) if __name__ __main__: main()预期输出类似TaoToken base_url: https://taotoken.net/api TaoToken key prefix: sk-you... inserted id: 1 {id: 1, name: alice, email: aliceexample.com}看到inserted id有值、list_users能查出刚插的数据说明 Connection 建立、cursor 执行、结果获取、关闭这一整条链路是通的。如果inserted id是 0 或者查不到数据先检查commit()有没有调用、autocommit是不是被误设成 True 又没提交。想验证 TaoToken 通道本身是否可用可以用模型对话页快速发一条测试https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。数据库和模型通道分开验证出问题时能快速定位是哪一层。6. 本篇常见错排查报错pymysql.err.OperationalError: (2003, Cant connect to MySQL server)先确认config.toml里的 host 和 port 对得上容器里跑的话127.0.0.1要换成数据库服务名。connect_timeout设太小也会误报默认 10 秒够用。报错InterfaceError: (0, )或cursor closed典型是 Connection 提前关了或者 cursor 在 with 块外被复用。记住 cursor 生命周期不能超过它的 Connection。用try/finally保证关闭顺序是先 cursor 后 conn。插入了数据但查不到九成是没commit()。PyMySQL 默认autocommitFalseINSERT/UPDATE/DELETE 必须显式提交。查询不需要 commit但要注意事务隔离级别下未提交的数据自己能看到、别人看不到。ModuleNotFoundError: No module named tomllibPython 版本低于 3.11。要么升级要么pip install tomli并把 import 改成import tomli as tomllib。连接池耗尽请求卡住maxcached设太小、或者有代码拿了 Connection 没close()。连接池的close()是归还而不是真关闭但你不还池子就空。检查每个get_conn()是否都有对应的finally: conn.close()。UnicodeEncodeError或中文乱码charset 统一用utf8mb4建库建表也用utf8mb4。别用utf8它在 MySQL 里是残缺的三字节实现存 emoji 会炸。7. 把配置和 Key 收拢到一处数据库访问层稳定之后下一步通常是把模型调用也接进来让同一套配置同时服务数据读写和智能处理。这时候统一 Key 的价值就出来了settings.json里一个taotoken.api_key数据库走config.toml两边都通过环境变量覆盖部署时只改环境变量不动代码。需要新建或轮换 Key 时去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 操作接入参数和错误码对照看 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。如果你在做长期编码或 Agent 任务Coding Plan 页面 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 有对应的额度说明。最后留一个实用习惯把config.toml和settings.json加进.gitignore仓库里只放config.example.toml和settings.example.json。这样既不会泄露凭证新人 clone 下来照着 example 填一遍就能跑。数据库连接参数和 API Key 分开管理出问题时排查范围直接减半。