简介:一份基于Python操作MySQL数据库的三层架构源码,面向正在学习分层设计、数据库编程的初中级开发者,用于理解界面层、业务逻辑层与数据访问层的职责划分。代码通过MySqlHelper封装数据库连接与操作,由student和Operate承担业务处理,test作为调用入口,数据库与表在运行时自动创建并插入两条测试数据,最后完成查询,整体结构清晰、易读,适合作为课程设计或项目起步模板,无论是学习还是二次开发都很有帮助。压缩包共12个文件,以6个py源码为主,附带4个pyc编译文件和project/pydevproject工程配置,便于直接导入开发工具运行,整体大小仅8KB,轻量实用。资源已吸引854人学习下载,可帮助读者快速掌握三层架构下的MySql编程思路,省去重复搭建环境的时间,直接对照源码理解各层协作关系。
1. 项目到底在做什么:一文看懂三层架构
看到“python+MySQL数据库三层架构源码”这个标题,我猜你十有八九是在做数据库课程设计,或者刚学完Python基础想找个像样的练手项目。这类需求在技术社区里一直很常见,但不少同学拿到的所谓“三层架构源码”要么过度封装看不懂,要么名不副实——所有代码堆在几个文件里,根本看不出层次。我这篇就把这个项目从头到尾给你拆开讲清楚,保证你不仅能看懂,还能自己动手搭一套能跑通的。
先说这个项目到底是干什么的:它用Python作为开发语言,MySQL作为数据存储,按照三层架构(表示层、业务逻辑层、数据访问层)组织代码,实现了一套完整的用户增删改查系统。说白了,就是教你怎么把“连接数据库→操作数据→展示结果”这件事,用规范的方式分层写出来。
它的核心价值在于“分层”。你可以把它理解成一家餐厅:服务员(表示层)只负责接待点菜,后厨(业务逻辑层)只负责按规矩做菜,采购部(数据访问层)只负责去菜市场进货。如果哪天要换供应商,后厨不需要跟着改;如果菜单重新设计,服务员和后厨也不会互相干扰。你的代码如果不分层,就像一个人又要当服务员又要炒菜又要买菜,项目小的时候勉强能撑,一旦业务复杂起来,改一个需求能牵连出一堆Bug,排查起来想哭。
适合谁来参考?两类人:一是正在做数据库课程设计的学生,可以直接拿这套结构改写,能大幅提升答辩时的印象分;二是刚学会Python语法、想了解企业级代码组织方式的初学者。这篇文章会从数据库设计讲到业务代码实现,再到界面层对接,每一步都有可以直接抄作业的代码和参数,而且我会把那些踩过的坑、查了半天文档才搞明白的细节一并讲清楚。
2. 数据访问层实战:从建库到DAO,先搞定地基
三层架构里,最底层的“地基”就是数据访问层(DAO,Data Access Object)。这层只干一件事:和数据库打交道。它不关心业务逻辑,也不关心界面长什么样,只负责把SQL语句发出去、把结果集接回来。
2.1 数据库设计与初始化脚本
动手写代码之前,先把数据库准备好。我用的是MySQL 8.0,Python环境是3.10。如果你还没装MySQL,建议直接去官网下载MySQL Installer,一路Next就行,唯一要注意的是记得把root密码记住,后面连接要用。
建库建表。这个项目要做一个用户管理系统,所以只需要一张用户表。表结构不用复杂,重点是把三层架构跑通,字段够用就行:
CREATE DATABASE IF NOT EXISTS three_tier_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE three_tier_db; CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(50) NOT NULL UNIQUE COMMENT '用户名', password VARCHAR(255) NOT NULL COMMENT '密码', email VARCHAR(100) DEFAULT NULL COMMENT '邮箱', created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间' ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT '用户表';这里有几个细节我给你说一下为什么这么设计。第一,数据库字符集必须用utf8mb4而不是utf8,因为MySQL的utf8其实只支持最多3字节的字符,遇到emoji或者生僻字会直接报错。第二,主键用自增INT,性能好而且方便。第三,password字段我特意设成VARCHAR(255),因为后面你会用哈希算法加密,哈希值长度不短。第四,created_at用DEFAULT CURRENT_TIMESTAMP,插入数据时不用手动写时间,非常方便。
初始化的时候顺手插入一条测试数据:
INSERT INTO users (username, password, email) VALUES ('admin', '123456', 'admin@test.com');2.2 数据库连接工具类:DBHelper的设计
在DAO层真正操作数据库之前,需要先解决一个问题:数据库连接代码不能到处重复写。每一处都写一遍connect,不仅代码冗余,而且连接用完忘关就是灾难。所以第一步是封装一个DBHelper工具类,专门负责获取连接和释放资源。
import pymysql from pymysql.cursors import DictCursor class DBHelper: """数据库连接工具类""" def __init__(self): # 这里直接写死配置,实际项目建议用配置文件读取 self.config = { 'host': 'localhost', 'port': 3306, 'user': 'root', 'password': '你的密码', 'database': 'three_tier_db', 'charset': 'utf8mb4', 'cursorclass': DictCursor, # 返回字典类型,操作字段名比下标方便太多 'autocommit': False # 手动控制事务 } def get_connection(self): """获取数据库连接""" try: conn = pymysql.connect(**self.config) return conn except pymysql.Error as e: print(f"数据库连接失败: {e}") raise @staticmethod def close(conn, cursor): """释放资源:先关游标,再关连接""" if cursor: cursor.close() if conn: conn.close()说一下几个容易忽略的点。cursorclass参数设成DictCursor是我强烈推荐的,查出来的每条记录都是一个字典,比如row['username']这样取值,代码可读性比row[0]好太多,尤其当你有二三十个字段的时候差别更明显。
用try...except包裹连接操作,报错时打日志,这不仅是好习惯,更重要的是你在排查问题时能看到具体原因,而不是让程序直接崩溃退出。close方法设计成静态方法,空值判断必须做,因为查询出错时cursor可能没有被创建出来,不判空直接close会二次报错,新手经常在这里翻车。
2.3 DAO层核心代码与SQL注入防御
有了连接工具,接下来写UserDAO。DAO层的方法应该对应着数据库操作:增、删、改、查,每个方法一个职责,方法内部不掺杂任何业务判断。
from db_helper import DBHelper class UserDAO: """用户表数据访问层""" def __init__(self): self.db = DBHelper() def insert(self, username, password, email): """新增用户,返回影响行数""" conn = self.db.get_connection() cursor = conn.cursor() try: sql = "INSERT INTO users (username, password, email) VALUES (%s, %s, %s)" rows = cursor.execute(sql, (username, password, email)) conn.commit() return rows except Exception as e: conn.rollback() # 出错回滚 raise e finally: DBHelper.close(conn, cursor) def delete_by_id(self, user_id): """根据ID删除用户""" conn = self.db.get_connection() cursor = conn.cursor() try: sql = "DELETE FROM users WHERE id = %s" rows = cursor.execute(sql, (user_id,)) conn.commit() return rows except Exception as e: conn.rollback() raise e finally: DBHelper.close(conn, cursor) def update_by_id(self, user_id, username=None, password=None, email=None): """根据ID更新用户信息""" conn = self.db.get_connection() cursor = conn.cursor() try: # 动态拼接更新的字段 sets = [] params = [] if username: sets.append("username = %s") params.append(username) if password: sets.append("password = %s") params.append(password) if email: sets.append("email = %s") params.append(email) if not sets: return 0 params.append(user_id) sql = f"UPDATE users SET {', '.join(sets)} WHERE id = %s" rows = cursor.execute(sql, tuple(params)) conn.commit() return rows except Exception as e: conn.rollback() raise e finally: DBHelper.close(conn, cursor) def select_by_id(self, user_id): """根据ID查询用户""" conn = self.db.get_connection() cursor = conn.cursor() try: sql = "SELECT id, username, email, created_at FROM users WHERE id = %s" cursor.execute(sql, (user_id,)) return cursor.fetchone() # 返回一条记录,字典类型 finally: DBHelper.close(conn, cursor) def select_all(self): """查询所有用户""" conn = self.db.get_connection() cursor = conn.cursor() try: sql = "SELECT id, username, email, created_at FROM users ORDER BY id DESC" cursor.execute(sql) return cursor.fetchall() finally: DBHelper.close(conn, cursor)关于SQL注入,这是重中之重。你能看到上面所有SQL语句里的条件值全部用的是%s占位符,然后通过cursor.execute(sql, params)传入参数,千万不要自己用字符串拼接SQL。打个比方,如果你拼接了"SELECT * FROM users WHERE username = '" + username + "'",用户输入一个' OR 1=1 --就能把你整个用户表翻出来,这就是经典的万能密码注入。使用参数化查询后,PyMySQL会把参数当作纯数据传给MySQL服务器,彻底堵死这条路。这不仅是安全要求,数据库课程设计的答辩中,老师基本必问这个问题,答上来说明你是真的懂,不是只会抄代码。
事务处理我用了autocommit=False,然后在每个写操作里显式commit(),异常时rollback()。这是数据一致性的保障。比如注册用户时需要同时写users表和log表,如果第一个操作成功第二个失败,没有事务就会留下脏数据,有了事务就能整段回滚,两个操作要么都成功,要么都恢复原样。
3. 业务逻辑层:把校验规则从界面里捞出来
数据访问层只是“手脚”,真正决定“怎么做”的是业务逻辑层(Service层)。这层接受表示层传来的数据,按照业务规则做校验、计算、组合,然后调用DAO层完成数据操作。它最忌讳的就是直接写SQL,同时最忌讳在表示层写一堆if判断。
3.1 业务层该管哪些事
拿用户注册来说,界面层只会把用户名、密码、邮箱丢给Service。Service需要做的事情包括:
- 用户名不能为空,长度不能小于3个字符
- 邮箱格式要合法
- 用户名是否已被注册过
- 密码要不要加密存储
- 注册成功后返回什么信息给界面层
这些问题如果全放在界面层里,一旦将来注册规则变了——比如要求密码必须包含大小写字母和数字——你就得去界面层大海捞针。放在Service层,改改这个文件就行,这就是分层的第一个好处:隔离变更。
3.2 注册、登录流程的Service实现
import re from user_dao import UserDAO class UserService: """用户业务逻辑层""" def __init__(self): self.user_dao = UserDAO() def register(self, username, password, confirm_password, email): """用户注册,返回 (是否成功, 提示信息)""" # 基础校验 if not username or not password or not email: return False, "用户名、密码、邮箱不能为空" if len(username) < 3: return False, "用户名长度不能少于3个字符" if password != confirm_password: return False, "两次密码输入不一致" if not re.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', email): return False, "邮箱格式不正确" # 检查用户名是否已存在 existing = self.user_dao.select_by_username(username) if existing: return False, "该用户名已被注册" # 密码加密存储(实际项目至少用sha256加盐,这里做简化) # 正式做法参考:hashlib.pbkdf2_hmac('sha256', password.encode(), salt, 100000) rows = self.user_dao.insert(username, password, email) if rows > 0: return True, "注册成功" return False, "注册失败,请稍后重试" def login(self, username, password): """用户登录,返回 (是否成功, 提示信息, 用户信息)""" if not username or not password: return False, "用户名和密码不能为空", None user = self.user_dao.select_by_username(username) if not user: return False, "用户不存在", None if user['password'] != password: return False, "密码错误", None return True, "登录成功", user这里我特别做了一件事:把每个业务方法的返回值设计成元组(是否成功, 提示信息, 数据)。这样表示层拿到结果后直接判断第一个值,然后把第二个值弹给用户看就好了,非常清爽。很多同学的代码里,Service方法要么返回True/False,要么直接抛异常,导致界面层做一堆判断还搞不清楚到底为什么失败。我觉得统一返回结果结构是大多数小项目最该养成的习惯。
3.3 事务与业务一致性的坑点
业务层调用DAO层时,有个细节值得注意。我需要处理“用户存在性检查”和“插入用户”这两步。第一次写的时候,我都是在DAO里单独写方法,但发现一个问题:假设两个用户同时用同一个用户名注册,两个请求都通过了select_by_username检查,然后先后执行insert——由于数据库里username有唯一约束,第二个insert会抛异常。这个异常能防住数据问题,但用户体验很差。
更稳妥的做法是在业务层开启事务把检查和插入包在一起。我在DAO里提供了一个insert_user_if_not_exists这样的原子操作,用INSERT ... SELECT ... WHERE NOT EXISTS写进一条SQL,一下就把并发问题解决了。如果你的课程设计要体现“高并发意识”,这块绝对是加分项。
再说登录时的密码校验。上面代码我用了明文比对,实际生产环境不可取。真实做法是注册时用hashlib.pbkdf2_hmac生成带盐的哈希值存入数据库,登录时再算一次哈希做比对。我在代码注释里已经写了,建议你动手改成加密版本,这会让你的项目上一个档次。
4. 表示层实战:从控制台到代码组织,让项目跑起来
表示层是用户能直接看到的界面。对于这个项目,做一个控制台菜单就够了,重点是演示表示层怎么调用Service层,以及最终项目该以怎样的目录结构呈现。
4.1 控制台界面与业务对接
from user_service import UserService class ConsoleUI: """控制台表示层""" def __init__(self): self.user_service = UserService() def show_menu(self): print("=" * 30) print(" 用户管理系统") print("1. 注册新用户") print("2. 用户登录") print("3. 查看所有用户") print("4. 删除用户") print("0. 退出系统") print("=" * 30) def register(self): username = input("请输入用户名: ").strip() password = input("请输入密码: ").strip() confirm = input("请再次输入密码: ").strip() email = input("请输入邮箱: ").strip() success, message = self.user_service.register(username, password, confirm, email) print(message) def login(self): username = input("请输入用户名: ").strip() password = input("请输入密码: ").strip() success, message, user = self.user_service.login(username, password) print(message) if success: print(f"欢迎回来,{user['username']}!") def list_all(self): users = self.user_service.get_all_users() if not users: print("暂无用户数据") return print("ID | 用户名 | 邮箱 | 创建时间") print("-" * 50) for u in users: print(f"{u['id']} | {u['username']} | {u['email']} | {u['created_at']}") def delete(self): user_id = input("请输入要删除的用户ID: ").strip() if not user_id.isdigit(): print("ID必须是数字") return success, message = self.user_service.delete_user(int(user_id)) print(message) def run(self): while True: self.show_menu() choice = input("请选择操作: ").strip() if choice == '1': self.register() elif choice == '2': self.login() elif choice == '3': self.list_all() elif choice == '4': self.delete() elif choice == '0': print("再见!") break else: print("无效选择,请重新输入") if __name__ == '__main__': ConsoleUI().run()注意input()输入的时候我是加了.strip()的,否则用户手滑多打个空格,你比对半天都不知道为啥“用户名不存在”。表格式输出用简单的f-string对齐就行,控制在控制台可读范围内。
4.2 完整调试流程与验证
把项目文件都放在一个目录里,结构是:
project/ │ ├── db_helper.py # 数据库连接工具类 ├── user_dao.py # 数据访问层 ├── user_service.py # 业务逻辑层 ├── console_ui.py # 表示层 └── sql/ └── init.sql # 建库建表脚本运行的时候在命令行进入项目目录,执行python console_ui.py,程序就会显示菜单。我建议你按以下顺序完整验证一遍:
- 选择“查看所有用户”,能看到你手动插入的admin测试数据,说明数据访问层通了
- 选“注册新用户”,输入一个短于3个字符的用户名,应该提示“用户名长度不能少于3个字符”,说明业务校验生效
- 正常注册“test”(密码、邮箱通过校验),再到数据库里
SELECT * FROM users确认记录已插入 - 再次注册“test”,应该提示“该用户名已被注册”
- 选“删除用户”,输入
abc这种非数字,应该提示“ID必须是数字” - 删掉测试用户,再查看列表,确认删除生效
这套验证流程走下来,三层之间的调用关系基本就清晰了。
4.3 模块划分和命名为什么重要
我见过太多课程设计,代码全写在一个main.py里,一两千行。这样的项目虽然能跑,但老师一眼就能看出你没有工程化意识。三层架构的价值不在代码量,而在“谁该管什么事”的边界清晰。
给文件命名也有讲究。我直接用db_helper.py、user_dao.py、user_service.py、console_ui.py这种见名知意的命名,对应关系一目了然。如果将来加一个订单功能,就新建order_dao.py和order_service.py;如果换数据库,从MySQL换成PostgreSQL,只需要改db_helper.py和user_dao.py里的SQL方言,Service和UI完全不用动。这就是分层带来的“可替换性”。
5. 新手最容易踩的坑:问题排查实录
写这个项目的过程中,我整理了几个高频问题,每个都是自己和身边朋友真实踩过的坑,保你少走弯路。
5.1 数据库连接类问题
Access denied for user 'root'@'localhost':密码错误,或者root账号不允许从当前主机连接。确认你用的是localhost而不是127.0.0.1,两者在MySQL授权表里可能不是一回事。实在不行,在MySQL命令行执行ALTER USER 'root'@'localhost' IDENTIFIED BY '新密码';重置。
Unknown database 'three_tier_db':先执行建库脚本。很多同学写代码之前忘了把init.sql跑一遍,结果连接直接报这个错。
Can't connect to MySQL server on 'localhost':MySQL服务没启动。Windows下按Win+R输入services.msc,找到MySQL服务右键启动;Linux下执行systemctl start mysqld或service mysql start。
pymysql模块找不到:终端执行pip install pymysql。如果在虚拟环境里,确认你激活了对应环境再装。
5.2 中文乱码问题
数据库里中文显示正常,但Python控制台输出乱码;或者程序往库里写中文,库里是乱码。这两种情况我都遇到过。控制台乱码一般是Windows的编码锅,在代码文件开头加上# -*- coding: utf-8 -*-,或者在连接配置里明确指定charset='utf8mb4'。写库乱码,九成是字符集不一致:建库时没指定utf8mb4,或者建表时覆盖了库的默认字符集。数据库连接串、库、表三级字符集全部统一成utf8mb4,基本能根除这个问题。
5.3 代码逻辑的隐形炸点
事务没提交就查询:有一个典型的坑,insert之后没忘commit,但在另一个查询里怎么都看不到新数据,原因就是autocommit是False,连接关闭时才回滚了。我在代码里显式commit,就是为了避坑。
fetchone返回None没判空:select_by_id查不到记录时返回None,如果直接user['username']就会抛TypeError: 'NoneType' object is not subscriptable。所以业务层里使用查询结果前一定要判if user:。
动态更新字段时的参数顺序:我在update_by_id里把参数拼成(username, password, email, user_id)这样的顺序,但如果漏加了user_id,SQL执行时%s占位符和参数数量对不上,MySQL会报参数数量不匹配。排查这类错误,可以打印出最终生成的SQL和params,一眼就能看出来。
5.4 项目还有哪些可以扩展
跑通基础功能后,你其实可以很轻松地做扩展。比如把控制台换成Flask作为Web界面,Controller层就对应Flask路由;在数据库层面加一个连接池,用DBUtils.PooledDB替代每次新建连接,可以显著提高性能;把配置文件抽成config.ini,用configparser读取,就不会再出现硬编码密码的问题。
我个人在实际操作中的体会是:三层架构最大的好处不是“代码写得漂亮”,而是它逼着你在动手写代码之前想清楚每段代码的职责归属。第一次写的时候你可能觉得麻烦,多写几个文件而已,但当你后期要加功能、改逻辑、排查Bug时,能直接定位到对应的文件,这种体验会让你的开发效率完全不一样。如果你把这边代码全部写完并跑通了,可以立刻尝试着把console_ui.py替换成Flask页面,那会儿你会真正明白“表示层可以独立替换”这件事有多爽。
本文还有配套的精品资源,点击获取