Python+MySQL三层架构实战：从连接池到业务分层设计-北京尧图网络

简介一套基于Python与MySQL的三层架构源码包面向初学分层开发或希望规范数据库操作代码的读者解决业务逻辑与SQL语句混杂、复用性差的问题。包内共12个文件以6个Python源文件为主另有4个pyc中间文件及工程配置说明压缩包仅8KB体量轻巧。源码按经典三层分包运行数据库访问层负责连接MySQL并封装通用增删改查业务逻辑层围绕学生信息定义实体与操作规则表层作为测试入口调用具体功能启动后自动完成建库建表、插入两条示例数据并执行查询直观展示数据从表层到底层再返回的完整调用链。已有854人学习下载适合通过阅读与改造代码快速掌握PythonMySQL的分层开发思路也可作为课程设计或小型管理系统的底层骨架代码风格简洁适合作为教学演示。这篇博客我打算用三四年在Python后端项目里的亲身经历来写。先把丑话说在前面如果你现在还在一个文件里堆pymysql和业务逻辑看到“三层架构”这四个字觉得是Java面试题那么这篇内容大概率对你有用。我做的这个“python-MySql数据库三层架构源码”本质上不是什么黑科技而是把一套非常成熟的工程思路落到了Python项目里让代码从“能跑”变成“好改、好查、好交接”。今天我会把这套源码的骨架、分层边界、连接管理、以及我栽过的几个跟头完整摊开尽量让有Python基础但没怎么做过大一点项目的朋友也能照着把项目搭起来。1. 先说说我为什么决定改用三层架构1.1 一个 Controller 里塞了 200 行 SQL 的教训第一次用 Python 写带数据库的项目时我走的是最省事的路线路由函数直接pymysql.connect()查完数据塞进模板或返回 JSON。小项目头几周确实很爽代码量增长飞快。等到需求开始迭代问题就来了。一个用户列表页面先是加搜索再加分页然后加一个“根据VIP等级过滤”的需求。本来很简单的事结果因为 SQL 和业务逻辑全搅在一个函数里每次改需求我都得把那段几十行代码重新读一遍生怕动错一个括号。最离谱的一次我只是想加个排序字段结果把WHERE条件复制错了线上用户列表直接查出了全表数据幸好是内部系统不然后果不堪设想。那段崩溃的经历给了我一个重要教训**代码分层不是给老板看的是给一个月之后的自己看的。**当代码量超过 500 行如果还不分层每一次修改都在透支未来排查问题的耐心。1.2 三层架构到底在解决什么问题首先要明确我们在讨论的“三层架构”指的是**表现层Controller— 业务层Service— 数据访问层DAO**这一套逻辑分层它和“物理上拆几个服务、部署几台机器”没有关系。表现层负责接收请求、参数校验、返回响应。它不需要知道数据从哪来也不需要知道业务怎么算。业务层负责实现具体的业务规则比如“下单时扣库存、生成订单、记录流水”这些流程。它是整个架构的核心也是最需要被测试的部分。数据访问层DAO负责和 MySQL 打交道CRUD 操作的 SQL 全部集中在这里。业务层不关心你是用原生 SQL 还是 ORM只关心“你给了我数据”。这三层就像餐厅的流程前台点单Controller、后厨做菜Service、采购去仓库取货DAO。如果前台下厨房、采购自己开菜单餐厅规模一大肯定乱套。实际动手之前心里要清楚一个原则**上层依赖下层不能反过来。**Controller 不能直接拼 SQLService 不能把 SQL 细节带出来DAO 则不做业务判断。这个边界会在后续源码里反复体现。2. 源码骨架一个可以直接跑的 Python MySQL 三层示例2.1 项目目录结构与职责划分先看整体目录结构这是整套源码的骨架project_root/ ├── config/ │ ├── __init__.py │ └── database.py ├── dao/ │ ├── __init__.py │ ├── base_dao.py │ └── user_dao.py ├── service/ │ ├── __init__.py │ └── user_service.py ├── controller/ │ ├── __init__.py │ └── user_controller.py ├── utils/ │ ├── __init__.py │ └── db_connection.py ├── main.py └── requirements.txt有几点我在实际项目中反复体会过值得说明第一config 和 utils 不是一层。config 是静态配置数据库的连接地址、账号密码、连接池大小都在这里。utils 放工具函数比如获取数据库连接的工厂方法。它们不属于三层架构里的任何一层但是所有层都可能用到它们。很多初学者会把这两者混在一起为了省事在一个db.py里既放配置又放连接函数一旦配置变化就得改代码很麻烦。第二dao 目录的base_dao.py很关键。我习惯把所有 DAO 的公共操作抽到这里比如通用的查询转换、日志记录、异常包装。这样每个具体的 DAO 只需要写和业务有关的 SQL公共逻辑不会重复。第三controller 层不直接调用 DAO。这是检查分层是否正确的第一道开关如果发现 Controller 里 import 了dao.user_dao那大概率是分层没做对除非你在写异常简单的脚本。2.2 config 层的数据库连接管理先看配置文件内容# config/database.py import os DB_CONFIG { host: os.getenv(MYSQL_HOST, 127.0.0.1), port: int(os.getenv(MYSQL_PORT, 3306)), user: os.getenv(MYSQL_USER, root), password: os.getenv(MYSQL_PASSWORD, ), database: os.getenv(MYSQL_DATABASE, demo_db), charset: utf8mb4, autocommit: False, pool_size: 8, pool_recycle: 3600, }为什么要尽量从环境变量读配置很多人开发时习惯直接把密码写在代码里项目上线前再统一改。问题在于开发环境的密码和生产环境通常不一样而且代码仓库一旦被人看到密码就等于公开了。用环境变量以后同一个代码库在不同环境只需要改变量代码本身不用动。如果你项目还没有上容器或配置中心至少把配置单独放一个文件、不要放进私有 Git 仓库。这里特别注意autocommit: False。这是我踩了很多次坑之后才确定的取值——宁可手动 commit也不能让每条 SQL 自动提交。原因后面在事务边界那里细说。然后是连接池逻辑# utils/db_connection.py import time import pymysql from dbutils.pooled_db import PooledDB from config.database import DB_CONFIG _pool None def get_connection(): 获取连接池中的连接首次调用时初始化连接池。 global _pool if _pool is None: _pool PooledDB( creatorpymysql, mincached2, maxcachedDB_CONFIG[pool_size], maxconnectionsDB_CONFIG[pool_size] * 2, blockingTrue, ping1, **{key: value for key, value in DB_CONFIG.items() if key not in (pool_size, pool_recycle)} ) return _pool.connection()需要安装依赖pip install dbutils pymysql。为什么用连接池而不是每次pymysql.connect()我们看数据库连接的代价TCP 握手、权限校验、字符集协商这一套流程下来在本地可能只要几毫秒但在网络环境里可能 20-50 毫秒。如果每次请求都新建连接100 个并发请求就要新建 100 次连接MySQL 的max_connections很容易被打满。连接池相当于建了一个“连接蓄水池”用完之后把连接归还而不是真断开下一次请求直接复用。ping1这个参数也值得解释。MySQL 的wait_timeout默认 8 小时如果连接超过 8 小时没有干活MySQL 服务端会把连接干掉。此时客户端手里的连接已经“死”了直接执行 SQL 会报MySQL server has gone away。设了ping1之后连接池会在取连接时先探测连接是否还活着如果断了就重建避免拿到废连接。2.3 dao 层只做数据访问DAO 是三层里离数据库最近的地方设计上必须“听话”只做 CRUD不做业务判断。以用户表为例# dao/base_dao.py import logging import pymysql from utils.db_connection import get_connection logger logging.getLogger(__name__) class BaseDAO: DAO 基类提供通用的查询封装。 def query_all(self, sql, paramsNone): conn get_connection() try: with conn.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute(sql, params or ()) return cursor.fetchall() except pymysql.Error as e: logger.error(query_all failed, sql%s, params%s, error%s, sql, params, e) raise finally: conn.close() def query_one(self, sql, paramsNone): conn get_connection() try: with conn.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute(sql, params or ()) return cursor.fetchone() except pymysql.Error as e: logger.error(query_one failed, sql%s, params%s, error%s, sql, params, e) raise finally: conn.close() def execute(self, sql, paramsNone): 执行 INSERT / UPDATE / DELETE返回受影响行数。 conn get_connection() try: with conn.cursor() as cursor: cursor.execute(sql, params or ()) conn.commit() return cursor.rowcount except pymysql.Error as e: conn.rollback() logger.error(execute failed, sql%s, params%s, error%s, sql, params, e) raise finally: conn.close()conn.close()在连接池场景下并不意味着真的关闭连接而是把连接还给池子这也是很多人刚开始容易绕晕的点。注释里我特意写清楚方便后来维护的人不至于误解。具体业务 DAO 再继承基类# dao/user_dao.py from dao.base_dao import BaseDAO class UserDAO(BaseDAO): def get_by_id(self, user_id): sql SELECT id, username, email, created_at FROM users WHERE id %s AND deleted 0 return self.query_one(sql, (user_id,)) def list_by_page(self, offset, limit, keywordNone): sql SELECT id, username, email, created_at FROM users WHERE deleted 0 params [] if keyword: sql AND username LIKE %s params.append(f%{keyword}%) sql ORDER BY id DESC LIMIT %s OFFSET %s params.extend([limit, offset]) return self.query_all(sql, tuple(params)) def create(self, username, email, password_hash): sql INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s) return self.execute(sql, (username, email, password_hash)) def delete_by_id(self, user_id): # 逻辑删除避免真正物理删除导致关联数据挂掉 sql UPDATE users SET deleted 1 WHERE id %s return self.execute(sql, (user_id,))注意代码里所有 SQL 都用了%s占位符参数通过cursor.execute的第二个参数传入。这样做可以有效防止 SQL 注入——不是“基本能防”而是pymysql 底层会把参数转义成安全字符串再拼进 SQL比你自己手动拼接安全得多。list_by_page里的动态 SQL 看起来很简单其实踩坑点不少。用LIKE %s配合f%{keyword}%生成参数本质上还是参数化查询所以%只是传入的普通字符不是拼进 SQL 的语法。如果你以前习惯直接fSELECT ... WHERE username LIKE %{keyword}%那么既经不住一些特殊字符的考验也容易被注入趁早改掉。2.4 service 层承载业务逻辑Service 层是三层架构里“业务规则的地盘”。比如注册这个动作不是简单往表里插一条记录它至少包含# service/user_service.py import hashlib import secrets from dao.user_dao import UserDAO class UserService: def __init__(self): self.user_dao UserDAO() def get_user(self, user_id): user self.user_dao.get_by_id(user_id) if not user: raise ValueError(用户不存在) # 去掉敏感字段再返回 user.pop(password_hash, None) return user def register(self, username, email, password): # 1. 参数合法性校验 if not username or len(username) 2: raise ValueError(用户名至少 2 个字符) if not in email: raise ValueError(邮箱格式不正确) if len(password) 8: raise ValueError(密码至少 8 位) # 2. 注册前先检查是否已存在 existing self.user_dao.get_by_username(username) if existing: raise ValueError(用户名已被占用) # 3. 密码做加盐哈希 salt secrets.token_hex(8) password_hash hashlib.sha256((salt password).encode(utf-8)).hexdigest() # 4. 交给 DAO 落库 rows self.user_dao.create(username, email, f{salt}${password_hash}) if rows ! 1: raise RuntimeError(用户注册失败影响行数异常) return True这里体现的规则是Controller 只管“这个请求我要做什么”Service 管“这件事应该怎么一步步做成”。register方法里每一步都是一个业务规则即使将来换了一个框架、换了一种 HTTP 封装这些规则也不需要改。为什么密码要加盐哈希而不是直接存明文这是基本安全常识但还是强调一下。明文密码的数据库一旦泄露所有用户的账号相当于全部暴露。加盐之后即使两个用户密码相同哈希值也不同拖库的人没法用彩虹表快速反推。secrets.token_hex(8)生成的是密码学安全的随机盐比random模块靠谱得多。2.5 controller 层只负责调度以 Flask 为例Controller 层长这样# controller/user_controller.py from flask import Blueprint, request, jsonify from service.user_service import UserService user_bp Blueprint(user, __name__) user_service UserService() user_bp.route(/api/users/int:user_id, methods[GET]) def get_user(user_id): try: user user_service.get_user(user_id) return jsonify({code: 0, data: user}) except ValueError as e: return jsonify({code: 400, message: str(e)}), 400 except Exception as e: # 统一异常捕获避免堆栈直接抛给前端 return jsonify({code: 500, message: 服务器内部错误}), 500 user_bp.route(/api/users, methods[POST]) def create_user(): payload request.get_json(silentTrue) or {} try: user_service.register( usernamepayload.get(username, ).strip(), emailpayload.get(email, ).strip(), passwordpayload.get(password, ) ) return jsonify({code: 0, message: 注册成功}) except ValueError as e: return jsonify({code: 400, message: str(e)}), 400 except Exception: return jsonify({code: 500, message: 服务器内部错误}), 500Controller 里的异常处理我特意做了分类业务异常ValueError返回 400未知异常统一返回 500。这样做的好处是把错误信息转换成用户能看懂的提示而不是直接把 Python 堆栈发给前端。同时 Service 里的校验逻辑也可以被其他调用方复用不局限于 HTTP 接口。这样一个 demo 就通了main.py里注册蓝图启动 Flask用户注册、查询接口就能跑起来。3. 连接池与 SQL 安全问题配置里最容易被忽略的两个坑3.1 为什么不能每次都 open 一个新的 connection如果你只是写个脚本一天跑一次直接connect()也没多大问题。但 Web 服务是高并发场景每次请求新建连接的坏处不只是慢还会拖垮 MySQL。我搭建这套源码时做过一次对比实验模拟 500 个并发请求用无连接池版本时pymysql.connect()反复建立连接MySQL 端的Threads_connected一度飙到 300 多部分请求直接报Too many connections。换成连接池后连接数稳定在 10 个左右压测请求全部正常返回。这个数字对比非常直观。连接池参数也不是越大越好。maxcached8是我在 4C8G 的普通云服务器上的常用取值如果机器配置更高可以适当调大。连接池过大会占用 MySQL 的并发连接数反而增加数据库压力过小则在高并发时请求阻塞等待。3.2 参数校验和 SQL 注入三层架构不是安全挡箭牌把 SQL 集中在 DAO 层以后直观的好处是审计方便想看看项目里哪些地方有 SQL只需要在dao目录里搜。但如果不养成参数化查询的习惯DAO 集中了反而让注入风险更集中。例如下面这种写法乍一看是“无字符串拼接”但本质是拼接# 错误示范不要模仿 sql fSELECT * FROM users WHERE username {username} self.query_all(sql)只要username传入 OR 11查出来的就是全表。正确写法必须是sql SELECT * FROM users WHERE username %s self.query_all(sql, (username,))另外一个常被忽略的点是“批量操作”也要参数化。比如批量插入用户sql INSERT INTO users (username, email, password_hash) VALUES (%s, %s, %s) data [(u1, e1test.com, hash1), (u2, e2test.com, hash2)] conn.executemany(sql, data)executemany底层也是走参数化不是拼接所以是安全的。但是要注意executemany的性能并不总比循环执行高很多如果数据量很大要考虑用LOAD DATA LOCAL INFILE或分批提交这两者的原理不同不要混用。3.3 编码与时区问题MySQL 连接中容易踩的隐形坑这是我实际项目里出现过的事故用户上传的昵称带 emoji写入数据库后变成一堆??。原因就是数据库连接没指定charset默认latin1不支持四字节 UTF-8 字符。上面配置里charset: utf8mb4非常重要。注意utf8在 MySQL 里其实不是真正的全量 UTF-8它最多支持三字节像 emoji 这种四字节字符会出错。utf8mb4才是完整的 UTF-8 编码。所以结论是建数据库、建表、连接串三处统一用utf8mb4。时区问题也常被忽略。如果DB_CONFIG里没指定时区连接会使用 MySQL 服务器的默认system时区。当 Python 端和 MySQL 端的时区不一致日期时间字段就会出现偏差。我通常在连接参数里补齐init_command: SET NAMES utf8mb4, use_unicode: True, client_flag: pymysql.constants.CLIENT.MULTI_STATEMENTS,init_command在连接建立后立刻执行确保字符集生效。单个查询里不要开MULTI_STATEMENTS只在明确需要一次执行多条语句时才开因为它同时也会放大 SQL 注入的风险。4. 我踩过的分层边界模糊问题典型事故复盘4.1 事故一业务逻辑偷偷写进了 DAO那是在一个报表需求里查询用户时要按照年龄段把他们划分成“青年/中年/老年”。当时图省事直接在 DAO 的 SQL 里用CASE WHEN把年龄段算好返回。初看没什么问题后来运营说“年龄段的划分标准变了”我不得不去翻 DAO 里的 SQL改完还要重新测试相关查询。这个问题的本质是把业务规则耦合进了数据访问逻辑。年龄段划分是业务规则应该放在 Service 层DAO 只需要返回用户的出生日期即可。改动需求时就只需要改 Service 层方法DAO 和 SQL 完全不用动。判断标准很简单如果一段逻辑离开了业务需求就讲不通那它应该是 Service 层而不是 DAO 层。DAO 关心的是“怎么把数据取出来”Service 关心的是“取出来之后怎么用”。很多重构觉得难就是因为边界当初没划清楚等到数据流动起来才去拆成本翻好几倍。4.2 事故二Service 层变成了“透传层”反过来还有一种情况Service 层没有业务逻辑只是把 Controller 的参数原封不动传给 DAO再把 DAO 结果原样返回。这种“透传层”让三层架构变成了两层还白白多了一层代码。出现透传层的场景通常是功能太简单比如“根据 ID 查一条配置”。这时候如果硬要套三层架构代码反而难看。我有段时间在这个问题上走了极端所有方法都套 Service结果 Service 里一半代码是空壳。后来我的处理原则是不是所有功能都必须经过 Service。纯粹的、没有业务规则的查询可以在 Controller 里直接调 DAO但要克制。一旦这个查询后续加了缓存、加了权限判断再补 Service 也不迟。架构是服务开发的不是反过来绑架开发的。4.3 事故三事务边界放错了位置事务这个坑极其隐蔽。一开始我把commit()分散在各个 DAO 方法里带来的问题很直接一个注册逻辑要同时写users表和user_profiles表两行都执行成功才能算注册成功。如果 DAO 里各自commit()第一张表提交了第二张表插入失败就会留下“用户已存在但资料为空”的脏数据。这个数据极难排查因为用户已经能在登录页面里看到账号一旦后续流程读取资料就会报空指针或 404。正确的事务控制应该在 Service 层。但原生pymysql 连接池的方式需要在 Service 层拿连接、传连接这样 DAO 方法的签名就要改代码会比较啰嗦。我当时快速采用的方案是连接池的每一条连接在get_connection()时开启事务DAO 里不commit()而是在 Service 层通过一个上下文管理器统一提交或回滚。from contextlib import contextmanager from utils.db_connection import get_connection contextmanager def transaction(): conn get_connection() try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close()Service 里用它包住多个 DAO 操作def register_with_profile(self, user_data, profile_data): with transaction(): user_id self.user_dao.create(...) profile_id self.profile_dao.create(user_id, ...)这样一来两个操作要么一起成功要么一起回滚。事务边界必须放在“业务动作”这层而不是“单条数据操作”这层这是三层架构中非常重要的一条规则。4.4 分层失败的一些迹象与自查方法经历了几次混乱之后我总结出一些迹象一旦在代码审查里看到基本就能判断分层已经失败Controller 里出现import pymysql或直接写 SQL 字符串。DAO 的方法命名带业务语气比如get_active_vip_users_for_report。Service 层大量重复代码只是调用顺序不同。一个 DAO 方法被多个 Service 方法以“几乎一样”的方式调用。修改一个业务规则需要动到 SQL。自查时我会做一件很笨但很有效的事用 IDE 的“调用关系图”看每个 DAO 方法被谁调用了。如果发现 Controller 直接调用了许多 DAO 方法就说明有些规则没有收口到 Service。如果发现 Service 里对同一个 DAO 方法有多个调用点就说明公共逻辑没有抽出来。这个检查过程比拍脑袋判断“分层是否合理”要靠谱得多。5. 这套代码跑起来之后的下一步演进思路5.1 从三层到仓库模式与工作单元当项目里表越来越多几个 DAO 的关系越来越复杂单纯的三层架构在事务控制上会开始吃力。这时候我建议向“仓库模式Repository Pattern 工作单元Unit of Work”演进。简单说Repository 在 DAO 之上再做一次封装让 Service 面对的是一个“对象集合”而不是 SQL 方法。工作单元则统一管理所有 Repository 的事务边界一次业务动作对应一个连接、一个事务。这样做的好处是事务可以和业务动作一一对应不再依赖手工传入连接。比如我想做一个“创建订单并扣减库存”的操作在 Repository 模式下 Service 只关心order_repo.create(...)和inventory_repo.decrease(...)工作单元保证这两个操作在一个事务里提交或回滚。代码的可读性和健壮性都会提升一个量级。5.2 引入 ORM 与保留原生 SQL 的权衡搭建三层架构时数据库访问层可以直接用pymysql原生 SQL也可以用 ORM比如 SQLAlchemy。我的个人经验是项目刚开始时用原生 SQL 连接池能让你对 SQL 的表现有更清晰的感觉也更容易定位慢查询。很多从 ORM 入门的同学写不出高效 SQL就是因为 ORM 隐藏了太多底层细节。当项目复杂到需要处理大量动态查询、多表联查和复杂聚合时再考虑在 DAO 层引入 SQLAlchemy 的 Core 或写原生 SQL。如果确定要用 ORM需要注意它并不替代分层。SQLAlchemy 的Session本身就是工作单元的实现但依然需要 Service 来承载业务规则不能把业务逻辑全堆在 ORM 模型里。模型是表结构的映射不是业务规则的容器。最后分享一个从实践中来、被反复验证的感受分层架构的收益最早体现在需求频繁变更的时期而不是开发的第一周。第一周直接平铺直写确实快但到了第二三个月当每个新需求都变得轻松可控当你把项目交接给同事之后对方能快速定位到要改的代码你会意识到当初坚持这个骨架没有白费。这套代码虽然看起来比“一个.py全搞定”多绕了几层但它把变化和稳定分开了SQL 找 DAO规则找 Service入口找 Controller任何一头调整都不至于牵连全局。这就是我觉得它值得被参考和复现的原因。本文还有配套的精品资源点击获取

Python+MySQL三层架构实战：从连接池到业务分层设计

相关推荐

nanochat 核心结构——SFT 微调

视频生成模型本地部署实战：LoRA加速与低显存优化指南

告别sudo静默等待：用SUDO_ASKPASS实现密码提示音与弹窗

更多资讯

Prompt指令设计绿皮书：从公式到场景模板的实战拆解

知识图谱+图神经网络：电影推荐系统从原理到实践

表单引擎的留白：把规则、依赖和渲染分开

1024程序员节寄语：给代码留一份能复查的说明

工笔层染给生成式UI的启发：内容未齐，交互先别着急

text-to-cad 实战：从自然语言到 STEP/STL 模型全流程

今日更新

防幻觉智能体架构：电源硬件设计中的大模型编排与工具化实践

OpenGL面试核心考点：渲染管线、矩阵变换与性能排查全解析

全国计算机二级C语言真题PDF高效刷题指南

本周精选

防幻觉智能体架构：电源硬件设计中的大模型编排与工具化实践

OpenGL面试核心考点：渲染管线、矩阵变换与性能排查全解析

全国计算机二级C语言真题PDF高效刷题指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？