尧图网页设计 
1. 为什么你的 pymysql 连接总在第一步就报错很多人第一次写 Python 连 MySQL代码照着教程敲完运行直接甩一个pymysql.err.OperationalError: (1045, Access denied for user rootlocalhost)然后就开始怀疑人生。其实问题往往不在代码而在连接参数没对齐主机地址、端口、用户名、密码、字符集、目标库这六个里错一个就连不上。pymysql 是纯 Python 实现的 MySQL 客户端驱动不需要编译 C 扩展pip install pymysql就能用适合快速做数据读写、脚本批处理、后台小工具这类任务。它适合谁适合刚学完 Python 基础语法、想把手里的列表和字典真正落到数据库里的同学也适合需要写定时任务把数据写进 MySQL 的后端初学者。我试过把连接参数写死在代码里换台机器就崩后来改成配置模板才省心。这一篇就按「连接建立 → 参数配置 → 增删改查 → 异常处理 → 结果验证」的完整链路走一遍每一步都给可直接复制的代码最后用一个银行账户的小项目把查询、取钱、存钱串起来。你跟着敲完至少能搞清楚三件事连接对象和游标到底啥关系、commit为什么必须写、fetchone和fetchall的指针行为差在哪。先说清楚一个概念避免后面懵。pymysql.connect()返回的是一个连接对象它代表你和 MySQL 服务器之间的一条通道而cursor()返回的游标对象才是真正执行 SQL、拿结果的工具。你可以把连接想成电话线游标想成话筒——线接通了你还得拿起话筒才能说话。很多人把这两个混在一起导致cursor用完不关、连接泄漏跑久了就报Too many connections。环境准备只有一步命令行执行pip install pymysql装完可以顺手验证版本确认装到了当前解释器python -c import pymysql; print(pymysql.__version__)能打印出版本号说明驱动就位。接下来所有代码都基于这个前提。数据库本身你得先有一个能登录的 MySQL 实例本地装一个或者用已有的都行记住 host、port、user、password 这四个值后面配置模板直接填。2. 连接参数怎么配才不踩坑pymysql.connect 完整参数模板连接这一步是整个流程的地基参数配错后面全白搭。下面这份模板你可以直接复制把值换成自己的即可。我把它写成字典再解包传入好处是配置和调用分离改起来只动一处。import pymysql DB_CONFIG { host: 127.0.0.1, # 数据库地址本地就是 127.0.0.1 或 localhost port: 3306, # MySQL 默认端口 user: root, # 用户名 password: your_pwd, # 你的密码 charset: utf8mb4, # 字符集强烈建议 utf8mb4 cursorclass: pymysql.cursors.DictCursor, # 让结果以字典返回 autocommit: False, # 手动提交写操作更可控 } def get_conn(dbNone): cfg dict(DB_CONFIG) if db: cfg[database] db return pymysql.connect(**cfg)这里有几个参数值得单独说。charset用utf8mb4而不是utf8因为 MySQL 里的utf8其实是残缺的三字节实现存 emoji 或者某些生僻字会报Incorrect string valueutf8mb4才是真正的四字节 UTF-8。cursorclass设成DictCursor后fetchone()返回的是{SNO: 200303016, SNAME: 王智刚}这种字典比元组下标好读太多字段顺序变了也不怕。autocommit默认是False意味着你执行insert、update、delete之后必须手动commit()否则数据不会真正落库——这是新手最容易忽略的一点代码跑完没报错去数据库一查啥都没有八成就是忘了提交。连接建立后创建游标conn get_conn() cursor conn.cursor() cursor.execute(select version();) print(cursor.fetchone()) cursor.close() conn.close()如果这一步能打印出 MySQL 版本说明连接链路完全通了。注意close()的顺序先关游标再关连接反过来虽然多数情况不报错但规范上应该由内到外释放。关于建库建表有个细节要提醒create schema和create database在 MySQL 里是等价的。建库时指定字符集cursor.execute(create database if not exists bank default charset utf8mb4;)加if not exists能避免重复执行时报database exists错误脚本可以反复跑。建表时字段类型的选择也有讲究char和varchar的区别经常被问到char(10)是定长存abc也占 10 个字节后面补 7 个空格varchar(10)是变长存abc只占 3 个字节10 是上限。所以像账号、密码这种长度固定的用char姓名、地址这种长度不定的用varchar。金额字段用decimal(10,2)表示总共 10 位、其中小数 2 位能精确表示钱千万别用float浮点数算钱会出现0.1 0.2 ! 0.3的经典问题。3. 增删改查全流程一份可复制的 settings 配置与 CRUD 代码这一节是核心把建表、插入、查询、更新、删除五个动作全部走一遍。先给一份可以直接落地的配置片段我习惯把它单独放一个db_config.py其他脚本import进来用。# db_config.py import pymysql DB_CONFIG { host: 127.0.0.1, port: 3306, user: root, password: your_pwd, charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor, autocommit: False, } def get_conn(dbNone): cfg dict(DB_CONFIG) if db: cfg[database] db return pymysql.connect(**cfg)建库建表并插入初始数据这段可以当初始化脚本跑一次from db_config import get_conn CREATE_TABLE_SQL create table if not exists account( account_id varchar(20) not null, account_passwd char(6) not null, money decimal(10,2) default 0.00, primary key(account_id) ) default charsetutf8mb4; INIT_DATA_SQL insert into account(account_id, account_passwd, money) values(001,123456,1000.00),(002,456789,5000.00) on duplicate key update moneyvalues(money); def init(): conn get_conn(bank) cursor conn.cursor() try: cursor.execute(CREATE_TABLE_SQL) cursor.execute(INIT_DATA_SQL) conn.commit() print(初始化成功) except Exception as e: conn.rollback() print(初始化失败, e) finally: cursor.close() conn.close() if __name__ __main__: init()注意insert里用了on duplicate key update这样脚本重复执行不会因为主键冲突报错而是把金额重置回初始值方便反复测试。查询是最常用的操作重点讲fetchone和fetchall的指针行为。游标内部维护一个位置指针fetchone()取一条指针就往后移一格fetchall()取剩余全部。所以如果你先fetchone()再fetchall()第二条开始的数据才会被fetchall拿到第一条已经消费掉了。def query_all(): conn get_conn(bank) cursor conn.cursor() try: cursor.execute(select * from account;) rows cursor.fetchall() for row in rows: print(row[account_id], row[account_passwd], row[money]) return rows finally: cursor.close() conn.close()带参数的查询一定要用占位符%s不要用字符串拼接否则会有 SQL 注入风险。pymysql 的占位符统一是%s哪怕参数是数字也用%sdef query_by_id(account_id): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( select * from account where account_id%s;, (account_id,) ) return cursor.fetchone() finally: cursor.close() conn.close()更新和删除同样要记得commit并且用rowcount判断是否真的影响到了行def update_money(account_id, delta): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( update account set moneymoney%s where account_id%s;, (delta, account_id) ) if cursor.rowcount 1: conn.commit() return True conn.rollback() return False except Exception as e: conn.rollback() print(更新失败, e) return False finally: cursor.close() conn.close()rowcount返回的是本次执行影响的行数更新成功就是 1账号不存在就是 0。这个判断很重要能防止「以为更新了其实没匹配到」的假成功。删除同理把 SQL 换成delete from account where account_id%s即可。4. 验证连接与结果怎么确认数据真的写进去了代码跑完不报错不代表数据真的进了库。验证分两层连接层验证和结果层验证。连接层验证最简单执行一条select 1或者select version()能返回结果就说明通道没问题def check_conn(): try: conn get_conn() cursor conn.cursor() cursor.execute(select 1 as ok;) result cursor.fetchone() cursor.close() conn.close() return result[ok] 1 except Exception as e: print(连接失败, e) return False结果层验证最直接的办法是写完之后立刻查一次对比预期。比如插入一条记录后用主键查回来看字段值是否一致def verify_insert(account_id): row query_by_id(account_id) if row: print(查到记录, row) return True print(没查到插入可能没提交) return False还有一个容易被忽略的点decimal类型从数据库取出来是Decimal对象不是float。直接print没问题但如果你要参与计算或者格式化得注意类型转换。比如保留两位小数import decimal money row[money] # Decimal(1000.00) formatted money.quantize(decimal.Decimal(0.00)) print(str(formatted)) # 1000.00quantize做四舍五入并固定小数位比round()更适合金额场景因为round()对Decimal的行为和浮点数不一样容易出意外。再给一个完整的验证脚本把连接、写入、读回三步串起来跑通它你就有了一个可复用的自检模板from db_config import get_conn def full_check(): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( insert into account(account_id, account_passwd, money) values(%s,%s,%s) on duplicate key update moneyvalues(money);, (003, 111111, 888.88) ) conn.commit() cursor.execute(select * from account where account_id%s;, (003,)) row cursor.fetchone() assert row is not None, 写入后查不到记录 assert str(row[money]) 888.88, 金额不一致 print(全流程验证通过, row) except Exception as e: conn.rollback() print(验证失败, e) finally: cursor.close() conn.close() if __name__ __main__: full_check()assert在这里当断言用条件不满足直接抛异常适合做自检。生产代码里可以换成if not ...: raise效果一样。5. 常见报错逐个排查1045、2003、Incorrect string value 怎么解报错不可怕怕的是看不懂。这一节把 pymysql 最常见的几个错误对照着讲每个都给原因和改法。1045 Access denied用户名或密码不对或者该用户没有从当前主机登录的权限。先确认密码有没有写错再确认host参数——localhost和127.0.0.1在 MySQL 权限体系里可能是两个不同的来源。如果密码里有特殊字符注意别被配置文件转义了。2003 Cant connect to MySQL server连不上服务器通常是 MySQL 服务没启动、端口不对、或者防火墙拦了。先在命令行mysql -u root -p试试能不能登能登说明服务正常问题在 Python 侧的 host/port 配置。1049 Unknown database连接时指定的database不存在。要么先建库要么连接时不传database连上后再cursor.execute(use bank;)。Incorrect string value字符集不匹配多半是用了utf8存了四字节字符。把连接和建表的字符集都改成utf8mb4已有表可以用alter table account convert to charset utf8mb4;转换。pymysql.err.ProgrammingError: (1064, ...)SQL 语法错误常见于占位符写错。记住 pymysql 只认%s不认?也不认:name参数要用元组或列表传单个参数也要写成(value,)带逗号。数据写不进去但没报错九成是忘了commit()。autocommitFalse时所有写操作都在事务里不提交就回滚。要么每次写完手动commit要么连接时设autocommitTrue但后者不利于批量操作时统一回滚建议还是手动控制。Too many connections连接没关泄漏了。检查每个get_conn()是否都有对应的close()最好用try/finally包起来或者用上下文管理器。pymysql 的连接对象支持with语法with get_conn(bank) as conn: with conn.cursor() as cursor: cursor.execute(select * from account;) print(cursor.fetchall())用with能自动关闭省心不少。Lost connection to MySQL server during query查询太大或者超时。可以调大connect_timeout和read_timeout或者分页查询别一次性fetchall几十万行。把这些报错和对应改法整理成一张对照表出问题时先查表再动手报错关键字大概率原因处理方向1045 Access denied账号密码错 / host 不匹配核对 user、password、host2003 Cant connect服务未启动 / 端口错命令行先验证能登1049 Unknown database库不存在先建库或去掉 database 参数Incorrect string value字符集不是 utf8mb4连接和建表都改 utf8mb41064 语法错误占位符写错统一用 %s参数传元组写不进无报错忘了 commit手动提交或开 autocommitToo many connections连接泄漏try/finally 或 with 管理排查顺序建议从外到内先确认 MySQL 服务本身能连再确认 Python 连接参数最后才看 SQL 语句。这样能快速定位问题在哪一层不至于一上来就改代码。6. 把读写串成一个能跑的小项目银行账户管理实战前面都是零件这一节把它们组装成一个完整可运行的程序。功能就四个查询余额、取钱、存钱、退出。核心逻辑是登录校验加金额增减涉及update和事务提交正好把前面讲的知识点全用上。先看账户类把登录检查和金额操作封装进去import decimal from db_config import get_conn class Account: def __init__(self, account_id, account_passwd): self.account_id account_id self.account_passwd account_passwd def check_account(self): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( select * from account where account_id%s and account_passwd%s;, (self.account_id, self.account_passwd) ) return cursor.fetchone() is not None finally: cursor.close() conn.close() def query_money(self): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( select money from account where account_id%s and account_passwd%s;, (self.account_id, self.account_passwd) ) row cursor.fetchone() if row and row[money] is not None: return str(row[money].quantize(decimal.Decimal(0.00))) return 0.00 finally: cursor.close() conn.close() def reduce_money(self, amount): conn get_conn(bank) cursor conn.cursor() try: has_money decimal.Decimal(self.query_money()) amount decimal.Decimal(amount) if amount has_money: cursor.execute( update account set moneymoney-%s where account_id%s and account_passwd%s;, (amount, self.account_id, self.account_passwd) ) if cursor.rowcount 1: conn.commit() return True conn.rollback() return False print(余额不足) return False except Exception as e: conn.rollback() print(取款异常, e) return False finally: cursor.close() conn.close() def add_money(self, amount): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( update account set moneymoney%s where account_id%s and account_passwd%s;, (decimal.Decimal(amount), self.account_id, self.account_passwd) ) if cursor.rowcount 1: conn.commit() return True conn.rollback() return False except Exception as e: conn.rollback() print(存款异常, e) return False finally: cursor.close() conn.close()主流程负责交互def main(): account_id input(请输入账号) account_passwd input(请输入密码) account Account(account_id, account_passwd) if not account.check_account(): print(账号或密码错误) return menu 请输入操作\n1、查询余额\n2、取钱\n3、存钱\n4、取卡\n choose input(menu) while choose ! 4: if choose 1: print(您的余额是 %s 元 % account.query_money()) elif choose 2: amount input(您的余额是 %s 元请输入取款金额 % account.query_money()) if account.reduce_money(amount): print(取款成功余额 %s 元 % account.query_money()) else: print(取款失败) elif choose 3: amount input(请输入存款金额) if account.add_money(amount): print(存款成功余额 %s 元 % account.query_money()) else: print(存款失败) else: print(无效操作) choose input(menu) print(谢谢使用) if __name__ __main__: main()跑起来的效果输入账号001、密码123456登录成功选 1 显示余额 1000.00选 2 输入 200余额变 800.00选 3 输入 500余额变 1300.00选 4 退出。整个过程数据实时落库重新登录余额还在。这里有几个设计上的取舍值得说。金额全程用Decimal处理避免浮点误差每次操作都独立开关连接虽然开销略大但避免了长连接超时和状态污染对初学者更友好update后用rowcount判断是否命中命中才commit否则rollback保证事务一致性。如果你想优化可以把连接复用起来或者引入连接池但那是下一步的事了。最后留个练习方向给账户表加一个交易流水表每次取钱存钱都往流水表插一条记录字段包括账号、操作类型、金额、时间。这样你就能把insert和update放在同一个事务里进一步体会commit和rollback的作用。写代码这件事看十遍不如自己敲一遍把上面的代码复制到本地跑通再按自己的需求改比什么都强。 Python 连接 MySQL 数据库：用 pymysql 打通数据读写全流程
1. 为什么你的 pymysql 连接总在第一步就报错很多人第一次写 Python 连 MySQL代码照着教程敲完运行直接甩一个pymysql.err.OperationalError: (1045, Access denied for user rootlocalhost)然后就开始怀疑人生。其实问题往往不在代码而在连接参数没对齐主机地址、端口、用户名、密码、字符集、目标库这六个里错一个就连不上。pymysql 是纯 Python 实现的 MySQL 客户端驱动不需要编译 C 扩展pip install pymysql就能用适合快速做数据读写、脚本批处理、后台小工具这类任务。它适合谁适合刚学完 Python 基础语法、想把手里的列表和字典真正落到数据库里的同学也适合需要写定时任务把数据写进 MySQL 的后端初学者。我试过把连接参数写死在代码里换台机器就崩后来改成配置模板才省心。这一篇就按「连接建立 → 参数配置 → 增删改查 → 异常处理 → 结果验证」的完整链路走一遍每一步都给可直接复制的代码最后用一个银行账户的小项目把查询、取钱、存钱串起来。你跟着敲完至少能搞清楚三件事连接对象和游标到底啥关系、commit为什么必须写、fetchone和fetchall的指针行为差在哪。先说清楚一个概念避免后面懵。pymysql.connect()返回的是一个连接对象它代表你和 MySQL 服务器之间的一条通道而cursor()返回的游标对象才是真正执行 SQL、拿结果的工具。你可以把连接想成电话线游标想成话筒——线接通了你还得拿起话筒才能说话。很多人把这两个混在一起导致cursor用完不关、连接泄漏跑久了就报Too many connections。环境准备只有一步命令行执行pip install pymysql装完可以顺手验证版本确认装到了当前解释器python -c import pymysql; print(pymysql.__version__)能打印出版本号说明驱动就位。接下来所有代码都基于这个前提。数据库本身你得先有一个能登录的 MySQL 实例本地装一个或者用已有的都行记住 host、port、user、password 这四个值后面配置模板直接填。2. 连接参数怎么配才不踩坑pymysql.connect 完整参数模板连接这一步是整个流程的地基参数配错后面全白搭。下面这份模板你可以直接复制把值换成自己的即可。我把它写成字典再解包传入好处是配置和调用分离改起来只动一处。import pymysql DB_CONFIG { host: 127.0.0.1, # 数据库地址本地就是 127.0.0.1 或 localhost port: 3306, # MySQL 默认端口 user: root, # 用户名 password: your_pwd, # 你的密码 charset: utf8mb4, # 字符集强烈建议 utf8mb4 cursorclass: pymysql.cursors.DictCursor, # 让结果以字典返回 autocommit: False, # 手动提交写操作更可控 } def get_conn(dbNone): cfg dict(DB_CONFIG) if db: cfg[database] db return pymysql.connect(**cfg)这里有几个参数值得单独说。charset用utf8mb4而不是utf8因为 MySQL 里的utf8其实是残缺的三字节实现存 emoji 或者某些生僻字会报Incorrect string valueutf8mb4才是真正的四字节 UTF-8。cursorclass设成DictCursor后fetchone()返回的是{SNO: 200303016, SNAME: 王智刚}这种字典比元组下标好读太多字段顺序变了也不怕。autocommit默认是False意味着你执行insert、update、delete之后必须手动commit()否则数据不会真正落库——这是新手最容易忽略的一点代码跑完没报错去数据库一查啥都没有八成就是忘了提交。连接建立后创建游标conn get_conn() cursor conn.cursor() cursor.execute(select version();) print(cursor.fetchone()) cursor.close() conn.close()如果这一步能打印出 MySQL 版本说明连接链路完全通了。注意close()的顺序先关游标再关连接反过来虽然多数情况不报错但规范上应该由内到外释放。关于建库建表有个细节要提醒create schema和create database在 MySQL 里是等价的。建库时指定字符集cursor.execute(create database if not exists bank default charset utf8mb4;)加if not exists能避免重复执行时报database exists错误脚本可以反复跑。建表时字段类型的选择也有讲究char和varchar的区别经常被问到char(10)是定长存abc也占 10 个字节后面补 7 个空格varchar(10)是变长存abc只占 3 个字节10 是上限。所以像账号、密码这种长度固定的用char姓名、地址这种长度不定的用varchar。金额字段用decimal(10,2)表示总共 10 位、其中小数 2 位能精确表示钱千万别用float浮点数算钱会出现0.1 0.2 ! 0.3的经典问题。3. 增删改查全流程一份可复制的 settings 配置与 CRUD 代码这一节是核心把建表、插入、查询、更新、删除五个动作全部走一遍。先给一份可以直接落地的配置片段我习惯把它单独放一个db_config.py其他脚本import进来用。# db_config.py import pymysql DB_CONFIG { host: 127.0.0.1, port: 3306, user: root, password: your_pwd, charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor, autocommit: False, } def get_conn(dbNone): cfg dict(DB_CONFIG) if db: cfg[database] db return pymysql.connect(**cfg)建库建表并插入初始数据这段可以当初始化脚本跑一次from db_config import get_conn CREATE_TABLE_SQL create table if not exists account( account_id varchar(20) not null, account_passwd char(6) not null, money decimal(10,2) default 0.00, primary key(account_id) ) default charsetutf8mb4; INIT_DATA_SQL insert into account(account_id, account_passwd, money) values(001,123456,1000.00),(002,456789,5000.00) on duplicate key update moneyvalues(money); def init(): conn get_conn(bank) cursor conn.cursor() try: cursor.execute(CREATE_TABLE_SQL) cursor.execute(INIT_DATA_SQL) conn.commit() print(初始化成功) except Exception as e: conn.rollback() print(初始化失败, e) finally: cursor.close() conn.close() if __name__ __main__: init()注意insert里用了on duplicate key update这样脚本重复执行不会因为主键冲突报错而是把金额重置回初始值方便反复测试。查询是最常用的操作重点讲fetchone和fetchall的指针行为。游标内部维护一个位置指针fetchone()取一条指针就往后移一格fetchall()取剩余全部。所以如果你先fetchone()再fetchall()第二条开始的数据才会被fetchall拿到第一条已经消费掉了。def query_all(): conn get_conn(bank) cursor conn.cursor() try: cursor.execute(select * from account;) rows cursor.fetchall() for row in rows: print(row[account_id], row[account_passwd], row[money]) return rows finally: cursor.close() conn.close()带参数的查询一定要用占位符%s不要用字符串拼接否则会有 SQL 注入风险。pymysql 的占位符统一是%s哪怕参数是数字也用%sdef query_by_id(account_id): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( select * from account where account_id%s;, (account_id,) ) return cursor.fetchone() finally: cursor.close() conn.close()更新和删除同样要记得commit并且用rowcount判断是否真的影响到了行def update_money(account_id, delta): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( update account set moneymoney%s where account_id%s;, (delta, account_id) ) if cursor.rowcount 1: conn.commit() return True conn.rollback() return False except Exception as e: conn.rollback() print(更新失败, e) return False finally: cursor.close() conn.close()rowcount返回的是本次执行影响的行数更新成功就是 1账号不存在就是 0。这个判断很重要能防止「以为更新了其实没匹配到」的假成功。删除同理把 SQL 换成delete from account where account_id%s即可。4. 验证连接与结果怎么确认数据真的写进去了代码跑完不报错不代表数据真的进了库。验证分两层连接层验证和结果层验证。连接层验证最简单执行一条select 1或者select version()能返回结果就说明通道没问题def check_conn(): try: conn get_conn() cursor conn.cursor() cursor.execute(select 1 as ok;) result cursor.fetchone() cursor.close() conn.close() return result[ok] 1 except Exception as e: print(连接失败, e) return False结果层验证最直接的办法是写完之后立刻查一次对比预期。比如插入一条记录后用主键查回来看字段值是否一致def verify_insert(account_id): row query_by_id(account_id) if row: print(查到记录, row) return True print(没查到插入可能没提交) return False还有一个容易被忽略的点decimal类型从数据库取出来是Decimal对象不是float。直接print没问题但如果你要参与计算或者格式化得注意类型转换。比如保留两位小数import decimal money row[money] # Decimal(1000.00) formatted money.quantize(decimal.Decimal(0.00)) print(str(formatted)) # 1000.00quantize做四舍五入并固定小数位比round()更适合金额场景因为round()对Decimal的行为和浮点数不一样容易出意外。再给一个完整的验证脚本把连接、写入、读回三步串起来跑通它你就有了一个可复用的自检模板from db_config import get_conn def full_check(): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( insert into account(account_id, account_passwd, money) values(%s,%s,%s) on duplicate key update moneyvalues(money);, (003, 111111, 888.88) ) conn.commit() cursor.execute(select * from account where account_id%s;, (003,)) row cursor.fetchone() assert row is not None, 写入后查不到记录 assert str(row[money]) 888.88, 金额不一致 print(全流程验证通过, row) except Exception as e: conn.rollback() print(验证失败, e) finally: cursor.close() conn.close() if __name__ __main__: full_check()assert在这里当断言用条件不满足直接抛异常适合做自检。生产代码里可以换成if not ...: raise效果一样。5. 常见报错逐个排查1045、2003、Incorrect string value 怎么解报错不可怕怕的是看不懂。这一节把 pymysql 最常见的几个错误对照着讲每个都给原因和改法。1045 Access denied用户名或密码不对或者该用户没有从当前主机登录的权限。先确认密码有没有写错再确认host参数——localhost和127.0.0.1在 MySQL 权限体系里可能是两个不同的来源。如果密码里有特殊字符注意别被配置文件转义了。2003 Cant connect to MySQL server连不上服务器通常是 MySQL 服务没启动、端口不对、或者防火墙拦了。先在命令行mysql -u root -p试试能不能登能登说明服务正常问题在 Python 侧的 host/port 配置。1049 Unknown database连接时指定的database不存在。要么先建库要么连接时不传database连上后再cursor.execute(use bank;)。Incorrect string value字符集不匹配多半是用了utf8存了四字节字符。把连接和建表的字符集都改成utf8mb4已有表可以用alter table account convert to charset utf8mb4;转换。pymysql.err.ProgrammingError: (1064, ...)SQL 语法错误常见于占位符写错。记住 pymysql 只认%s不认?也不认:name参数要用元组或列表传单个参数也要写成(value,)带逗号。数据写不进去但没报错九成是忘了commit()。autocommitFalse时所有写操作都在事务里不提交就回滚。要么每次写完手动commit要么连接时设autocommitTrue但后者不利于批量操作时统一回滚建议还是手动控制。Too many connections连接没关泄漏了。检查每个get_conn()是否都有对应的close()最好用try/finally包起来或者用上下文管理器。pymysql 的连接对象支持with语法with get_conn(bank) as conn: with conn.cursor() as cursor: cursor.execute(select * from account;) print(cursor.fetchall())用with能自动关闭省心不少。Lost connection to MySQL server during query查询太大或者超时。可以调大connect_timeout和read_timeout或者分页查询别一次性fetchall几十万行。把这些报错和对应改法整理成一张对照表出问题时先查表再动手报错关键字大概率原因处理方向1045 Access denied账号密码错 / host 不匹配核对 user、password、host2003 Cant connect服务未启动 / 端口错命令行先验证能登1049 Unknown database库不存在先建库或去掉 database 参数Incorrect string value字符集不是 utf8mb4连接和建表都改 utf8mb41064 语法错误占位符写错统一用 %s参数传元组写不进无报错忘了 commit手动提交或开 autocommitToo many connections连接泄漏try/finally 或 with 管理排查顺序建议从外到内先确认 MySQL 服务本身能连再确认 Python 连接参数最后才看 SQL 语句。这样能快速定位问题在哪一层不至于一上来就改代码。6. 把读写串成一个能跑的小项目银行账户管理实战前面都是零件这一节把它们组装成一个完整可运行的程序。功能就四个查询余额、取钱、存钱、退出。核心逻辑是登录校验加金额增减涉及update和事务提交正好把前面讲的知识点全用上。先看账户类把登录检查和金额操作封装进去import decimal from db_config import get_conn class Account: def __init__(self, account_id, account_passwd): self.account_id account_id self.account_passwd account_passwd def check_account(self): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( select * from account where account_id%s and account_passwd%s;, (self.account_id, self.account_passwd) ) return cursor.fetchone() is not None finally: cursor.close() conn.close() def query_money(self): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( select money from account where account_id%s and account_passwd%s;, (self.account_id, self.account_passwd) ) row cursor.fetchone() if row and row[money] is not None: return str(row[money].quantize(decimal.Decimal(0.00))) return 0.00 finally: cursor.close() conn.close() def reduce_money(self, amount): conn get_conn(bank) cursor conn.cursor() try: has_money decimal.Decimal(self.query_money()) amount decimal.Decimal(amount) if amount has_money: cursor.execute( update account set moneymoney-%s where account_id%s and account_passwd%s;, (amount, self.account_id, self.account_passwd) ) if cursor.rowcount 1: conn.commit() return True conn.rollback() return False print(余额不足) return False except Exception as e: conn.rollback() print(取款异常, e) return False finally: cursor.close() conn.close() def add_money(self, amount): conn get_conn(bank) cursor conn.cursor() try: cursor.execute( update account set moneymoney%s where account_id%s and account_passwd%s;, (decimal.Decimal(amount), self.account_id, self.account_passwd) ) if cursor.rowcount 1: conn.commit() return True conn.rollback() return False except Exception as e: conn.rollback() print(存款异常, e) return False finally: cursor.close() conn.close()主流程负责交互def main(): account_id input(请输入账号) account_passwd input(请输入密码) account Account(account_id, account_passwd) if not account.check_account(): print(账号或密码错误) return menu 请输入操作\n1、查询余额\n2、取钱\n3、存钱\n4、取卡\n choose input(menu) while choose ! 4: if choose 1: print(您的余额是 %s 元 % account.query_money()) elif choose 2: amount input(您的余额是 %s 元请输入取款金额 % account.query_money()) if account.reduce_money(amount): print(取款成功余额 %s 元 % account.query_money()) else: print(取款失败) elif choose 3: amount input(请输入存款金额) if account.add_money(amount): print(存款成功余额 %s 元 % account.query_money()) else: print(存款失败) else: print(无效操作) choose input(menu) print(谢谢使用) if __name__ __main__: main()跑起来的效果输入账号001、密码123456登录成功选 1 显示余额 1000.00选 2 输入 200余额变 800.00选 3 输入 500余额变 1300.00选 4 退出。整个过程数据实时落库重新登录余额还在。这里有几个设计上的取舍值得说。金额全程用Decimal处理避免浮点误差每次操作都独立开关连接虽然开销略大但避免了长连接超时和状态污染对初学者更友好update后用rowcount判断是否命中命中才commit否则rollback保证事务一致性。如果你想优化可以把连接复用起来或者引入连接池但那是下一步的事了。最后留个练习方向给账户表加一个交易流水表每次取钱存钱都往流水表插一条记录字段包括账号、操作类型、金额、时间。这样你就能把insert和update放在同一个事务里进一步体会commit和rollback的作用。写代码这件事看十遍不如自己敲一遍把上面的代码复制到本地跑通再按自己的需求改比什么都强。