Python+MySQL员工信息管理系统：连接池、事务与分页CRUD实战

简介：面向数据库课程设计学习者的员工信息管理系统源码包，围绕Python与MySQL的交互实现员工信息的增删改查，适合需要完成课程设计、理解数据库建模与后端编程衔接的学生参考。压缩包共9个文件，以4张PNG图示、2个Python脚本、2份Word文档和1份Markdown说明为主，整体约6.71MB，包含E-R图、数据流图、系统界面截图、课程设计报告与代码说明等材料，便于快速了解项目结构。已有4344人学习，说明该案例在数据库课程实践中具有一定参考价值。读者可从中获得基于pymysql连接MySQL、建表与SQL查询更新、事务处理及异常处理的实现思路，也能借助报告和图示梳理需求分析、表结构设计与系统功能模块，用于对照完善自己的课程设计，或作为小型管理系统的练习蓝本。

1. 员工信息管理系统长什么样：一份能跑通的数据库课程设计该有的样子

一份数据库课程设计的源码，最怕的不是功能少，而是跑不起来。这个员工信息管理系统用 Python 写业务逻辑、MySQL 存数据，管的是部门、岗位、员工档案这几张核心表，能加人、改人、查人、离职、调岗，还带一个登录和操作留痕。它适合两类人：赶课程设计需要一份完整源码照着改的，以及刚学完 SQL 想知道 Python 到底怎么把数据库真正用起来的新手。

整套代码没有花哨框架，控制台菜单加分层的结构，代码量不大但五脏俱全。

别小看它。连接池、事务、分页、字符集这几件事有没有理顺，恰恰是「能交作业」和「能写进简历」的分水岭。

2. 环境搭建：Python 3 + MySQL 8 + PyMySQL 这套组合怎么一次装对

课程设计翻车，十次里有六次不是业务写错，而是环境没配对。我见过最常见的场面：MySQL 8 装好了，pip install pymysql也成功了，代码一跑就抛RuntimeError: cryptography is required for sha256_password or caching_sha2_password，然后人就开始怀疑是不是 Python 版本不对。其实原因是 MySQL 8 默认的身份认证插件换成了caching_sha2_password，PyMySQL 要处理这种认证就得依赖cryptography这个包。这种坑属于「不知道就卡一天，知道了三十秒解决」的典型，先把它写进依赖清单里最省事。

2.1 依赖清单与虚拟环境

依赖我一般固定成四五个，够用就好。PyMySQL是纯 Python 实现的驱动，不需要编译，装起来比mysqlclient省心得多，课程设计这种场景完全够；DBUtils负责连接池，后面db.py会用到；cryptography就是上面说的那个认证依赖；prettytable纯粹是为了控制台输出好看一点，打印员工列表时能对齐成表格，不加也不影响功能。

# requirements.txt PyMySQL==1.1.0 DBUtils==3.0.3 cryptography==42.0.5 prettytable==3.10.0

装的时候一定走虚拟环境，别往全局 Python 里塞。Windows 和 macOS/Linux 的激活命令不一样，搞混了会出现「明明装了却 import 不到」的玄学问题，本质是装到了另一个解释器里。

# 创建并激活虚拟环境 python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate # 安装依赖 pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

激活成功的标志是命令行前面出现(venv)。后面所有python main.py的操作都要在这个环境里执行，否则连接池和驱动都会去找全局环境，报错信息还特别有迷惑性。

2.2 建库建表：把约束写进 DDL 而不是代码里

表结构决定了后面代码能写多干净。我在这个项目里保留四张表：department部门、employee员工、transfer_log调岗记录、sys_user登录账号。字段设计上有三个决定值得说清楚：工资用DECIMAL(10,2)而不是FLOAT，员工状态用status字段做逻辑删除而不是真删，部门和员工之间用外键但设成ON DELETE SET NULL。

-- sql/init.sql CREATE DATABASE IF NOT EXISTS emp_mgr DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci; USE emp_mgr; CREATE TABLE department ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50) NOT NULL UNIQUE COMMENT '部门名称', manager VARCHAR(30) DEFAULT NULL COMMENT '负责人', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE employee ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, emp_no VARCHAR(20) NOT NULL UNIQUE COMMENT '工号', name VARCHAR(30) NOT NULL COMMENT '姓名', gender TINYINT NOT NULL DEFAULT 0 COMMENT '0未知 1男 2女', phone VARCHAR(20) DEFAULT NULL, dept_id INT UNSIGNED DEFAULT NULL, position VARCHAR(50) DEFAULT NULL COMMENT '岗位', salary DECIMAL(10,2) NOT NULL DEFAULT 0.00, hire_date DATE NOT NULL, status TINYINT NOT NULL DEFAULT 1 COMMENT '1在职 0离职', KEY idx_dept (dept_id), KEY idx_name (name), CONSTRAINT fk_emp_dept FOREIGN KEY (dept_id) REFERENCES department(id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE transfer_log ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, emp_id INT UNSIGNED NOT NULL, from_dept INT UNSIGNED DEFAULT NULL, to_dept INT UNSIGNED DEFAULT NULL, reason VARCHAR(200) DEFAULT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_emp (emp_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

DECIMAL(10,2)是硬性要求。用FLOAT存工资，十个人加起来能给你算出12345.679999这种数，导出报表时被问一句「为什么有小数尾巴」就很尴尬。status字段用来做逻辑删除，员工离职后档案还在，历史调岗记录还能追，这是课程设计里少数几个能体现「想过业务」的细节。外键设成SET NULL而不是RESTRICT，是为了删部门时不至于被在职员工卡死，同时又不丢员工数据。

idx_name和idx_dept两个索引对应后面检索功能最常用的两个条件，先建上，别等数据多了再回头补。

2.3 建专用账号并做一次连通性验证

很多教程让人直接用root连库，本地自己玩没问题，但一旦这份代码要交上去、要在别人机器上跑，密码写在config.ini里就很尴尬。我习惯单独建一个只能操作emp_mgr的账号，权限收紧，出问题也好定位。

-- 建应用账号，只给这个库的增删改查权限 CREATE USER 'emp_app'@'localhost' IDENTIFIED BY 'Emp@2024'; GRANT SELECT, INSERT, UPDATE, DELETE ON emp_mgr.* TO 'emp_app'@'localhost'; FLUSH PRIVILEGES;

账号建完，别急着写业务代码。先用二十行脚本把「能不能连上、字符集对不对、版本是多少」这三件事验证掉，把所有环境问题挡在业务代码之外。这一步很关键，相当于给后面所有报错先排除一个变量。

# check_conn.py：最小连通性验证 import pymysql conn = pymysql.connect( host="127.0.0.1", port=3306, user="emp_app", password="Emp@2024", database="emp_mgr", charset="utf8mb4", # 必须和建库时一致，否则中文写入会出问题 cursorclass=pymysql.cursors.DictCursor, ) with conn.cursor() as cur: cur.execute("SELECT VERSION() AS v, @@character_set_database AS cs") print(cur.fetchone()) # 期望 {'v': '8.0.x', 'cs': 'utf8mb4'} conn.close()

charset参数一定要显式写utf8mb4，别指望默认值。cursorclass设成DictCursor之后，查询结果直接是字典，后面 DAO 层取值写成row["name"]比row[1]可读太多。脚本能打印出版本号和utf8mb4，环境这关就算过了，可以动真格的代码。

3. 目录结构与分层：一次查询从菜单到 SQL 的完整链路

小项目最容易写成一坨，全部逻辑塞进main.py，两百行之后自己都看不懂。这份源码按main → service → dao → db四层拆开，依赖方向是单向的，上层可以调下层，下层绝不回头调上层。这么拆不是为了显得专业，而是为了改需求时能只改一个文件：加字段改 model 和 DAO，加校验规则改 service，换数据库只动db.py。

3.1 目录树与各层职责

emp_mgr/ ├── config.ini # 数据库连接配置 ├── requirements.txt ├── main.py # 入口：控制台菜单 + 输入收集 ├── db.py # 连接池 + 事务上下文管理器 ├── models/ │ └── employee.py # 员工实体，负责字段校验 ├── dao/ │ ├── employee_dao.py # 员工表 SQL │ ├── dept_dao.py # 部门表 SQL │ └── user_dao.py # 登录表 SQL ├── service/ │ ├── employee_service.py # 业务规则：调岗、离职、重名校验 │ └── auth_service.py # 登录与密码校验 ├── utils/ │ ├── validator.py # 手机号、工号格式校验 │ └── logger.py # 统一日志输出 └── sql/ └── init.sql

各层职责用一张表说清楚，改代码时先定位改哪一层，比满仓库搜索快得多。

最典型的新手写法是把INSERT INTO employee ...直接写在菜单的if choice == "1"分支里，一开始很快，等到要加一个「工号重复要提示」的规则，就得在三个地方各改一遍。分层之后这条规则只在employee_service.create()里出现一次。

3.2 db.py：连接池与事务上下文

连接池是这份代码里最值得抄的一段。每次操作都pymysql.connect()再close()，在本地单机看不出问题，一旦菜单循环跑起来，MySQL 那边SHOW PROCESSLIST能看到连接数缓慢上涨，最后报Too many connections。DBUtils的PooledDB解决的就是这个。

# db.py import configparser from contextlib import contextmanager from pathlib import Path import pymysql from dbutils.pooled_db import PooledDB _cfg = configparser.ConfigParser() _cfg.read(Path(__file__).parent / "config.ini", encoding="utf-8") _s = _cfg["mysql"] POOL = PooledDB( creator=pymysql, maxconnections=int(_s["max_connections"]), # 池子上限，超出会阻塞等待 mincached=2, # 启动时先建 2 个空闲连接 maxcached=5, # 空闲池最多留 5 个 blocking=True, # 池满时等待而不是直接抛异常 ping=1, # 每次取连接先探活，避免 MySQL 8 小时断连 host=_s["host"], port=int(_s["port"]), user=_s["user"], password=_s["password"], database=_s["database"], charset="utf8mb4", autocommit=False, # 关掉自动提交，事务由上下文管理器控制 cursorclass=pymysql.cursors.DictCursor, ) @contextmanager def get_conn(): """取一个连接，正常结束提交，异常回滚，最后归还连接池。""" conn = POOL.connection() try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() # 注意：这里是把连接还给池子，不是真的关闭

几个参数值得单独说。ping=1的意思是每次从池里拿连接前先发一次心跳，MySQL 默认wait_timeout是 8 小时，课程设计当然跑不了那么久，但很多人是晚上开着程序第二天再点，一操作就(2013, 'Lost connection')，加了这个参数基本绝迹。autocommit=False配合get_conn()的commit/rollback，让「多条 SQL 要么全成功要么全失败」变成默认行为，不用在每个方法里手动写。

finally里的conn.close()是最容易误解的一行。它看起来像关闭连接，实际是归还给连接池。新手看到连接池代码后喜欢把它删掉「省一次关闭」，结果池子里的连接只借不还，跑一会儿就卡死。

3.3 config.ini 与配置读取

配置独立成文件，是为了让代码能换机器跑。别人拿到这份源码，只需要改config.ini里的账号密码，不用去翻db.py。

# config.ini [mysql] host = 127.0.0.1 port = 3306 user = emp_app password = Emp@2024 database = emp_mgr max_connections = 10

max_connections设成 10 对单机控制台程序绰绰有余，池子开太大反而占内存。如果这份代码以后要挂到 Web 服务上给几十个人同时用，这个值要跟着并发量调，并且 MySQL 服务端的max_connections也得留出余量。

提示：config.ini里带明文密码，提交到公开仓库前记得加进.gitignore，再补一个config.ini.example放占位值。

4. 核心模块拆解：CRUD、条件检索与调岗事务怎么写

真正体现水平的不是能查出数据，而是异常路径有没有处理干净。这一章拆三段代码：单表增删改查、带条件的分页检索、跨两张表的调岗事务。三段加起来不到一百行，但把参数化查询、逻辑删除、分页总数、行锁这几件事全占了，抄过去稍作修改就能用在别的管理类项目里。

4.1 员工 DAO：参数化查询与逻辑删除

# dao/employee_dao.py from db import get_conn class EmployeeDao: def insert(self, emp: dict) -> int: sql = ("INSERT INTO employee " "(emp_no, name, gender, phone, dept_id, position, salary, hire_date, status) " "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)") params = ( emp["emp_no"], emp["name"], emp.get("gender", 0), emp.get("phone"), emp.get("dept_id"), emp.get("position"), emp["salary"], emp["hire_date"], emp.get("status", 1), ) with get_conn() as conn, conn.cursor() as cur: cur.execute(sql, params) # 参数交给驱动转义，不拼字符串 return cur.lastrowid # 自增主键，方便后面写调岗记录 def soft_delete(self, emp_id: int) -> int: """逻辑删除：把 status 置 0，档案保留。""" with get_conn() as conn, conn.cursor() as cur: return cur.execute( "UPDATE employee SET status = 0 WHERE id = %s AND status = 1", (emp_id,), ) def get_by_id(self, emp_id: int): with get_conn() as conn, conn.cursor() as cur: cur.execute( "SELECT id, emp_no, name, dept_id, position, salary, hire_date, status " "FROM employee WHERE id = %s", (emp_id,), ) return cur.fetchone() # 查不到返回 None，由 service 层判断

占位符用%s而不是 f-string，这不是风格问题。用f"WHERE name = '{name}'"拼出来的 SQL，遇到名字里带单引号直接语法错误，遇到恶意输入就是注入漏洞，课程设计答辩时被问一句「这里有没有 SQL 注入」就答不上来。%s在 PyMySQL 里是通用占位符，不管字段是字符串还是数字都用它，由驱动负责转义。

soft_delete返回的是受影响行数。这个返回值要接着用：返回 0 说明该员工本来就不存在或已经离职，service 层据此提示「操作失败」，而不是无脑打印成功。lastrowid在插入后立刻取，能拿到自增主键，写调岗日志时正好需要。

4.2 条件检索与分页：两段 SQL 必须共用同一个 WHERE

分页是课程设计里最容易写出 bug 的地方。常见错误是列表 SQL 带了检索条件，统计总数的 SQL 忘了带，结果页面上显示「共 200 条」，翻到第三页却没数据。

# dao/employee_dao.py（续） def search(self, keyword: str = None, dept_id: int = None, page: int = 1, size: int = 10): where = ["e.status = 1"] # 默认只看在职 params = [] if keyword: where.append("(e.name LIKE %s OR e.emp_no LIKE %s)") like = f"%{keyword}%" params.extend([like, like]) if dept_id: where.append("e.dept_id = %s") params.append(dept_id) where_sql = " WHERE " + " AND ".join(where) count_sql = "SELECT COUNT(*) AS total FROM employee e" + where_sql data_sql = ( "SELECT e.id, e.emp_no, e.name, d.name AS dept_name, " " e.position, e.salary, e.hire_date " "FROM employee e LEFT JOIN department d ON e.dept_id = d.id" + where_sql + " ORDER BY e.id DESC LIMIT %s OFFSET %s" ) page = max(1, int(page)) size = min(100, max(1, int(size))) # 限制每页条数，防止误传 100000 offset = (page - 1) * size with get_conn() as conn, conn.cursor() as cur: cur.execute(count_sql, params) total = cur.fetchone()["total"] if total == 0: return 0, [] cur.execute(data_sql, params + [size, offset]) return total, cur.fetchall()

结构上把 WHERE 片段抽出来，让统计和取数共用，是这段代码的核心。params列表也是共用的，只在最后追加size和offset，顺序不能乱——LIMIT %s OFFSET %s的占位符和参数是一一对应的。

page和size做了范围收敛。用户输入-1或者abc时，int()会抛异常或算出负的 offset，MySQL 直接报语法错误。size上限设 100，避免有人手滑输入100000把内存打满。total == 0时提前返回，省掉一次没有意义的查询。

用LEFT JOIN而不是INNER JOIN，是因为员工的dept_id允许为空（部门被删时置 NULL），用INNER JOIN这些人会从列表里凭空消失，属于典型的「数据看起来少了但查不出原因」。

4.3 调岗事务：两条 SQL 要么都成要么都不成

调岗要改员工表的部门，同时往transfer_log插一条记录。这两步必须原子，否则改完部门但日志没写，历史就断了。

# service/employee_service.py from db import get_conn def transfer(emp_id: int, to_dept: int, reason: str) -> bool: """调岗：改部门 + 写日志，同一事务内完成。""" with get_conn() as conn: with conn.cursor() as cur: # 加行锁，防止两个人同时调同一个员工 cur.execute( "SELECT dept_id FROM employee WHERE id = %s AND status = 1 FOR UPDATE", (emp_id,), ) row = cur.fetchone() if not row: raise ValueError("员工不存在或已离职") from_dept = row["dept_id"] if from_dept == to_dept: return False # 同部门调动，直接返回 cur.execute("UPDATE employee SET dept_id = %s WHERE id = %s", (to_dept, emp_id)) cur.execute( "INSERT INTO transfer_log (emp_id, from_dept, to_dept, reason) " "VALUES (%s, %s, %s, %s)", (emp_id, from_dept, to_dept, reason), ) return True

两条 SQL 之间没有手动commit，全靠get_conn()这个上下文管理器：正常走完就提交，中途抛异常就整体回滚。这就是把事务边界收到一处的价值——业务代码里看不到commit，但每个方法天然是原子的。

FOR UPDATE是行级锁，锁住这一行直到事务结束。控制台单人操作其实用不上，但加上它有两个好处：一是避免「两个人同时调同一个员工，日志里出现两条 from_dept 相同」的脏数据；二是答辩时能讲清楚乐观锁和悲观锁的区别。raise ValueError抛出去之后不要再try/except吞掉，一旦吞掉，上下文管理器就认为一切正常并提交，等于白写。

5. 常见问题排查：五个让课程设计当场翻车的坑

下面这几条都是真跑起来才会遇到的，按「现象 → 原因 → 解决」整理。每条都值得在提交前自查一遍，尤其是第一条和第二条，出问题时的报错信息完全不指向真实原因，查起来很费时间。

5.1 现象：列表里的中文全是问号???

原因通常有三层：数据库建库时没指定utf8mb4；连接串里没写charset；Windows 控制台默认 GBK 编码显示不了。三者任一不满足都会出问号，所以经常改了连接串还是问号，因为库本身就不是 utf8mb4。

解决顺序是从底往上查。先执行SHOW VARIABLES LIKE 'character_set_database';确认是utf8mb4，不是就ALTER DATABASE emp_mgr CHARACTER SET utf8mb4;；再确认pymysql.connect(...)里显式带了charset="utf8mb4"；最后在 Windows 上执行chcp 65001把控制台切到 UTF-8。三步都做对，中文一定正常。

5.2 现象：程序跑十几分钟后卡死，或者报Too many connections

原因几乎都是连接没归还。最典型的写法是在 DAO 里conn = POOL.connection()之后，中间某个return提前退出，conn.close()永远执行不到；另一种是try里抛了异常，finally没写。池子里的连接被借光后，blocking=True会让下一次取连接一直等，表现出来就是「程序无响应」。

解决是把取连接的动作全部收进with get_conn()上下文管理器，业务代码里不出现裸的POOL.connection()。排查时可以另开一个客户端执行SHOW PROCESSLIST;，如果看到十几个Sleep状态的连接全是同一个账号，基本可以确诊。这个坑我给它的评价是「后悔药都来不及吃」，因为一旦池子泄漏，只能重启进程。

5.3 现象：翻到第二页报You have an error in your SQL syntax near ''10''

原因是分页参数被当成字符串拼进去了，生成的 SQL 变成LIMIT '10' OFFSET '0'，MySQL 的LIMIT不接受带引号的数字。这种情况一般出现在混用了%s和字符串格式化的代码里，比如先用 f-string 把size塞进 SQL，又用%s传offset。

解决是LIMIT %s OFFSET %s全部走参数，并且在传之前int()转换一次。别用 f-string 拼这两个值，虽然数字看似安全，但一旦上游传进来的是"10; DROP TABLE"就麻烦了。分页参数永远从用户输入到 SQL 之间做一次类型收敛。

5.4 现象：删除部门时报Cannot delete or update a parent row: a foreign key constraint fails

原因是有在职员工的dept_id还指向这个部门，外键约束拦住了删除。如果建表时写的是ON DELETE RESTRICT（默认行为），这是预期内的保护；如果是SET NULL，说明约束在别的表上，比如transfer_log里存了部门 ID 但没建外键或者建了RESTRICT。

解决分两种情况：业务上允许部门下还有人就先调岗再删，我在 service 层加了一步「查该部门在职人数，大于 0 直接拒绝并提示」；确实要强删，就把外键改成ON DELETE SET NULL，同时接受历史调岗记录里的部门 ID 变成孤儿数据。别用SET FOREIGN_KEY_CHECKS=0绕过，那是把问题藏进黑匣子。

5.5 现象：工资合计导出后带一长串小数，或者对不上账

原因是金额字段用了FLOAT或DOUBLE。浮点数是二进制近似存储，0.1 + 0.2在浮点里不等于0.3，几十条记录累加下来偏差肉眼可见。另一个来源是 Python 端用float接收DECIMAL字段，一样会丢精度。

解决是建表就用DECIMAL(10,2)，Python 端查询结果保持字符串或转成decimal.Decimal再计算，不要中途float()。如果已经有数据是FLOAT，用ALTER TABLE employee MODIFY salary DECIMAL(10,2) NOT NULL DEFAULT 0.00;迁一次，MySQL 会自动四舍五入到两位。

6. 二次开发改造：把课程设计往「能讲」的方向推一步

源码跑通之后，多数人会停在这里交作业。但如果想让这份东西在面试里能讲上五分钟，我建议动三个地方，成本都不高。

第一个是加一层接口。service层的函数签名本来就跟 HTTP 无关，直接在上面套一个 Flask 蓝图，/api/employees走GET分页、POST新增，/api/employees/<id>/transfer走调岗，代码几乎不用改，只是把main.py里的input()换成request.json。db.py的连接池正好复用，不用重写。这一步做完，项目从「控制台程序」变成「前后端分离的后端」，讲的时候能说清楚分层带来的收益。

第二个是补权限和审计。现在sys_user表只有账号密码，加一个role字段区分管理员和普通员工，管理员能删改、普通员工只能查自己。密码别存明文，用hashlib.pbkdf2_hmac加盐存哈希，登录时比对。审计这块，transfer_log已经是个雏形，照着它再加一张operation_log，记录谁在什么时候改了哪条记录，出问题能追。

第三个是补测试。至少给EmployeeDao.search()和transfer()写两个用例：一个验证带关键词和部门的组合查询总数正确，一个验证调岗中途抛异常时员工部门没变、日志也没写。第二个用例尤其值钱，它是验证事务边界有没有写对的唯一手段。用pytest加一个测试库emp_mgr_test，每次跑之前TRUNCATE一遍，比手动点菜单靠谱得多。

改造的验证方法很简单：改完一层就跑一次原来的控制台流程，功能不变说明重构没破坏行为；再跑一遍pytest，全绿说明事务和分页逻辑还稳。

这类小系统我接手过几次，血泪经验是：先定表结构和连接管理，再写业务代码，顺序反过来一定返工。从那以后我每次开新项目，都强制先把init.sql和db.py写完、跑通连通性脚本，再动第一行业务逻辑。希望帮到你。

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