
1. 为什么 Connection 和 Cursor 总被写混刚接触 Python 数据库开发时最容易犯的错就是把 Connection 和 Cursor 当成一回事。我见过不少代码连上数据库后直接拿连接对象去execute结果报AttributeError也有人每次查询都新建一个连接跑几百次循环直接把数据库连接数打满。这两个对象到底谁管什么其实用一句话就能记住Connection 是电话线Cursor 是话筒。Connection 负责的是通不通的问题。它拿着 host、port、user、password、db 这些参数去和数据库服务器握手握手成功后维持一条会话通道事务的提交和回滚也挂在它身上。Cursor 负责的是说什么的问题。它依附在某个 Connection 上真正把 SQL 语句发出去、把结果集取回来execute、fetchone、fetchall、rowcount这些方法都属于它。一个连接可以开多个游标就像一条电话线可以接多个分机但分机不能脱离电话线单独存在。理解了这个边界再来看配置就顺了。这篇要解决的是在 Python 项目里用一份settings.json统一管理数据库连接参数同时把 TaoToken 的统一 Key 和 API 通道也纳入同一份配置骨架让数据库连接和模型调用走同一套配置加载逻辑最后跑一次真实的连接建立加游标执行来验证整条链路通不通。适合正在写数据脚本、做后端接口、或者要给 Agent 接数据库的开发者。2. TaoToken 前置统一 Key 与配置骨架的位置在动手写settings.json之前先把 TaoToken 这边的准备工作做完。TaoToken 在这里扮演的角色是统一入口你不需要在代码里散落一堆不同的 Key 和地址而是把模型通道的凭证集中到一份配置里和数据库参数并列管理。这样做的直接好处是换环境、换 Key、加新通道时只改一个文件。第一步是拿到 API Key。打开控制台进入 API Keys 页面创建一个新的 Key复制出来先存好后面要填进settings.json。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite API Keys 页面是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。创建时建议按用途命名比如db-agent-dev方便以后区分。第二步是确认 API 基地址。TaoToken 的 API 入口是 https://taotoken.net/api 注意这个地址后面不加任何查询参数代码里拼接路径时直接在它后面接/v1/...这类标准路径即可。如果你用的是兼容 OpenAI 风格的 SDK把base_url指向这个地址就行。第三步如果你打算让模型直接参与写 SQL、解释表结构、或者做 coding 相关的自动化可以顺手了解一下 Coding Plan它面向的是长期编码和 Agent 场景地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。这一步不是必须的但如果你后面要把数据库操作接进 Agent 流程提前规划会更省事。准备工作做完你手上应该有三样东西一个 API Key、一个 API 基地址、以及数据库本身的连接参数host、port、user、password、dbname。接下来把它们全部塞进settings.json。3. 可复制的 settings.json 配置骨架下面这份骨架是我实际项目里精简出来的版本分三个区块database放数据库连接参数taotoken放统一 Key 和 API 地址runtime放一些运行时开关。你可以直接复制把值替换成自己的。{ database: { driver: mysql, host: 127.0.0.1, port: 3306, user: app_user, password: your_db_password, dbname: demo_db, charset: utf8mb4, connect_timeout: 10, autocommit: false }, taotoken: { api_base: https://taotoken.net/api, api_key: sk-your-taotoken-key, default_model: claude-sonnet, timeout: 60 }, runtime: { pool_size: 5, pool_recycle: 1800, echo_sql: false } }几个参数值得单独说。autocommit设成false是有意的这样事务边界由你在代码里显式控制commit和rollback的时机清晰排查问题时不会出现数据莫名其妙写进去了的情况。connect_timeout给 10 秒避免网络抖动时程序卡死。pool_recycle设 1800 秒是因为 MySQL 默认的wait_timeout通常是 8 小时但中间如果有防火墙或代理掐连接长连接会变成死连接定期回收能绕开这个坑。读取这份配置的代码可以写得很薄import json from pathlib import Path def load_settings(path: str settings.json) - dict: with Path(path).open(r, encodingutf-8) as f: return json.load(f) settings load_settings() db_conf settings[database] tt_conf settings[taotoken]这里没有引入任何第三方配置库标准库的json就够了。如果你项目里已经有pydantic-settings或dynaconf把这份 JSON 喂进去做校验也可以但骨架本身保持纯 JSON迁移成本最低。注意api_key不要提交到 Git。生产环境用环境变量覆盖比如读取时先看os.environ.get(TAOTOKEN_API_KEY)有值就优先用环境变量。4. 连接建立与游标执行的验证动作配置就位后跑一次完整的验证建立 Connection开 Cursor执行一条查询取回结果然后正确关闭。下面这段代码用pymysql作为驱动示例其他驱动psycopg2、mysql-connector-python的 Connection 和 Cursor 方法名基本一致替换 import 即可。import pymysql from settings_loader import load_settings settings load_settings() db_conf settings[database] conn None cursor None try: conn pymysql.connect( hostdb_conf[host], portdb_conf[port], userdb_conf[user], passworddb_conf[password], databasedb_conf[dbname], charsetdb_conf[charset], connect_timeoutdb_conf[connect_timeout], autocommitdb_conf[autocommit], ) print(Connection 建立成功:, conn) cursor conn.cursor() cursor.execute(SELECT VERSION()) row cursor.fetchone() print(数据库版本:, row[0]) cursor.execute(SELECT 1 AS ping) print(连通性 ping:, cursor.fetchone()) conn.commit() except pymysql.MySQLError as e: if conn: conn.rollback() print(数据库操作失败:, e) finally: if cursor: cursor.close() if conn: conn.close()这段代码里有几个动作是刻意分开的正好对应两个对象的职责。pymysql.connect(...)返回的是 Connection它只负责握手和会话。conn.cursor()返回的是 Cursorexecute和fetchone都挂在它身上。conn.commit()是连接级别的事务提交不是游标的方法。cursor.close()先关游标conn.close()再关连接顺序反了在某些驱动上会报警告。跑通之后你会看到类似输出Connection 建立成功: pymysql.connections.Connection object at 0x... 数据库版本: 8.0.36 连通性 ping: (1,)到这里数据库这条链路就验证完了。接下来验证 TaoToken 通道。用同一份配置里的api_base和api_key发一个最小请求import json import urllib.request settings load_settings() tt settings[taotoken] payload json.dumps({ model: tt[default_model], messages: [{role: user, content: 回复两个字通了}], max_tokens: 16 }).encode(utf-8) req urllib.request.Request( tt[api_base] /v1/chat/completions, datapayload, headers{ Content-Type: application/json, Authorization: Bearer tt[api_key], }, methodPOST, ) with urllib.request.urlopen(req, timeouttt[timeout]) as resp: body json.loads(resp.read().decode(utf-8)) print(TaoToken 通道返回:, body[choices][0][message][content])如果返回里能看到模型回复的内容说明统一 Key 和 API 通道也是通的。想更直观地对比不同模型的表现可以直接在模型对话页面里试地址是 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 把同样的 prompt 丢进去看输出和代码里的结果对照一下。5. 本篇常见错排查报错一AttributeError: Connection object has no attribute execute这是最典型的职责混淆。execute是 Cursor 的方法不是 Connection 的。检查你的代码是不是写成了conn.execute(sql)改成先cursor conn.cursor()再cursor.execute(sql)。报错二pymysql.err.OperationalError: (2003, Cant connect to MySQL server on ...)连接层的问题和 Cursor 无关。按顺序排查host和port是否写对数据库服务是否在跑防火墙是否放行connect_timeout是否太短。如果本地能连、远程连不上优先看数据库的bind-address配置。报错三pymysql.err.ProgrammingError: Cursor is closed游标已经close()了还在用。常见于把 cursor 定义在函数外、函数结束后被关闭下次调用又拿它执行。解决办法是每次操作前重新conn.cursor()或者用上下文管理器with conn.cursor() as cur:让作用域自动管理。报错四查询结果为空但表里明明有数据先看autocommit和事务。如果连接设了autocommitFalse而你在另一个会话里插入的数据还没commit当前会话是看不到的。另外检查charsetutf8和utf8mb4在部分字符上表现不同中文乱码时优先换成utf8mb4。报错五TaoToken 请求返回 401api_key没填对或者Authorization头拼错了。确认格式是Bearer加空格再加 Key别漏了空格。如果 Key 是从控制台复制的注意别把首尾空白带进去。报错六连接池耗尽Too many connections循环里反复connect()没关闭。把连接复用起来或者引入连接池runtime.pool_size就是给这个用的。每次操作完确保finally里执行了close()。6. 把配置和验证固化进项目到这里settings.json的骨架、Connection 与 Cursor 的职责边界、以及两条链路的验证动作都跑通了。我的建议是把第 4 节的两段验证代码抽成一个healthcheck.py每次改完配置先跑它比在业务代码里 debug 快得多。数据库参数和 TaoToken 的 Key 放在同一份配置里好处是环境切换时只动一个文件坏处是这份文件敏感度变高记得加进.gitignore生产环境用环境变量覆盖。如果你后面要把数据库操作接进 Agent 或者自动化编码流程可以看看 Coding Plan 的接入方式地址是 https://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/claudecode?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite 需要的话一并看。最后留一个我踩过的坑cursor.fetchall()在结果集很大时会一次性把数据全读进内存几百万行的表能把进程撑爆。这种场景改用fetchmany(size)分批取arraysize设成 1000 左右配合生成器逐批处理内存占用会平稳很多。