ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Python数据库添加时间:用pymysql与datetime把TaoToken接入数据写入链路

Python数据库添加时间:用pymysql与datetime把TaoToken接入数据写入链路 1. 为什么时间字段总在写入时出问题写 Python 后端连 MySQL时间字段几乎是绕不开的一环。你建表时写了created_at DATETIME代码里用datetime.now()拼进 SQL本地跑得好好的一上测试环境就报Incorrect datetime value或者查出来时间差了 8 小时。这类问题我遇到过太多次根子往往不在 SQL 本身而在三个地方时间对象怎么生成、驱动怎么转义、连接通道怎么配置。先说清楚这篇要解决什么。pymysql是 Python 里最常用的纯 Python MySQL 驱动datetime是标准库自带的时间模块两者配合能把时间戳准确写进数据库。但很多教程只给一段INSERT就结束了没讲时区、没讲批量写入、没讲连接配置怎么统一管理。更现实的问题是当你的项目里数据库连接、模型调用、密钥管理散落在各个文件时改一个配置要翻五个地方很容易漏。所以这篇的目标是给你一套能直接复制的建表 SQL、pymysql连接与插入代码、时间格式校验脚本以及查询回读验证步骤。同时演示怎么把数据库连接配置收敛到 TaoToken 的统一 Key/API 通道上让后端写入链路和模型调用走同一套凭证管理。适合正在写 Flask/FastAPI/Django 后端、需要处理时间字段写入的开发者也适合想把项目配置统一起来的人。核心检索词先摆出来Python 用 pymysql 往 MySQL 添加时间字段配合 datetime 生成时间戳并统一连接配置。下面从建表开始一步步来。2. TaoToken 统一 Key 与 API 通道的前置准备在写插入代码之前先把连接配置这件事理清楚。传统做法是把host、user、password、database硬编码在config.py里或者塞进环境变量。项目小的时候没问题一旦要接模型 API、要换环境、要给多个服务发凭证就会乱。TaoToken 提供的是一个统一入口把 API Key 和请求通道集中管理数据库连接配置也可以按同样的思路收敛。你需要先拿到一个可用的 Key。打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台创建 API 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 。API 的基础地址是 https://taotoken.net/api 注意这个地址不带 UTM 参数直接用于代码里的base_url。这里要说明一点TaoToken 的 API 通道主要面向模型调用场景数据库连接本身还是走你自己的 MySQL 实例。但项目里往往同时有「写数据库」和「调模型」两条链路把 Key 和 base_url 统一放在一个配置模块里两边共用一套读取逻辑维护成本会低很多。我试过把数据库密码和 API Key 都放进.env用python-dotenv加载代码里只引用变量名换环境时只改.env一个文件。具体准备三样东西第一一个 TaoToken API Key形如sk-开头的一串字符第二你的 MySQL 连接信息包括 host、port、user、password、database第三Python 环境里装好pymysql和python-dotenv。安装命令pip install pymysql python-dotenvdatetime是标准库不需要额外安装。如果你看到某些教程写pip install datetime那是错的会装到一个同名的无关包别踩这个坑。配置文件的组织方式建议这样项目根目录放一个.env里面写数据库和 API 的凭证再放一个config.py负责读取并暴露配置对象。这样插入代码和模型调用代码都从config.py拿配置不会各写各的。下一节给出完整的可复制片段。3. 可复制的建表 SQL 与 pymysql 插入配置先建表。时间字段用DATETIME默认值设CURRENT_TIMESTAMP这样即使代码里漏传时间数据库也会兜底。建表 SQL 如下CREATE TABLE IF NOT EXISTS triads ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT NOT NULL, entity_one VARCHAR(128) NOT NULL, relation VARCHAR(64) NOT NULL, entity_two VARCHAR(128) NOT NULL, industry VARCHAR(32) DEFAULT csv, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_user_created (user_id, created_at) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;注意created_at和updated_at的区别前者只在插入时写一次后者在每次更新时自动刷新。如果你要手动控制时间就把默认值去掉全部由 Python 侧传入。接下来是配置文件。.env内容DB_HOST127.0.0.1 DB_PORT3306 DB_USERroot DB_PASSWORDyour_db_password DB_NAMEyour_database TAOTOKEN_API_KEYsk-your-key-here TAOTOKEN_BASE_URLhttps://taotoken.net/apiconfig.py读取配置import os from dotenv import load_dotenv load_dotenv() class Config: DB_HOST os.getenv(DB_HOST, 127.0.0.1) DB_PORT int(os.getenv(DB_PORT, 3306)) DB_USER os.getenv(DB_USER, root) DB_PASSWORD os.getenv(DB_PASSWORD, ) DB_NAME os.getenv(DB_NAME, ) TAOTOKEN_API_KEY os.getenv(TAOTOKEN_API_KEY, ) TAOTOKEN_BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) classmethod def db_kwargs(cls): return { host: cls.DB_HOST, port: cls.DB_PORT, user: cls.DB_USER, password: cls.DB_PASSWORD, database: cls.DB_NAME, charset: utf8mb4, autocommit: False, }这里autocommit设为False意味着你需要显式commit()。批量插入时这样更安全出错可以rollback()。如果你希望每条语句自动提交改成True但批量场景不推荐。然后是插入代码。核心是用datetime.now()生成时间对象再用strftime格式化成 MySQL 认识的字符串。为什么不用datetime.now()直接传因为pymysql对datetime对象本身是支持的但如果你在拼接 SQL 时手动转字符串格式必须严格匹配%Y-%m-%d %H:%M:%S否则报错。两种方式都可以我下面用参数化查询让驱动自己处理转义更安全。import datetime import pymysql from config import Config class TriadStore: def __init__(self): self.conn pymysql.connect(**Config.db_kwargs()) self.cursor self.conn.cursor() def store_data(self, user_id, spo_list, industrycsv): sql ( INSERT INTO triads (user_id, entity_one, relation, entity_two, industry, created_at) VALUES (%s, %s, %s, %s, %s, %s) ) now_str datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) try: combine [] for item in spo_list: combine.append( [str(user_id)] list(item) [industry, now_str] ) self.cursor.executemany(sql, combine) self.conn.commit() return True except Exception as e: print(finsert failed: {e}) self.conn.rollback() return False def close(self): self.cursor.close() self.conn.close()关键点executemany的第二个参数是「列表的列表」每个内层列表对应一行顺序必须和 SQL 里的占位符一一对应。now_str在循环外生成一次保证同一批数据时间一致。如果你希望每行时间略有差异把strftime挪进循环。关于时区datetime.now()返回的是本地时间不带时区信息。如果你的服务器时区和 MySQL 时区不一致写进去的时间会有偏差。稳妥做法是统一用 UTC或者在建连接时设置init_commandSET time_zone08:00。我一般让服务器和数据库都走 UTC展示层再转本地。4. 验证请求与查询回读的完整步骤代码写完不能只看「没报错」就完事要回读验证。先写一个时间格式校验脚本确认生成的字符串符合 MySQL 的DATETIME格式import datetime import re def validate_time_format(dt_str): pattern r^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$ if not re.match(pattern, dt_str): raise ValueError(fbad datetime format: {dt_str}) datetime.datetime.strptime(dt_str, %Y-%m-%d %H:%M:%S) return True now_str datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) print(now_str, validate_time_format(now_str))跑一下输出类似2025-01-15 14:32:07 True。如果格式不对strptime会抛异常提前暴露问题。接着执行插入并回读from store import TriadStore store TriadStore() spo_list [ [北京, 位于, 中国], [上海, 位于, 中国], ] ok store.store_data(user_id1001, spo_listspo_list, industrygeo) print(insert ok:, ok) store.cursor.execute( SELECT id, user_id, entity_one, relation, entity_two, created_at FROM triads WHERE user_id %s ORDER BY id DESC LIMIT 5, (1001,), ) for row in store.cursor.fetchall(): print(row) store.close()预期输出里created_at应该是datetime.datetime(2025, 1, 15, 14, 32, 7)这样的对象pymysql会自动把DATETIME转成 Python 的datetime。如果你看到的是字符串说明连接参数里没开转换检查charset和驱动版本。再验证一下时间是否落在合理区间import datetime store.cursor.execute( SELECT created_at FROM triads WHERE user_id %s, (1001,) ) rows store.cursor.fetchall() now datetime.datetime.now() for (created_at,) in rows: delta abs((now - created_at).total_seconds()) assert delta 60, ftime drift too large: {delta}s print(time check passed)这个断言能帮你发现时区偏移。如果delta是 28800 秒8 小时基本就是时区没对齐。如果你同时要调模型接口可以用同一个 Key 发一个验证请求确认通道可用。模型对话入口在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。用requests发一个最小请求import requests from config import Config resp requests.post( f{Config.TAOTOKEN_BASE_URL}/v1/chat/completions, headers{Authorization: fBearer {Config.TAOTOKEN_API_KEY}}, json{ model: gpt-4o-mini, messages: [{role: user, content: ping}], }, timeout30, ) print(resp.status_code, resp.json().get(choices, [{}])[0].get(message, {}).get(content))返回 200 且有内容说明 Key 和 base_url 配置正确。数据库写入和模型调用两条链路都验证过配置才算真正落地。5. 常见报错排查401、local proxy failed 与时间格式异常这一节按真实报错来。第一个高频错误是pymysql.err.OperationalError: (1045, Access denied for user)。这通常是.env里的密码写错或者 MySQL 用户没有对应数据库权限。排查顺序先用命令行mysql -u root -p确认密码能登录再检查config.py读到的值是不是空字符串。load_dotenv()默认从当前工作目录找.env如果你在子目录跑脚本可能读不到用load_dotenv(dotenv_path/abs/path/.env)指定绝对路径。第二个是Incorrect datetime value: for column created_at。这说明传进去的是空字符串。检查spo_list里每行的长度是否和 SQL 占位符数量一致。executemany不会帮你对齐列数少一个元素就会错位。建议在store_data里加一行断言for item in spo_list: assert len(item) 3, fspo item length mismatch: {item}第三个是模型调用侧的401 Unauthorized。这表示 Key 无效或没带上。检查Authorization头是不是Bearer sk-xxx格式中间有空格。如果 Key 是从控制台复制的注意别把首尾空格带进去用.strip()处理一下。Key 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 可以重新生成一个对比测试。第四个是local proxy failed或连接超时。这类报错通常和网络环境有关检查你的base_url是不是写成了https://taotoken.net/api/带尾斜杠有些客户端拼接路径时会出问题统一去掉尾斜杠。另外确认requests的timeout设了合理值默认不设会一直挂起。第五个是pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)。多半是表名或字段名用了 MySQL 保留字比如order、group。解决办法是用反引号包起来或者改名。建表时字段名尽量避开保留字。第六个是datetime相关AttributeError: module datetime has no attribute now。这是因为你import datetime后写成了datetime.now()正确写法是datetime.datetime.now()。或者你写from datetime import datetime那就可以直接datetime.now()。两种导入方式别混用。如果你用 Claude Code 或 Cline 这类工具做开发配置里需要同时填 Base URL、Key、Model ID 三件套。Base URL 用https://taotoken.net/apiKey 用控制台生成的Model ID 按文档里的模型名填。Coding Plan 入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 适合长期编码场景。Claude Code 的接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里有详细说明配置项名称以文档为准。排查时养成一个习惯先单独测数据库连接再单独测模型接口最后测组合链路。哪一步失败就锁定哪一步别一上来就怀疑整个系统。6. 把配置收敛成一套后续维护更省心走到这里你已经有了建表 SQL、pymysql插入代码、时间校验脚本和回读验证步骤。最后说一个实际项目里的做法把数据库配置和 API 配置都收进Config类所有模块只从这一个地方取。这样换环境时改.env就行不用满项目搜password。如果你后续要接更多模型或更多数据库可以按同样的模式扩展。API Key 和 base_url 统一走 TaoToken 通道模型对话、Coding Plan、接入文档都在官网能找到对应入口。数据库那边保持自己的连接池管理两边互不干扰但凭证读取逻辑一致。时间字段这块记住三个原则生成时用datetime.now()或 UTC 时间写入时用参数化查询让驱动转义回读时断言时间偏差在合理范围。做到这三点Incorrect datetime value和时区偏移基本不会再找上门。
返回列表