企业官网建设流程全解析

简介：面向Python开发者，这份PDF以Python 3.7与PyMySQL 0.9.3为基础，系统讲解如何连接MySQL数据库并完成增删改查操作。文中封装了一个可复用的数据库连接类，提供do_one与select两个核心方法，分别用于执行插入、删除、修改等变更操作和数据查询，二者均可接收包含主机、端口、用户名、密码、数据库名、字符集的参数字典，从而在不同数据库实例之间灵活切换。代码示例中还完整覆盖了游标创建、事务提交、异常捕获、连接关闭等环节，帮助理解实际项目中容易遗漏的细节，可用于快速搭建数据库操作模块或作为学习参考，尤其适合在需要管理多个数据库连接的项目中对照使用。包体为单个PDF文件，大小约50KB，内容紧凑、便于离线查阅。该资源已有2395人浏览学习，适合需要快速掌握Python操作MySQL的初中级开发者。

1. 为什么要在 PyMySQL 0.9.3 之上再包一层连接类

写 Python 连接 mysql 数据库，入门教程通常只教三行代码：import pymysql、pymysql.connect()、cursor.execute()，然后就没有然后了。可一旦进入真实后端项目，这种裸写方式会迅速暴露问题：每个模块都要重复做连接和关闭，连接参数散落在十几个文件里，遇到主库、从库、归档库多个数据源时，改一处漏一处。这里拆解的是一个不到百行的 mysql 数据库连接封装类，运行环境是 Python 3.7 和 PyMySQL 0.9.3。它把连接参数和 SQL 执行分开，对外只留两个方法：do_one() 做增删改，select() 做查询。正在做 python 入门练习的读者可以拿它当封装范例，写后端服务的同事也能把它作为数据库工具类的简化模板。

2. 连接参数 dict 与 pymysql.connect(**conn) 的完整语义

真正理解这个类，先从 do_one 和 select 共用的第一个参数 conn 开始。它看起来像一个普通字典，却在调用pymysql.connect时被展开成关键字参数，这一层设计直接影响后面多数据库切换的灵活性。

2.1 传字典而非常量参数的工程原因

代码里的调用方式是这样的：

config = {'host': '127.0.0.1', 'port': 3060, 'user': 'test', 'password': 'test', 'db': 'TEST', 'charset': 'utf8mb4'} connection = pymysql.connect(**config)

**config会把字典展开成关键字参数传入 pymysql.connect，等价于逐一写出 host、port、user 等参数。这段写法的价值不在节省几个字符，而在于把“数据库配置”从“执行逻辑”中剥离。dict 内容可以来自配置文件、环境变量或配置中心的 JSON，业务代码只负责传参；切库时不用改 SQL 执行方法，替换 dict 即可。

我一般会把配置读取单独拆一个函数，避免密码出现在代码仓库：

import os def load_db_config() -> dict: return { "host": os.getenv("MYSQL_HOST", "127.0.0.1"), "port": int(os.getenv("MYSQL_PORT", "3306")), "user": os.getenv("MYSQL_USER", "test"), "password": os.getenv("MYSQL_PASSWORD", "test"), "db": os.getenv("MYSQL_DB", "TEST"), "charset": "utf8mb4", }

这里有一点要注意：**conn展开时，字典里的键必须和 pymysql.connect 的形参严格对应。多写一个不认识的关键字会在运行时抛 TypeError，少写一个则可能因为缺参直接失败。这种错误只在真正执行连接时才暴露，所以把配置集中到一个读取函数里，比散落在各处手写字典更安全。

2.2 六个核心参数与选值

以这个封装类为例，最常用的连接参数可以整理成下表：

host 在本地调试时可以写 127.0.0.1，但如果 Python 服务跑在 Docker 容器里而 MySQL 跑在宿主机上，host 要换成宿主机的局域网 IP 或网关地址，不能照抄 127.0.0.1。port 默认是 3306，示例代码里写的 3060 并不是标准端口，我见过不少线上事故是开发环境下改了端口号、配置文件没同步导致的，排查时先用SELECT 1验证。

2.3 charset=utf8mb4 和 autocommit 的默认行为

charset 用 utf8mb4 而不是 utf8，是因为 MySQL 的 utf8 最多支持三个字节，存 emoji 或生僻字时会报 “Incorrect string value”。utf8mb4 是 UTF-8 的完整超集，对中文和表情符号都能正确处理。

这段封装里没有设置 autocommit，而 PyMySQL 的默认 autocommit 是 False。这意味着执行 insert、update、delete 后必须显式 commit，否则变更只在当前连接内可见，连接关闭后数据不会落库。这个默认行为恰恰是 do_one 方法里每次执行后都调用 commit 的原因，先理解这一点，第 3 章的事务处理才看得明白。

3. do_one 实现增删改：事务提交、异常返回与连接回收

3.1 空 SQL 校验的写法对比

原版 do_one 开头有一段防御性校验：

if not len(sql) > 0 and sql == "": res = {"status": False, "result": (), "msg": "传入执行sql为空，请确认！！！"} return res

这个判断本身没有错，但写法偏绕。not len(sql) > 0已经能表达“字符串为空”，后面又用and sql == ""重复约束。更简洁的等价写法是：

if not sql or not sql.strip(): return {"status": False, "result": (), "msg": "传入执行sql为空，请确认！！！"}

not sql能同时拦截 None 和空字符串，sql.strip()能拦住只包含空格的情况。对于 do_one 这种接收外部 SQL 的入口，多一道空白校验比到数据库层才报错要友好得多。

3.2 事务提交与游标生命周期

先看 do_one 的正常执行链路：

connection = pymysql.connect(**conn) cursor = connection.cursor() result = cursor.execute(sql) connection.commit() cursor.close() connection.close()

整个流程分四步：连接数据库、创建游标、执行 SQL、提交事务。cursor.execute() 返回的是受影响行数，比如 insert 成功通常是 1，update 和 delete 返回实际变化行数。这里关键动作是connection.commit()，因为 PyMySQL 默认不开启 autocommit，缺少这一步时，即使 execute 返回成功，数据也没有真正写入数据库。

关闭顺序也有讲究，应该先关游标再关连接。如果反过来先关 connection，游标可能还持有未释放的结果集，在部分版本的 PyMySQL 中会报错或导致连接无法正常回收。所以代码里 cursor.close() 和 connection.close() 的顺序不能调换。

3.3 异常分支与 finally 的隐藏问题

原版 do_one 的异常处理值得改进，它把 close 写在了 except 分支里：

except Exception as err: cursor.close() connection.close() res["status"] = False res["msg"] = [sql, err]

问题在于：如果 execute 或 commit 抛异常时，cursor 或 connection 尚未成功创建，这里的 close 会二次报错，把原始错误信息替换掉。另外，异常发生后没有调用 rollback，事务处于未完成状态；这种连接如果被放进连接池复用，未提交的事务会带到下一个请求里。

常见做法是用 finally 统一回收连接，并在异常分支里主动回滚：

def do_one(conn_cfg: dict, sql: str) -> dict: res = {"status": True, "result": (), "msg": ""} if not sql or not sql.strip(): res["status"] = False res["msg"] = "传入执行sql为空，请确认！！！" return res connection = None cursor = None try: connection = pymysql.connect(**conn_cfg) cursor = connection.cursor() affected = cursor.execute(sql) connection.commit() res["result"] = affected except Exception as err: if connection is not None: connection.rollback() res["status"] = False res["msg"] = [sql, str(err)] finally: if cursor is not None: cursor.close() if connection is not None: connection.close() return res

变量先初始化成 None，finally 里做过判空再 close，可以避免“异常中的异常”。except 里通过 rollback 结束未提交事务，保证连接以干净状态回收。这里还用str(err)替代直接塞异常对象，调用方打印日志时不用再转换一次。

注意：execute 成功但 commit 失败时，数据库侧可能已经执行部分操作，建议在业务上配合幂等键或唯一索引做兜底。

3.4 INSERT、UPDATE、DELETE 调用示例与受影响行数

封装类实例化后，增删改的调用方式完全一致：

db = conn() config = load_db_config() insert_sql = "INSERT INTO user (name, age) VALUES ('Kurt', 28)" r1 = db.do_one(config, insert_sql) print(r1["status"], r1["result"]) # True 1 update_sql = "UPDATE user SET age = 29 WHERE id = 1" r2 = db.do_one(config, update_sql) print(r2["result"]) # 0 或 1 delete_sql = "DELETE FROM user WHERE id = 100" r3 = db.do_one(config, delete_sql) print(r3["result"]) # 实际删除行数

这里 result 返回的是受影响行数，不同 SQL 的语义有差别：

我特别提醒一下 update 返回 0 的情况。很多同学把它当成“更新失败”，实际只是因为 WHERE 条件命中了行、但新值和旧值相同，MySQL 不认为发生了变更。判断是否存在记录，应该用SELECT EXISTS(...)，而不是看 update 的结果。

4. select 查询方法：游标默认类型、fetchall 结果与返回体设计

4.1 select 的执行路径与查询语义

select 方法的执行链路和 do_one 很像，但少了 commit：

connection = pymysql.connect(**conn_cfg) cursor = connection.cursor() cursor.execute(sql) result = cursor.fetchall() cursor.close() connection.close()

查询操作不需要 commit，因为 SELECT 没有未提交事务。fetchall() 一次性取出所有结果行，适合中小结果集；如果查询返回几十万行还直接用 fetchall，内存会明显吃紧。这个封装类定位是基础操作，不做流式查询，但在大表场景下要有这个意识。

调用方式也很直接：

res = db.select(config, "SELECT id, name, age FROM user ORDER BY age DESC") if res["status"]: rows = res["result"] for row in rows: print(row) else: print(res["msg"])

ORDER BY age DESC可以直接放在 SQL 层完成 mysql 排序，不必把数据拉到 Python 里再排序。这样可以利用 MySQL 索引和优化器，比 Python 侧排序高效得多。

4.2 默认游标返回嵌套元组

PyMySQL 默认游标返回的是嵌套元组：

((1, 'Kurt', 28), (2, 'Amy', 25))

对于简单打印或按固定列序号取值，这种结构够用。真正的问题是字段顺序和 SELECT 语句强绑定。今天写SELECT id, name, age，代码里也许已经用了 row[1] 当 name；明天有人改成SELECT name, id, age，name 和 id 的位置互换，整段解析逻辑全部错位。这个坑在联表查询里尤其容易踩，字段一多，按下标取数基本不可维护。

4.3 用 DictCursor 保留字段名

更稳妥的方式是连接时指定 DictCursor：

import pymysql.cursors def select_with_dict(conn_cfg: dict, sql: str) -> dict: connection = pymysql.connect(**conn_cfg, cursorclass=pymysql.cursors.DictCursor) try: with connection.cursor() as cursor: cursor.execute(sql) rows = cursor.fetchall() return {"status": True, "result": rows, "msg": ""} except Exception as err: return {"status": False, "result": (), "msg": [sql, str(err)]} finally: connection.close()

这时 fetchall() 返回的是列表套字典：

[{'id': 1, 'name': 'Kurt', 'age': 28}, {'id': 2, 'name': 'Amy', 'age': 25}]

直接按字段名取值，日志和调试输出可读性高很多，转 JSON 时也几乎不需要处理。代价是每条行记录多一层字典包装，数据量大时内存占用比嵌套元组高一些。对于这个封装类面向的中小型查询场景，DictCursor 是更合理的默认选型。

如果想要元组的高压缩比和字段名的可读性，可以考虑pymysql.cursors.NamedTupleCursor，返回的是带字段名的命名元组，但它对字段别名处理比较严格，用得没有 DictCursor 普遍。

4.4 status/result/msg 返回体的工程价值

select 返回的是 dict 而不是裸 rows，这是整个封装里比较有价值的一个约定。status 表示执行成功与否，result 存放查询结果，msg 携带错误信息。调用方只需要统一判断 res["status"]，错误的 SQL 和异常信息都能从 msg 里直接打印出来：

res = db.select(config, "SELECT * FROM user WHERE id = 999") if not res["status"]: sql_text, err_text = res["msg"] print(sql_text, err_text)

如果直接返回 fetchall() 的结果，异常时只能抛出去靠 try/except 接，正常返回和异常返回的类型不统一，调用方写起来会很不舒服。这个返回结构虽然是字典，但字段固定，后续改造成 dataclass 或 NamedTuple 也容易。

5. 多库切换、连接池边界与最小改造方向

5.1 多数据库注册表

既然所有方法都接收 dict 形式的连接参数，实际项目里可以维护一张配置注册表：

DATABASES = { "main": {"host": "127.0.0.1", "port": 3306, "user": "app", "password": "***", "db": "app_main", "charset": "utf8mb4"}, "report": {"host": "192.168.1.10", "port": 3306, "user": "report", "password": "***", "db": "app_report", "charset": "utf8mb4"}, } def get_db_config(name: str) -> dict: if name not in DATABASES: raise KeyError(f"unknown database config: {name}") return DATABASES[name]

调用时写db.do_one(get_db_config("main"), sql)或db.select(get_db_config("report"), sql)。新增数据库只需要在注册表里加一行，业务代码完全不用感知主机和密码的变化。这正是原项目里“通过传入链接参数指定调用对应数据库”的设计意图。

5.2 连接池与长事务边界

这个类每次执行都会新建连接、用完关闭，低频任务和脚本完全没有问题。请求量上来后，每次 execute 都伴随一次 TCP 握手，性能瓶颈会先出现在连接建立上。常见做法是用 DBUtils.PooledDB 或 SQLAlchemy 连接池替换最底层的pymysql.connect，连接复用后吞吐能提升一个量级。

还有一个需要提前知道的边界：由于连接在使用后立刻关闭，这个类无法保证多次 do_one 调用处于同一个事务。比如先从 A 表扣减余额，再从 B 表增加积分，两步都执行成功但第二步连接断开，第一步的 rollback 已经无法生效。要支持跨语句事务，需要把 connection 暴露出来，或者引入 session/ORM 做统一管理，这不是当前这个类该承担的责任。

5.3 快速验证与常见排错

拿到封装类后，先跑一条最简查询验证整个链路：

db = conn() res = db.select(get_db_config("main"), "SELECT 1") assert res["status"] is True, res["msg"]

如果 status 为 False，检查顺序可以按这个列表走：

1. host 和 port 是否能连通，用 mysql workbench 或命令行先测试同款配置
2. 确认端口确实是 MySQL 实例在监听，示例里的 3060 和常见默认端口不一致
3. 看报错信息里是否包含 caching_sha2_password 相关字样

提示：MySQL 8 默认认证插件是 caching_sha2_password，和 PyMySQL 0.9.3 组合时偶发连接失败。如果排错卡在这一步，优先确认 mysql 安装配置教程 里的认证插件设置，或者把连接用户的插件切换为 mysql_native_password 做对比测试。

另外，不建议从搜索引擎下载非官方 MySQL 安装包，版本不一致很容易踩插件兼容问题；优先按 mysql 下载官网的版本和系统类型选择。

5.4 值得落地的三个改造点

给 do_one 增加 SQL 参数化能力是最优先的改造：

def do_one(self, conn_cfg, sql, args=None): ... affected = cursor.execute(sql, args)

参数化之后，变量可以用%s占位，由 PyMySQL 负责转义，避免用户输入拼接进 SQL 造成注入问题。select 方法也可以同步加上 args 参数。

第二个改造是统一在 select 中使用 DictCursor，并把 cursorclass 作为连接参数的一部分传入，保留以后切换回元组游标的余地。第三个改造是给两个方法增加超时控制，可以用pymysql.connect(read_timeout=10, write_timeout=10)，避免慢 SQL 长时间占用线程。这三步都不改变现有 status/result/msg 的调用约定，但能明显提升代码在生产环境下的稳定性和可读性。

本文还有配套的精品资源，点击获取