尧图网页设计 
1. 为什么你的 Python 连 MySQL 总是卡在第一步很多刚接触 Python 后端或者数据脚本的朋友第一次写pymysql.connect()的时候都会遇到一个很尴尬的局面代码明明照着教程敲的运行起来要么是Access denied要么是Cant connect to MySQL server要么就是插入了数据但查出来是空的。我自己刚开始做数据同步脚本那会儿光是一个连接参数就折腾了一下午后来才发现问题根本不在代码逻辑而在于连接配置的写法太随意了。这篇内容聚焦的就是 Python 操作 MySQL 的完整闭环连接、建表、插入、读取。我会把连接参数拆成可复制的 config 骨架再给一份settings.json示例让你不用把密码硬编码在脚本里。同时每一步都配上验证动作连接完立刻测、建完表立刻查、插入完立刻回读确保你跑通的是真实可用的读写流程而不是“看起来能跑”的代码。适合谁看如果你正在写数据采集脚本、做本地数据管理、或者想给一个小工具加个持久化存储这套流程可以直接拿去改。热词里提到的 python、mysql、数据库、连接、建表我都会用可复制的代码块覆盖到不玩虚的。另外提一句如果你后续想把模型调用和数据库操作串在一条链路里统一 Key 通道可以省掉不少管理成本这个后面会顺带说到。2. 前置准备TaoToken 统一 Key 与 MySQL 环境确认在写代码之前先把两件事确认清楚一是 MySQL 服务本身能连二是如果你打算把数据库操作和模型能力结合Key 的管理方式要提前定好。MySQL 这边你需要确认四个信息主机地址、端口默认 3306、用户名、密码以及你要操作的数据库名。本地开发通常是127.0.0.1如果你用的是云数据库地址就是服务商给的外网或内网地址。验证 MySQL 是否可达最直接的方式是用命令行mysql -h 127.0.0.1 -P 3306 -u root -p输入密码后能进到mysql提示符说明服务端没问题。如果这一步就报错先别急着写 Python先把 MySQL 服务起起来或者检查防火墙端口。关于统一 Key 通道如果你后续要在脚本里调用模型接口做数据处理建议把 Key 统一放在一个配置中心管理而不是散落在各个脚本里。TaoToken 的 API 入口是https://taotoken.net/apiKey 的创建和管理在控制台的 API Keys 页面。这样做的好处是数据库配置和模型 Key 都走同一套环境变量或配置文件迁移和排障都省事。需要提前拿好的东西MySQL 的连接四要素host、port、user、password目标数据库名没有的话先CREATE DATABASEPython 环境建议 3.8 以上安装驱动pip install pymysql如果你还没创建 Key可以到 API Keys 页面生成一个后面配置模型调用时会用到。数据库这边不需要 TaoToken 介入它只管模型通道两者是配合关系不是替代关系。3. 可复制配置config 骨架与 settings.json 示例硬编码密码是新手最容易踩的坑一旦脚本传到 Git 或者分享给别人密码就泄露了。正确做法是把连接信息抽到配置文件里代码只读配置。先看一个 Python 侧的 config 骨架我习惯用一个db_config.py来承载# db_config.py import json import os def load_db_config(pathsettings.json): if not os.path.exists(path): raise FileNotFoundError(f配置文件不存在: {path}) with open(path, r, encodingutf-8) as f: cfg json.load(f) required [host, port, user, password, database, charset] for key in required: if key not in cfg: raise KeyError(f配置缺少字段: {key}) return cfg对应的settings.json示例{ host: 127.0.0.1, port: 3306, user: root, password: your_password_here, database: test_db, charset: utf8mb4 }这里有几个参数值得说明。charset一定要写utf8mb4不要写utf8否则遇到 emoji 或者部分生僻字会报错。port是整数不要写成字符串否则 pymysql 在某些版本下会抛类型错误。database必须提前存在pymysql 不会帮你自动建库。如果你要把模型 Key 也统一管理可以在同一个settings.json里加一段{ host: 127.0.0.1, port: 3306, user: root, password: your_password_here, database: test_db, charset: utf8mb4, taotoken: { api_base: https://taotoken.net/api, api_key: sk-xxxxxxxx } }这样数据库和模型通道的配置就在一个文件里脚本读一次配置就能拿到全部依赖。注意settings.json要加到.gitignore别提交到仓库。连接函数的封装建议加上超时和自动重连参数import pymysql from db_config import load_db_config def get_connection(): cfg load_db_config() return pymysql.connect( hostcfg[host], portcfg[port], usercfg[user], passwordcfg[password], databasecfg[database], charsetcfg[charset], connect_timeout10, cursorclasspymysql.cursors.DictCursor )cursorclass用DictCursor的好处是查询结果直接是字典后面读取的时候不用靠下标猜字段可读性高很多。4. 四步实操连接、建表、插入、读取全流程4.1 连接测试先确认能握手再往下走连接是所有操作的前提这一步不过后面全是空谈。写一个独立的测试脚本import pymysql from db_config import load_db_config def test_connection(): cfg load_db_config() try: conn pymysql.connect( hostcfg[host], portcfg[port], usercfg[user], passwordcfg[password], databasecfg[database], charsetcfg[charset], connect_timeout10 ) with conn.cursor() as cursor: cursor.execute(SELECT VERSION()) version cursor.fetchone() print(连接成功MySQL 版本:, version) conn.close() return True except pymysql.err.OperationalError as e: print(连接失败错误码:, e.args[0]) print(错误信息:, e.args[1]) return False if __name__ __main__: test_connection()运行后如果打印出版本号说明连接参数全部正确。如果报1045是用户名或密码错报2003是 host 或 port 不通报1049是数据库不存在。这三个错误码覆盖了九成以上的连接问题。4.2 建表用 IF NOT EXISTS 保证可重复执行建表语句建议写成可重复执行的这样脚本跑多次也不会报“表已存在”。我用一个口罩数据表做例子字段包含城市、需求量、供应量、时间def create_table(): conn get_connection() sql CREATE TABLE IF NOT EXISTS mask_stats ( id INT AUTO_INCREMENT PRIMARY KEY, city VARCHAR(64) NOT NULL, required INT NOT NULL, supply INT NOT NULL, stat_time INT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; with conn.cursor() as cursor: cursor.execute(sql) conn.commit() conn.close() print(建表完成)这里比原始写法多了几个细节加了自增主键id方便后续按行定位字段类型用VARCHAR而不是TEXT因为城市名长度可控VARCHAR索引效率更高加了created_at默认时间戳排查数据什么时候写入的很方便。建表校验动作执行完后用SHOW TABLES LIKE mask_stats确认表存在再用DESC mask_stats看字段结构是否符合预期。4.3 插入参数化查询防注入commit 不能漏插入数据最容易犯两个错一是用字符串拼接 SQL二是忘了commit。参数化查询用%s占位把变量作为第二个参数传给executedef insert_data(city, required, supply, stat_time): conn get_connection() sql INSERT INTO mask_stats (city, required, supply, stat_time) VALUES (%s, %s, %s, %s) with conn.cursor() as cursor: cursor.execute(sql, (city, required, supply, stat_time)) conn.commit() conn.close() print(f插入成功: {city}) if __name__ __main__: insert_data(北京, 213214, 892375, 2020) insert_data(上海, 187654, 765432, 2020)注意%s是 pymysql 的占位符不管字段是字符串还是数字都用%s驱动会自动处理类型转换。不要写成%d也不要手动加引号。批量插入可以用executemany效率比循环单条插入高很多def insert_batch(rows): conn get_connection() sql INSERT INTO mask_stats (city, required, supply, stat_time) VALUES (%s, %s, %s, %s) with conn.cursor() as cursor: cursor.executemany(sql, rows) conn.commit() conn.close() print(f批量插入 {len(rows)} 条完成)commit是必须的不提交的话数据只在当前连接的事务里连接一关就没了。这是新手最常踩的坑代码不报错但数据查不到八成就是漏了commit。4.4 读取fetchall 拿列表DictCursor 拿字典读取用SELECT配合fetchall()返回的是一个列表。如果用默认游标每行是元组靠下标取值如果用DictCursor每行是字典靠字段名取值def read_data(): conn get_connection() sql SELECT city, required, stat_time FROM mask_stats ORDER BY id with conn.cursor() as cursor: cursor.execute(sql) rows cursor.fetchall() conn.close() for row in rows: print(f城市{row[city]}, 需求{row[required]}, 时间{row[stat_time]}) return rows如果数据量很大不要一次性fetchall用fetchone逐行读或者用SSCursor流式游标避免内存爆掉。小数据量场景fetchall完全够用。读取校验动作插入后立刻调用read_data()看输出里有没有刚插入的那条记录。如果有说明插入和读取链路都通了如果没有回去检查commit有没有执行。5. 本篇常见报错排查5.1 pymysql.err.OperationalError: (1045, Access denied)这个错误是认证失败原因通常是密码错、用户名错或者该用户没有从当前主机连接的权限。MySQL 的用户是userhost的形式rootlocalhost和root%是两个不同的用户。如果你从远程连接需要确认用户是否有对应 host 的授权。排查方式用命令行mysql -h host -u user -p试一下命令行能进说明 Python 侧参数写错了。5.2 pymysql.err.OperationalError: (2003, Cant connect to MySQL server)连接不通可能是 MySQL 没启动、端口不对、防火墙拦截或者 bind-address 只监听了127.0.0.1。本地开发先确认3306端口在监听netstat -an | grep 3306。云数据库的话检查安全组有没有放行你的 IP。5.3 pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)SQL 语法错常见于建表语句字段类型写错、逗号多写或少写、引号不匹配。把 SQL 打印出来复制到 MySQL 命令行里执行一遍能快速定位。另外注意%s占位符只在execute的参数化查询里用不要用在建表语句里。5.4 插入成功但查询为空九成是漏了conn.commit()。pymysql 默认不开自动提交不 commit 的话事务不落盘。另一个可能是查询用了不同的数据库连接连到了别的库。确认settings.json里的database字段和插入时一致。5.5 中文乱码建表时CHARSET写成了utf8而不是utf8mb4或者连接参数里没指定charset。两边都统一成utf8mb4就能解决。如果表已经建好了可以ALTER TABLE mask_stats CONVERT TO CHARACTER SET utf8mb4;改一下。5.6 连接数过多导致 Too many connections脚本里每次操作都新建连接、用完不关连接数会堆积。正确做法是用with上下文管理或者用连接池。简单场景下确保每个函数最后conn.close()复杂场景建议引入DBUtils的PooledDB。6. 把数据库读写接进你的工作流跑通上面四步之后你手里就有了一套可复用的 MySQL 读写骨架。接下来可以做的事把insert_data接到你的数据采集脚本里把read_data接到报表生成或者接口返回里。配置全部走settings.json换环境只改配置不改代码。如果你后续要在脚本里调用模型做数据清洗、摘要或者分类Key 的管理建议和数据库配置放在同一层。TaoToken 的 API 入口是https://taotoken.net/apiKey 在控制台的 API Keys 页面创建接入文档里有各语言的调用示例。模型对话调试可以在模型对话页面直接试长期跑编码类任务的话 Coding Plan 更划算。数据库这边没有捷径连接、建表、插入、读取四步走通剩下的就是按业务加字段、加索引、加查询条件。先把这套骨架跑起来比看十篇教程都管用。 Python 操作 MySQL 数据库：连接、建表、插入、读取的完整配置与验证
1. 为什么你的 Python 连 MySQL 总是卡在第一步很多刚接触 Python 后端或者数据脚本的朋友第一次写pymysql.connect()的时候都会遇到一个很尴尬的局面代码明明照着教程敲的运行起来要么是Access denied要么是Cant connect to MySQL server要么就是插入了数据但查出来是空的。我自己刚开始做数据同步脚本那会儿光是一个连接参数就折腾了一下午后来才发现问题根本不在代码逻辑而在于连接配置的写法太随意了。这篇内容聚焦的就是 Python 操作 MySQL 的完整闭环连接、建表、插入、读取。我会把连接参数拆成可复制的 config 骨架再给一份settings.json示例让你不用把密码硬编码在脚本里。同时每一步都配上验证动作连接完立刻测、建完表立刻查、插入完立刻回读确保你跑通的是真实可用的读写流程而不是“看起来能跑”的代码。适合谁看如果你正在写数据采集脚本、做本地数据管理、或者想给一个小工具加个持久化存储这套流程可以直接拿去改。热词里提到的 python、mysql、数据库、连接、建表我都会用可复制的代码块覆盖到不玩虚的。另外提一句如果你后续想把模型调用和数据库操作串在一条链路里统一 Key 通道可以省掉不少管理成本这个后面会顺带说到。2. 前置准备TaoToken 统一 Key 与 MySQL 环境确认在写代码之前先把两件事确认清楚一是 MySQL 服务本身能连二是如果你打算把数据库操作和模型能力结合Key 的管理方式要提前定好。MySQL 这边你需要确认四个信息主机地址、端口默认 3306、用户名、密码以及你要操作的数据库名。本地开发通常是127.0.0.1如果你用的是云数据库地址就是服务商给的外网或内网地址。验证 MySQL 是否可达最直接的方式是用命令行mysql -h 127.0.0.1 -P 3306 -u root -p输入密码后能进到mysql提示符说明服务端没问题。如果这一步就报错先别急着写 Python先把 MySQL 服务起起来或者检查防火墙端口。关于统一 Key 通道如果你后续要在脚本里调用模型接口做数据处理建议把 Key 统一放在一个配置中心管理而不是散落在各个脚本里。TaoToken 的 API 入口是https://taotoken.net/apiKey 的创建和管理在控制台的 API Keys 页面。这样做的好处是数据库配置和模型 Key 都走同一套环境变量或配置文件迁移和排障都省事。需要提前拿好的东西MySQL 的连接四要素host、port、user、password目标数据库名没有的话先CREATE DATABASEPython 环境建议 3.8 以上安装驱动pip install pymysql如果你还没创建 Key可以到 API Keys 页面生成一个后面配置模型调用时会用到。数据库这边不需要 TaoToken 介入它只管模型通道两者是配合关系不是替代关系。3. 可复制配置config 骨架与 settings.json 示例硬编码密码是新手最容易踩的坑一旦脚本传到 Git 或者分享给别人密码就泄露了。正确做法是把连接信息抽到配置文件里代码只读配置。先看一个 Python 侧的 config 骨架我习惯用一个db_config.py来承载# db_config.py import json import os def load_db_config(pathsettings.json): if not os.path.exists(path): raise FileNotFoundError(f配置文件不存在: {path}) with open(path, r, encodingutf-8) as f: cfg json.load(f) required [host, port, user, password, database, charset] for key in required: if key not in cfg: raise KeyError(f配置缺少字段: {key}) return cfg对应的settings.json示例{ host: 127.0.0.1, port: 3306, user: root, password: your_password_here, database: test_db, charset: utf8mb4 }这里有几个参数值得说明。charset一定要写utf8mb4不要写utf8否则遇到 emoji 或者部分生僻字会报错。port是整数不要写成字符串否则 pymysql 在某些版本下会抛类型错误。database必须提前存在pymysql 不会帮你自动建库。如果你要把模型 Key 也统一管理可以在同一个settings.json里加一段{ host: 127.0.0.1, port: 3306, user: root, password: your_password_here, database: test_db, charset: utf8mb4, taotoken: { api_base: https://taotoken.net/api, api_key: sk-xxxxxxxx } }这样数据库和模型通道的配置就在一个文件里脚本读一次配置就能拿到全部依赖。注意settings.json要加到.gitignore别提交到仓库。连接函数的封装建议加上超时和自动重连参数import pymysql from db_config import load_db_config def get_connection(): cfg load_db_config() return pymysql.connect( hostcfg[host], portcfg[port], usercfg[user], passwordcfg[password], databasecfg[database], charsetcfg[charset], connect_timeout10, cursorclasspymysql.cursors.DictCursor )cursorclass用DictCursor的好处是查询结果直接是字典后面读取的时候不用靠下标猜字段可读性高很多。4. 四步实操连接、建表、插入、读取全流程4.1 连接测试先确认能握手再往下走连接是所有操作的前提这一步不过后面全是空谈。写一个独立的测试脚本import pymysql from db_config import load_db_config def test_connection(): cfg load_db_config() try: conn pymysql.connect( hostcfg[host], portcfg[port], usercfg[user], passwordcfg[password], databasecfg[database], charsetcfg[charset], connect_timeout10 ) with conn.cursor() as cursor: cursor.execute(SELECT VERSION()) version cursor.fetchone() print(连接成功MySQL 版本:, version) conn.close() return True except pymysql.err.OperationalError as e: print(连接失败错误码:, e.args[0]) print(错误信息:, e.args[1]) return False if __name__ __main__: test_connection()运行后如果打印出版本号说明连接参数全部正确。如果报1045是用户名或密码错报2003是 host 或 port 不通报1049是数据库不存在。这三个错误码覆盖了九成以上的连接问题。4.2 建表用 IF NOT EXISTS 保证可重复执行建表语句建议写成可重复执行的这样脚本跑多次也不会报“表已存在”。我用一个口罩数据表做例子字段包含城市、需求量、供应量、时间def create_table(): conn get_connection() sql CREATE TABLE IF NOT EXISTS mask_stats ( id INT AUTO_INCREMENT PRIMARY KEY, city VARCHAR(64) NOT NULL, required INT NOT NULL, supply INT NOT NULL, stat_time INT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; with conn.cursor() as cursor: cursor.execute(sql) conn.commit() conn.close() print(建表完成)这里比原始写法多了几个细节加了自增主键id方便后续按行定位字段类型用VARCHAR而不是TEXT因为城市名长度可控VARCHAR索引效率更高加了created_at默认时间戳排查数据什么时候写入的很方便。建表校验动作执行完后用SHOW TABLES LIKE mask_stats确认表存在再用DESC mask_stats看字段结构是否符合预期。4.3 插入参数化查询防注入commit 不能漏插入数据最容易犯两个错一是用字符串拼接 SQL二是忘了commit。参数化查询用%s占位把变量作为第二个参数传给executedef insert_data(city, required, supply, stat_time): conn get_connection() sql INSERT INTO mask_stats (city, required, supply, stat_time) VALUES (%s, %s, %s, %s) with conn.cursor() as cursor: cursor.execute(sql, (city, required, supply, stat_time)) conn.commit() conn.close() print(f插入成功: {city}) if __name__ __main__: insert_data(北京, 213214, 892375, 2020) insert_data(上海, 187654, 765432, 2020)注意%s是 pymysql 的占位符不管字段是字符串还是数字都用%s驱动会自动处理类型转换。不要写成%d也不要手动加引号。批量插入可以用executemany效率比循环单条插入高很多def insert_batch(rows): conn get_connection() sql INSERT INTO mask_stats (city, required, supply, stat_time) VALUES (%s, %s, %s, %s) with conn.cursor() as cursor: cursor.executemany(sql, rows) conn.commit() conn.close() print(f批量插入 {len(rows)} 条完成)commit是必须的不提交的话数据只在当前连接的事务里连接一关就没了。这是新手最常踩的坑代码不报错但数据查不到八成就是漏了commit。4.4 读取fetchall 拿列表DictCursor 拿字典读取用SELECT配合fetchall()返回的是一个列表。如果用默认游标每行是元组靠下标取值如果用DictCursor每行是字典靠字段名取值def read_data(): conn get_connection() sql SELECT city, required, stat_time FROM mask_stats ORDER BY id with conn.cursor() as cursor: cursor.execute(sql) rows cursor.fetchall() conn.close() for row in rows: print(f城市{row[city]}, 需求{row[required]}, 时间{row[stat_time]}) return rows如果数据量很大不要一次性fetchall用fetchone逐行读或者用SSCursor流式游标避免内存爆掉。小数据量场景fetchall完全够用。读取校验动作插入后立刻调用read_data()看输出里有没有刚插入的那条记录。如果有说明插入和读取链路都通了如果没有回去检查commit有没有执行。5. 本篇常见报错排查5.1 pymysql.err.OperationalError: (1045, Access denied)这个错误是认证失败原因通常是密码错、用户名错或者该用户没有从当前主机连接的权限。MySQL 的用户是userhost的形式rootlocalhost和root%是两个不同的用户。如果你从远程连接需要确认用户是否有对应 host 的授权。排查方式用命令行mysql -h host -u user -p试一下命令行能进说明 Python 侧参数写错了。5.2 pymysql.err.OperationalError: (2003, Cant connect to MySQL server)连接不通可能是 MySQL 没启动、端口不对、防火墙拦截或者 bind-address 只监听了127.0.0.1。本地开发先确认3306端口在监听netstat -an | grep 3306。云数据库的话检查安全组有没有放行你的 IP。5.3 pymysql.err.ProgrammingError: (1064, You have an error in your SQL syntax)SQL 语法错常见于建表语句字段类型写错、逗号多写或少写、引号不匹配。把 SQL 打印出来复制到 MySQL 命令行里执行一遍能快速定位。另外注意%s占位符只在execute的参数化查询里用不要用在建表语句里。5.4 插入成功但查询为空九成是漏了conn.commit()。pymysql 默认不开自动提交不 commit 的话事务不落盘。另一个可能是查询用了不同的数据库连接连到了别的库。确认settings.json里的database字段和插入时一致。5.5 中文乱码建表时CHARSET写成了utf8而不是utf8mb4或者连接参数里没指定charset。两边都统一成utf8mb4就能解决。如果表已经建好了可以ALTER TABLE mask_stats CONVERT TO CHARACTER SET utf8mb4;改一下。5.6 连接数过多导致 Too many connections脚本里每次操作都新建连接、用完不关连接数会堆积。正确做法是用with上下文管理或者用连接池。简单场景下确保每个函数最后conn.close()复杂场景建议引入DBUtils的PooledDB。6. 把数据库读写接进你的工作流跑通上面四步之后你手里就有了一套可复用的 MySQL 读写骨架。接下来可以做的事把insert_data接到你的数据采集脚本里把read_data接到报表生成或者接口返回里。配置全部走settings.json换环境只改配置不改代码。如果你后续要在脚本里调用模型做数据清洗、摘要或者分类Key 的管理建议和数据库配置放在同一层。TaoToken 的 API 入口是https://taotoken.net/apiKey 在控制台的 API Keys 页面创建接入文档里有各语言的调用示例。模型对话调试可以在模型对话页面直接试长期跑编码类任务的话 Coding Plan 更划算。数据库这边没有捷径连接、建表、插入、读取四步走通剩下的就是按业务加字段、加索引、加查询条件。先把这套骨架跑起来比看十篇教程都管用。