ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

Python中cursor操作数据库:TaoToken统一Key接入与可复制配置验证

Python中cursor操作数据库:TaoToken统一Key接入与可复制配置验证 1. Python 里 cursor 操作数据库到底在做什么如果你写过 Python 连数据库大概率见过conn.cursor()这行代码。很多人第一次看到会懵连接都建好了为什么还要多一个 cursor简单说conn是「电话线」cursor是「话筒」。线接通了不代表你能说话得拿起话筒才能一句一句把 SQL 发出去、把结果收回来。这个比喻我试过讲给刚入门的同事基本一遍就懂。cursor能做的事分两类。一类是「发命令」execute()执行单条 SQLexecutemany()用同一套参数批量执行callproc()调存储过程nextset()跳到下一个结果集。另一类是「收结果」fetchone()拿一行fetchmany(size)拿指定条数fetchall()全拿scroll()移动游标指针。日常查询写入90% 的场景就是execute配fetchall或fetchone。那这跟 TaoToken 有什么关系关系在于现在很多工程场景里数据库操作不再是孤立的旁边往往还挂着大模型调用——比如用模型生成 SQL、解释慢查询、把查询结果做自然语言总结。这时候如果数据库连接和模型调用各管各的 Key、各配各的通道配置会散得到处都是。TaoToken 提供的是统一 Key 和统一 API 通道把模型侧和工具侧的接入收敛到一处Python 脚本里读一个环境变量就能同时驱动数据库流程和模型流程。这篇就聚焦「Python 用 cursor 操作数据库」这条主线把可复制的连接配置、cursor 执行 SQL 的验证动作、以及和 TaoToken 统一 Key 的配合方式讲清楚让你能直接跑通查询与写入。适合谁看正在写数据脚本的后端、做数据清洗的分析同学、以及想把「数据库操作 模型辅助」串成一条流水线的工程同学。下面从环境准备开始一步步来。2. TaoToken 统一 Key 与 Python 数据库工程的前置准备先说清楚 TaoToken 在这套流程里的位置。它不是一个数据库也不替代mysql.connector或psycopg2这类驱动。它做的是「统一接入层」你申请一个 Key通过统一的 API 地址去调用模型能力同时把这个 Key 作为工程里的统一凭证来管理。数据库连接本身还是走你自己的 MySQL/PostgreSQLcursor 还是那个 cursor只是整个脚本的配置来源变干净了。前置准备分三块。第一块是数据库侧。你需要一个能连的数据库实例拿到 host、port、user、password、database 五个信息。本地用 Docker 起一个 MySQL 最省事docker run -d --name demo-mysql \ -e MYSQL_ROOT_PASSWORDroot123 \ -e MYSQL_DATABASEdemo_db \ -p 3306:3306 mysql:8.0起来之后进容器建张测试表CREATE TABLE user_score ( id INT AUTO_INCREMENT PRIMARY KEY, user_name VARCHAR(64) NOT NULL, score INT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );第二块是 Python 依赖。用mysql-connector-python就够了它是官方驱动纯 Python装起来不折腾pip install mysql-connector-python如果你用的是 PostgreSQL把驱动换成psycopg2-binarycursor 的用法几乎一样后面代码改连接串即可。第三块是 TaoToken 的 Key。到控制台创建一个 API Key地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。创建完把 Key 存到环境变量里别硬编码进脚本export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api注意 Base URL 这里不带任何查询参数就是干净的https://taotoken.net/api。模型 ID 按你实际要用的填比如做 SQL 生成或结果总结时选一个对话模型即可。这三样——Base URL、Key、Model ID——是后面所有配置的「三件套」缺一个都跑不通。提示Key 只创建时完整显示一次记得先复制到安全的地方。环境变量方式比写死在代码里安全也方便 CI 里注入。到这里前置就齐了。数据库能连、驱动装好、Key 拿到手接下来进入真正可复制的配置环节。3. 可复制的连接配置与 cursor 执行 SQL 完整代码这一节给一份能直接跑的完整脚本。它做三件事用统一配置连数据库、用 cursor 执行查询和写入、把结果用zip()整理成好读的结构。配置部分我单独抽成字典方便你替换。先看配置片段这是整个工程的「单一事实来源」import os import mysql.connector DB_CONFIG { host: os.getenv(DB_HOST, 127.0.0.1), port: int(os.getenv(DB_PORT, 3306)), user: os.getenv(DB_USER, root), password: os.getenv(DB_PASSWORD, root123), database: os.getenv(DB_NAME, demo_db), use_unicode: True, charset: utf8mb4, } TAOTOKEN_CONFIG { base_url: os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), api_key: os.getenv(TAOTOKEN_API_KEY, ), model_id: os.getenv(TAOTOKEN_MODEL_ID, 你的模型ID), }如果你更习惯用配置文件而不是环境变量可以放一个settings.toml[db] host 127.0.0.1 port 3306 user root password root123 database demo_db [taotoken] base_url https://taotoken.net/api api_key sk-你的key model_id 你的模型ID读取用标准库tomllibPython 3.11import tomllib with open(settings.toml, rb) as f: cfg tomllib.load(f) DB_CONFIG cfg[db] TAOTOKEN_CONFIG cfg[taotoken]两种方式都行团队里我更推荐环境变量因为 Key 不进版本库。接下来是连接和 cursor 操作的核心。注意conn和cursor的分工以及写入后必须commit()def get_connection(): conn mysql.connector.connect(**DB_CONFIG) return conn def insert_scores(conn, rows): rows 是 [(name, score), ...] cursor conn.cursor() sql INSERT INTO user_score (user_name, score) VALUES (%s, %s) cursor.executemany(sql, rows) conn.commit() affected cursor.rowcount cursor.close() return affected def query_scores(conn, min_score0): cursor conn.cursor() sql SELECT user_name, score FROM user_score WHERE score %s ORDER BY score DESC cursor.execute(sql, (min_score,)) rows cursor.fetchall() cursor.close() return rows这里有个细节值得说execute和executemany的参数用%s占位驱动会自动做转义别用字符串拼接否则就是 SQL 注入的活靶子。executemany适合批量插入比循环execute快不少。查询结果拿到的是元组列表直接打印不直观。用zip()把列名和值配对输出就清楚了def rows_to_dicts(columns, rows): return [dict(zip(columns, row)) for row in rows]zip()会把columns和每一行row按位置打包成元组再转成字典。如果列数和值数量不一致zip会以短的为准截断所以列名列表要和 SELECT 的字段严格对应。Python 3 里zip()返回的是迭代器要展示成列表得手动list()包一层这点和 Python 2 不一样迁移老代码时容易踩。把上面拼起来跑一次if __name__ __main__: conn get_connection() n insert_scores(conn, [(alice, 88), (bob, 92), (carol, 75)]) print(finserted {n} rows) columns [user_name, score] rows query_scores(conn, min_score80) for item in rows_to_dicts(columns, rows): print(item) conn.close()预期输出inserted 3 rows {user_name: bob, score: 92} {user_name: alice, score: 88}到这一步cursor 的查询和写入流程就完整跑通了。TaoToken 的配置此时还没被调用但已经作为统一凭证就位下一步验证请求时会用到。4. 验证请求与成功结果从 cursor 到统一 Key 的联动光跑通数据库还不够工程里更常见的诉求是把查询结果交给模型做总结或者让模型根据自然语言生成 SQL 再交给 cursor 执行。这一步验证的就是「统一 Key 能不能同时驱动这两条链路」。先验证 TaoToken 通道本身通不通。用requests发一个最小请求import requests def ask_model(prompt): url f{TAOTOKEN_CONFIG[base_url]}/v1/chat/completions headers { Authorization: fBearer {TAOTOKEN_CONFIG[api_key]}, Content-Type: application/json, } payload { model: TAOTOKEN_CONFIG[model_id], messages: [{role: user, content: prompt}], } resp requests.post(url, headersheaders, jsonpayload, timeout30) resp.raise_for_status() data resp.json() return data[choices][0][message][content]跑一句测试print(ask_model(用一句话说明什么是数据库游标))成功的话你会拿到一段正常文本。如果这里报错先看第 5 节的排查表。通道通了之后把数据库结果接进去。比如让模型把查询结果总结成一句话def summarize_scores(conn): columns [user_name, score] rows query_scores(conn, min_score0) records rows_to_dicts(columns, rows) prompt f以下是用户分数记录{records}。请用一句话总结整体情况。 return ask_model(prompt)实测下来这条链路的价值在于cursor 负责把结构化数据取出来模型负责把数据讲成人话而两者共用同一个 Key 和 Base URL配置只有一份。你不需要为模型调用单独维护一套凭证也不用担心两边的地址写串。再进一步让模型生成 SQL、cursor 执行。这个场景要小心模型生成的 SQL 不能直接无脑执行最好限定只读def nl_to_sql(question): prompt ( 表结构user_score(id, user_name, score, created_at)。 f请只输出一条 SELECT 语句不要解释。问题{question} ) return ask_model(prompt).strip() def run_generated_sql(conn, sql): if not sql.lower().startswith(select): raise ValueError(只允许执行 SELECT) cursor conn.cursor() cursor.execute(sql) rows cursor.fetchall() columns [d[0] for d in cursor.description] cursor.close() return rows_to_dicts(columns, rows)cursor.description会返回结果集的列信息取每项的第一个元素就是列名这样不用手写 columns 列表通用性更好。跑一次sql nl_to_sql(分数大于 80 的用户有哪些) print(sql) print(run_generated_sql(conn, sql))成功结果应该是模型输出一条 SELECT然后 cursor 执行后返回对应记录。到这一步统一 Key 接入 cursor 操作数据库的完整闭环就验证完了。5. 本篇常见报错排查对照这一节按真实会遇到的报错来。每条都给现象、原因、处理。401 Unauthorized。现象是调用模型接口返回 401。原因通常是 Key 没读到或写错。检查TAOTOKEN_API_KEY环境变量是否真的导出成功echo $TAOTOKEN_API_KEY看一眼。注意 Key 前面有没有多余空格复制时容易带上。如果用的是配置文件确认api_key字段没被引号包错。local proxy failed / connection refused。现象是请求发不出去提示本地代理失败。这多半是环境里设了HTTP_PROXY/HTTPS_PROXY指向一个没启动的本地端口。检查env | grep -i proxy把不需要的代理变量清掉再试。数据库连接报Cant connect to MySQL server也是类似思路先确认 host/port 对不对、容器起没起。reading choices 报 KeyError。现象是data[choices]取不到。原因一般是返回体不是预期的对话结构可能是模型 ID 填错、或者请求体格式不对。先把resp.text打出来看原始返回别急着取字段。确认model字段用的是你账号下真实可用的模型 ID。OAuth / 认证相关报错。如果你在别的工具里配过 OAuth 流程切到 API Key 方式时可能残留旧配置。检查有没有旧的 token 文件或环境变量在干扰清掉后重新用 Key 认证。cursor 相关报错。cursor.execute报ProgrammingError多半是 SQL 语法或表名写错fetchall返回空列表先确认commit()有没有执行写入没提交的话查询看不到数据。executemany报参数数量不匹配检查每个元组的长度和占位符数量是否一致。中文乱码。连接配置里加charsetutf8mb4和use_unicodeTrue建表时也用 utf8mb4两边对齐就不会乱。注意排查时优先看原始返回文本和完整堆栈别只看最后一行错误信息很多线索在前面几行。6. 把统一 Key 用进你的日常数据脚本走到这里你已经有了三样东西一份可复制的数据库连接配置、一套 cursor 查询写入的验证代码、以及一个能同时驱动模型调用的统一 Key。接下来怎么用取决于你的场景。如果你只是偶尔写数据脚本把DB_CONFIG和TAOTOKEN_CONFIG抽成一个config.py所有脚本 import 它改配置只改一处。如果你在做长期的数据流水线建议把 Key 走环境变量或密钥管理服务别进代码库。如果你要跑的是长时间、多轮的编码或 Agent 任务可以了解下 Coding Plan地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合持续性的开发场景。想快速试模型对话效果可以直接用模型对话页 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有各语言的调用示例。控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 可以看用量和 Key 管理。最后留一个实用习惯每次改完数据库配置先跑一句SELECT 1确认连接活着再跑业务 SQL。这个动作花不了两秒但能帮你把「连接问题」和「SQL 问题」分开排查时省很多事。
返回列表