Python数据库篇：sqlite3、pymysql、aiomysql、sqlalchemy、pydantic 配 TaoToken 统一 Key 通道

1. 数据层工具链接 AI 时Key 到底该放哪写 Python 数据层代码的人迟早会遇到一个尴尬场景sqlite3 脚本跑得好好的pymysql 连接池也调通了aiomysql 的异步任务在 FastAPI 里稳定运行sqlalchemy 的 ORM 映射和 pydantic 的校验模型都写完了结果要往里面加一个「AI 能力」——比如让模型帮忙生成 SQL、解析字段含义、给 pydantic 模型自动补 description——这时候 Key 往哪放就成了问题。我见过太多项目把 Key 硬编码在config.py里或者散落在.env、settings.json、config.toml三份配置里各写一遍。更麻烦的是sqlite3 这种本地文件数据库的脚本、pymysql 的同步连接、aiomysql 的异步连接池、sqlalchemy 的 engine、pydantic 的 Settings 类它们读配置的方式完全不同。一个 Key 要在五个地方维护改一次漏一处线上就报 401。这篇要解决的就是这件事用 TaoToken 作为统一的 Key/API 通道把 sqlite3、pymysql、aiomysql、sqlalchemy、pydantic 这五类数据层工具在接入 AI 能力时的配置收敛到一处。适合正在写数据管道、异步爬虫落库、FastAPI 数据服务或者想给 ORM 模型加 AI 辅助生成逻辑的开发者。读完你能拿到一份可直接复制的settings.json和config.toml骨架以及一次能跑通的连通性验证动作。TaoToken 在这里的角色很简单它是一个统一的 API 通道你只需要维护一个 Key数据层代码通过它去调用模型能力不用在每个数据库工具里各配一套凭证。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。2. 前置准备把 Key 和通道先理清楚在动数据库代码之前先把通道打通。这一步不做后面所有配置都是空中楼阁。2.1 拿到统一 Key登录 TaoToken 控制台在 API Keys 页面创建一个 Key。这个 Key 就是你后面所有数据层工具共用的那一个。控制台地址带 deep linkhttps://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 。创建时注意两点一是权限范围如果你只是做数据层的辅助调用生成 SQL、补字段描述给最小必要权限就行二是把 Key 复制到剪贴板后立刻存进密码管理器页面刷新后不再显示完整 Key。2.2 确认 API 基址TaoToken 的 API 基址是https://taotoken.net/api注意这个地址不带 UTM 参数是纯 API 端点。所有数据层工具在配置时都指向这个基址区别只在于它们各自用什么方式读取配置。2.3 环境变量先行不管后面用 json 还是 toml我建议先把 Key 放进环境变量配置文件里只放引用。这样即使配置文件被误提交到仓库Key 也不会泄露。# Linux / macOS export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api # Windows PowerShell $env:TAOTOKEN_API_KEYsk-你的key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api这一步做完后面所有工具都从环境变量读配置文件只负责组织其他参数。3. 可复制配置settings.json 与 config.toml 骨架数据层工具链的配置分两派一派吃 jsonpydantic Settings、部分 FastAPI 项目一派吃 toml现代 Python 项目、poetry、部分 sqlalchemy 项目。两份都给你按项目习惯选。3.1 settings.json 骨架这份配置面向 pydantic Settings 和需要 json 配置的场景。核心思路是把数据库连接信息和 AI 通道信息分开AI 部分只留一个 Key 引用。{ database: { sqlite: { path: ./data/app.db }, mysql: { host: 127.0.0.1, port: 3306, user: root, password: your_db_password, database: test, charset: utf8mb4, pool_recycle: 3600 } }, ai: { provider: taotoken, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, default_model: claude-sonnet-4-20250514, timeout: 30, max_retries: 3 }, pydantic: { from_attributes: true, validate_assignment: true } }注意api_key_env这个字段它存的是环境变量名而不是 Key 本身。pydantic Settings 读取时用os.environ去取这样配置文件可以安全地进版本控制。3.2 config.toml 骨架toml 版本更适合 sqlalchemy 项目和现代 Python 工具链。结构上和 json 对齐但 toml 支持注释可读性更好。[database.sqlite] path ./data/app.db [database.mysql] host 127.0.0.1 port 3306 user root password your_db_password database test charset utf8mb4 pool_recycle 3600 minsize 5 maxsize 20 [ai] provider taotoken base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY default_model claude-sonnet-4-20250514 timeout 30 max_retries 3 [pydantic] from_attributes true validate_assignment true3.3 用 pydantic Settings 统一读取不管底层是 json 还是 toml最终都用 pydantic 的BaseSettings收敛成一个对象。这样 sqlite3、pymysql、aiomysql、sqlalchemy 都从同一个 Settings 实例拿配置。import os import json from pathlib import Path from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict class DatabaseConfig(BaseSettings): host: str 127.0.0.1 port: int 3306 user: str root password: str database: str test charset: str utf8mb4 pool_recycle: int 3600 minsize: int 5 maxsize: int 20 class AIConfig(BaseSettings): provider: str taotoken base_url: str https://taotoken.net/api api_key_env: str TAOTOKEN_API_KEY default_model: str claude-sonnet-4-20250514 timeout: int 30 max_retries: int 3 property def api_key(self) - str: key os.environ.get(self.api_key_env) if not key: raise RuntimeError(f环境变量 {self.api_key_env} 未设置) return key class Settings(BaseSettings): model_config SettingsConfigDict( env_nested_delimiter__, extraignore, ) database: DatabaseConfig Field(default_factoryDatabaseConfig) ai: AIConfig Field(default_factoryAIConfig) classmethod def from_json(cls, path: str | Path) - Settings: data json.loads(Path(path).read_text(encodingutf-8)) return cls(**data) settings Settings.from_json(./settings.json)这段代码的关键点是api_key做成 property每次访问时从环境变量读而不是在初始化时固化。这样你在测试里可以临时改环境变量不用重建 Settings 对象。4. 五类工具接入验证从 sqlite3 到 pydantic配置骨架有了接下来逐个验证。每个工具我都给一段能直接跑的代码重点是展示它怎么从统一 Settings 拿 AI 通道信息。4.1 sqlite3本地脚本的轻量接入sqlite3 是标准库不需要额外安装。它的场景是本地小数据、脚本工具、原型验证。接入 AI 通道时通常是用模型帮你生成建表语句或解析字段。import sqlite3 import json import urllib.request from settings import settings def ask_ai(prompt: str) - str: 通过 TaoToken 统一通道调用模型 payload json.dumps({ model: settings.ai.default_model, messages: [{role: user, content: prompt}], }).encode(utf-8) req urllib.request.Request( f{settings.ai.base_url}/v1/messages, datapayload, headers{ Content-Type: application/json, x-api-key: settings.ai.api_key, anthropic-version: 2023-06-01, }, methodPOST, ) with urllib.request.urlopen(req, timeoutsettings.ai.timeout) as resp: body json.loads(resp.read().decode(utf-8)) return body[content][0][text] def init_db(): conn sqlite3.connect(settings.database.sqlite_path if hasattr(settings.database, sqlite_path) else ./data/app.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS user ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, email TEXT UNIQUE ) ) conn.commit() cursor.close() conn.close() if __name__ __main__: init_db() result ask_ai(用一句话说明 sqlite3 的 WAL 模式适合什么场景) print(result)这里用标准库urllib是为了不引入额外依赖实际项目里换成httpx或requests更顺手。注意请求头用的是x-api-key这是 Anthropic 兼容格式如果你走的是 OpenAI 兼容格式改成Authorization: Bearer。4.2 pymysql同步连接的配置复用pymysql 是同步 MySQL 客户端Django 项目和老代码里常见。它的连接参数直接从 Settings 的 DatabaseConfig 拿。import pymysql from pymysql.cursors import DictCursor from settings import settings def get_conn(): db settings.database return pymysql.connect( hostdb.host, portdb.port, userdb.user, passworddb.password, databasedb.database, charsetdb.charset, autocommitTrue, cursorclassDictCursor, ) def create_table(): conn get_conn() cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS sys_user2 ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, username VARCHAR(15) NOT NULL COMMENT 用户名, gender TINYINT UNSIGNED DEFAULT 0 COMMENT 性别(0:女 1:男), amount DECIMAL(10,2) DEFAULT NULL, create_time DATETIME DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk (username) USING BTREE ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 ) cursor.close() conn.close() def insert_many(rows): conn get_conn() cursor conn.cursor() sql INSERT INTO sys_user2(username, gender, amount) VALUES(%s, %s, %s) cursor.executemany(sql, rows) cursor.close() conn.close()注意autocommitTrue时不需要显式 commit但如果你要做事务把它设成 False 然后手动conn.commit()。这个坑在 aiomysql 那边更明显后面会讲。4.3 aiomysql异步连接池与 AI 调用共存aiomysql 是 asyncio 生态下的 MySQL 客户端FastAPI、aiohttp 项目里常用。它的连接池参数和 AI 通道配置都从同一个 Settings 拿这是统一 Key 通道的价值所在。import asyncio import json import httpx import aiomysql from settings import settings async def create_pool(): db settings.database return await aiomysql.create_pool( hostdb.host, portdb.port, userdb.user, passworddb.password, dbdb.database, charsetdb.charset, minsizedb.minsize, maxsizedb.maxsize, pool_recycledb.pool_recycle, autocommitFalse, ) async def ask_ai_async(prompt: str) - str: async with httpx.AsyncClient(timeoutsettings.ai.timeout) as client: resp await client.post( f{settings.ai.base_url}/v1/messages, headers{ x-api-key: settings.ai.api_key, anthropic-version: 2023-06-01, Content-Type: application/json, }, json{ model: settings.ai.default_model, messages: [{role: user, content: prompt}], }, ) resp.raise_for_status() return resp.json()[content][0][text] async def query_with_ai(pool, user_id: int): async with pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cur: await cur.execute( SELECT id, username, amount FROM sys_user2 WHERE id %s, (user_id,), ) row await cur.fetchone() if row: desc await ask_ai_async(f用一句话描述用户 {row[username]} 的账户状态) row[ai_desc] desc return row async def main(): pool await create_pool() try: result await query_with_ai(pool, 1) print(json.dumps(result, ensure_asciiFalse, indent2)) finally: pool.close() await pool.wait_closed() if __name__ __main__: asyncio.run(main())这里有个关键点pool_recycle3600必须设。MySQL 默认wait_timeout是 8 小时但运维经常调成几分钟不设 recycle 的话连接池里会攒一堆失效连接报Lost connection或MySQL server has gone away。4.4 sqlalchemyORM 与 AI 通道的配置对齐sqlalchemy 的 engine 创建需要连接串这个连接串从 Settings 拼出来。同时 AI 通道的配置也走同一个 Settings保证一致性。from sqlalchemy import create_engine, Column, BigInteger, String, DECIMAL, DateTime, func from sqlalchemy.orm import declarative_base, sessionmaker from settings import settings db settings.database DATASOURCE_URL ( fmysqlpymysql://{db.user}:{db.password}{db.host}:{db.port}/ f{db.database}?charset{db.charset} ) engine create_engine( DATASOURCE_URL, echoFalse, pool_recycledb.pool_recycle, pool_sizedb.minsize, max_overflowdb.maxsize - db.minsize, ) Base declarative_base() class User(Base): __tablename__ user id Column(BigInteger, primary_keyTrue, autoincrementTrue) username Column(String(15), uniqueTrue, nullableFalse, comment用户名) amount Column(DECIMAL(10, 2), nullableTrue) create_time Column(DateTime, server_defaultfunc.now(), comment注册时间) SessionLocal sessionmaker(bindengine) def get_session(): return SessionLocal()注意pool_size和max_overflow的配合pool_size是常驻连接数max_overflow是峰值时额外允许的连接数。总上限是两者之和别超过 MySQL 的max_connections。4.5 pydantic模型校验与 AI 配置的最终收敛pydantic 在这里有两个角色一是校验数据库实体二是作为 Settings 的基类。前面 3.3 已经用了BaseSettings这里补充一个业务模型的例子展示Field的常用参数。from enum import Enum from typing import List, Optional from pydantic import BaseModel, Field, ConfigDict, ValidationError class Intent(str, Enum): CHAT chat MEETING meeting UNKNOWN unknown class MeetingAction(BaseModel): owner: str Field(..., description负责人) task: str Field(..., description任务内容) deadline: Optional[str] Field(defaultNone, description截止日期) class MeetingSummary(BaseModel): model_config ConfigDict(from_attributesTrue) title: str Field(..., description会议标题) date: str Field(..., description会议日期) attendees: List[str] Field(..., description参会人员) summary: str Field(..., max_length8000, description会议摘要) decisions: List[str] Field(default_factorylist, description决议事项) action_items: List[MeetingAction] Field(default_factorylist, description行动项) notes: Optional[str] Field(defaultNone, description备注) class RouterDecision(BaseModel): intent: Intent confidence: float Field(ge0.0, le1.0) reasoning: str extracted_info: Optional[dict] None if __name__ __main__: decision RouterDecision( intentIntent.CHAT, confidence0.95, reasoning用户在进行日常对话, ) print(decision.model_dump_json(indent2)) try: RouterDecision(intentIntent.CHAT, confidence1.5, reasoning越界) except ValidationError as exc: print(校验失败:, exc.error_count(), 处)default_factorylist这个点必须强调如果你写decisions: List[str] []所有实例会共享同一个列表对象改一个影响全部。这是 Python 的经典可变默认值陷阱pydantic 用default_factory规避。5. 连通性验证一次可复制的动作配置写完了怎么确认整条链路是通的我给一个最小验证脚本它同时检查数据库连接和 AI 通道。import asyncio import httpx import aiomysql from settings import settings async def verify_db(): db settings.database conn await aiomysql.connect( hostdb.host, portdb.port, userdb.user, passworddb.password, dbdb.database, charsetdb.charset, ) async with conn.cursor() as cur: await cur.execute(SELECT 1) row await cur.fetchone() conn.close() return row[0] 1 async def verify_ai(): async with httpx.AsyncClient(timeoutsettings.ai.timeout) as client: resp await client.post( f{settings.ai.base_url}/v1/messages, headers{ x-api-key: settings.ai.api_key, anthropic-version: 2023-06-01, Content-Type: application/json, }, json{ model: settings.ai.default_model, max_tokens: 32, messages: [{role: user, content: 回复 OK 两个字母}], }, ) resp.raise_for_status() return resp.json()[content][0][text] async def main(): db_ok await verify_db() print(f数据库连通: {db_ok}) ai_reply await verify_ai() print(fAI 通道连通: {ai_reply}) if __name__ __main__: asyncio.run(main())跑通的话你会看到类似输出数据库连通: True AI 通道连通: OK如果数据库那行报错检查 MySQL 是否启动、账号密码是否正确、settings.json里的database字段是否指向存在的库。如果 AI 那行报 401检查环境变量TAOTOKEN_API_KEY是否设置、Key 是否过期。如果报 404检查base_url是否写成了https://taotoken.net/api而不是带路径的完整地址。6. 本篇常见错排查6.1 aiomysql 插了数据但库里没有这是最高频的问题。aiomysql 默认autocommitFalse你execute了 INSERT 但没commit连接归还池子时自动 rollback。解决方式二选一要么在create_pool时设autocommitTrue适合写少读多要么每次写操作后显式await conn.commit()。# 方式一池级 autocommit pool await aiomysql.create_pool(..., autocommitTrue) # 方式二显式 commit async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute(INSERT INTO logs(msg) VALUES (%s), (hello,)) await conn.commit()6.2 pymysql 的占位符加了引号pymysql 和 aiomysql 的占位符统一是%s不管字段类型是字符串、整数还是日期都不加引号。写成%s会导致参数被当成字符串字面量查询结果为空或报错。# 正确 await cur.execute(SELECT * FROM users WHERE name %s, (Alice,)) # 错误多了引号 await cur.execute(SELECT * FROM users WHERE name %s, (Alice,))6.3 sqlalchemy 连接池耗尽pool_size设太小、max_overflow设 0高并发时请求会卡在获取连接上报TimeoutError: QueuePool limit of size X overflow Y reached。调大pool_size和max_overflow同时确认 MySQL 的max_connections够用。6.4 pydantic 的 from_attributes 没开用model_validate从 sqlalchemy 对象转 pydantic 模型时如果没设from_attributesTrue会报Input should be a valid dictionary or instance of X。在model_config里加上就行。class UserModel(BaseModel): model_config ConfigDict(from_attributesTrue) username: str age: int6.5 环境变量没生效settings.json里写的是api_key_env: TAOTOKEN_API_KEY但你在 shell 里export之后又开了新终端环境变量丢了。用echo $TAOTOKEN_API_KEY确认一下。Windows 下注意 PowerShell 和 CMD 的语法不同。6.6 连接失效没重试网络抖动或 MySQL 重启会导致连接失效。aiomysql 建议设pool_recyclesqlalchemy 建议配pool_pre_pingTrue。关键操作外层可以加tenacity重试。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(min1, max5)) async def safe_query(pool, sql, params): async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute(sql, params) return await cur.fetchall()7. 下一步按场景选通道配置和验证都跑通之后接下来看你具体要做什么。如果你是在排障、接入新项目或者需要重新生成 Key去 API Keys 页面和接入文档https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 和 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。如果你想先验证模型输出质量比如让它生成 SQL 或补字段描述用模型对话页面快速试https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 。如果你要把 AI 能力长期嵌进数据管道、写 Agent 自动处理数据库任务Coding Plan 更适合https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。最后提醒一句settings.json和config.toml里永远不要写明文 Key用环境变量引用。配置文件进仓库Key 进密码管理器这是数据层项目接 AI 通道时最该守住的一条线。