简介一份基于Python操作MySQL数据库的三层架构源码面向正在学习分层设计、数据库编程的初中级开发者用于理解界面层、业务逻辑层与数据访问层的职责划分。代码通过MySqlHelper封装数据库连接与操作由student和Operate承担业务处理test作为调用入口数据库与表在运行时自动创建并插入两条测试数据最后完成查询整体结构清晰、易读适合作为课程设计或项目起步模板无论是学习还是二次开发都很有帮助。压缩包共12个文件以6个py源码为主附带4个pyc编译文件和project/pydevproject工程配置便于直接导入开发工具运行整体大小仅8KB轻量实用。资源已吸引854人学习下载可帮助读者快速掌握三层架构下的MySql编程思路省去重复搭建环境的时间直接对照源码理解各层协作关系。1. 项目到底在做什么一文看懂三层架构看到“pythonMySQL数据库三层架构源码”这个标题我猜你十有八九是在做数据库课程设计或者刚学完Python基础想找个像样的练手项目。这类需求在技术社区里一直很常见但不少同学拿到的所谓“三层架构源码”要么过度封装看不懂要么名不副实——所有代码堆在几个文件里根本看不出层次。我这篇就把这个项目从头到尾给你拆开讲清楚保证你不仅能看懂还能自己动手搭一套能跑通的。先说这个项目到底是干什么的它用Python作为开发语言MySQL作为数据存储按照三层架构表示层、业务逻辑层、数据访问层组织代码实现了一套完整的用户增删改查系统。说白了就是教你怎么把“连接数据库→操作数据→展示结果”这件事用规范的方式分层写出来。它的核心价值在于“分层”。你可以把它理解成一家餐厅服务员表示层只负责接待点菜后厨业务逻辑层只负责按规矩做菜采购部数据访问层只负责去菜市场进货。如果哪天要换供应商后厨不需要跟着改如果菜单重新设计服务员和后厨也不会互相干扰。你的代码如果不分层就像一个人又要当服务员又要炒菜又要买菜项目小的时候勉强能撑一旦业务复杂起来改一个需求能牵连出一堆Bug排查起来想哭。适合谁来参考两类人一是正在做数据库课程设计的学生可以直接拿这套结构改写能大幅提升答辩时的印象分二是刚学会Python语法、想了解企业级代码组织方式的初学者。这篇文章会从数据库设计讲到业务代码实现再到界面层对接每一步都有可以直接抄作业的代码和参数而且我会把那些踩过的坑、查了半天文档才搞明白的细节一并讲清楚。2. 数据访问层实战从建库到DAO先搞定地基三层架构里最底层的“地基”就是数据访问层DAOData Access Object。这层只干一件事和数据库打交道。它不关心业务逻辑也不关心界面长什么样只负责把SQL语句发出去、把结果集接回来。2.1 数据库设计与初始化脚本动手写代码之前先把数据库准备好。我用的是MySQL 8.0Python环境是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 创建时间 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 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, admintest.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, usernameNone, passwordNone, emailNone): 根据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 fUPDATE 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 11 --就能把你整个用户表翻出来这就是经典的万能密码注入。使用参数化查询后PyMySQL会把参数当作纯数据传给MySQL服务器彻底堵死这条路。这不仅是安全要求数据库课程设计的答辩中老师基本必问这个问题答上来说明你是真的懂不是只会抄代码。事务处理我用了autocommitFalse然后在每个写操作里显式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 rootlocalhost密码错误或者root账号不允许从当前主机连接。确认你用的是localhost而不是127.0.0.1两者在MySQL授权表里可能不是一回事。实在不行在MySQL命令行执行ALTER USER rootlocalhost IDENTIFIED BY 新密码;重置。Unknown database three_tier_db先执行建库脚本。很多同学写代码之前忘了把init.sql跑一遍结果连接直接报这个错。Cant connect to MySQL server on localhostMySQL服务没启动。Windows下按WinR输入services.msc找到MySQL服务右键启动Linux下执行systemctl start mysqld或service mysql start。pymysql模块找不到终端执行pip install pymysql。如果在虚拟环境里确认你激活了对应环境再装。5.2 中文乱码问题数据库里中文显示正常但Python控制台输出乱码或者程序往库里写中文库里是乱码。这两种情况我都遇到过。控制台乱码一般是Windows的编码锅在代码文件开头加上# -*- coding: utf-8 -*-或者在连接配置里明确指定charsetutf8mb4。写库乱码九成是字符集不一致建库时没指定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_idSQL执行时%s占位符和参数数量对不上MySQL会报参数数量不匹配。排查这类错误可以打印出最终生成的SQL和params一眼就能看出来。5.4 项目还有哪些可以扩展跑通基础功能后你其实可以很轻松地做扩展。比如把控制台换成Flask作为Web界面Controller层就对应Flask路由在数据库层面加一个连接池用DBUtils.PooledDB替代每次新建连接可以显著提高性能把配置文件抽成config.ini用configparser读取就不会再出现硬编码密码的问题。我个人在实际操作中的体会是三层架构最大的好处不是“代码写得漂亮”而是它逼着你在动手写代码之前想清楚每段代码的职责归属。第一次写的时候你可能觉得麻烦多写几个文件而已但当你后期要加功能、改逻辑、排查Bug时能直接定位到对应的文件这种体验会让你的开发效率完全不一样。如果你把这边代码全部写完并跑通了可以立刻尝试着把console_ui.py替换成Flask页面那会儿你会真正明白“表示层可以独立替换”这件事有多爽。本文还有配套的精品资源点击获取