一、项目结构长这样
demo0720/
└── app/
├── core/ # 核心配置(如通用工具、常量等)
├── models/ # 数据模型层
│ ├── __init__.py
│ ├── user.py # 用户/文章/分类/标签模型
│ └── yuangong.py # 部门/员工/员工档案模型
├── routers/ # API 路由层
│ ├── __init__.py
│ ├── article_api.py
│ ├── category_api.py
│ ├── tag_api.py
│ ├── user_api.py
│ └── yuangong_api.py # 员工管理相关接口
├── schemas/ # Pydantic 请求/响应校验层
│ ├── article.py
│ ├── category.py
│ ├── tag.py
│ ├── user.py
│ └── yuangong.py
├── services/ # 业务逻辑层(新加的一层)
│ └── yuangong.py # 部门/员工/档案的业务逻辑
├── config.py
└── main.py
├── migrations/
├── pyproject.toml
└── test_main.http
跟前几天的项目相比,这次多了一个 services 目录。之前路由函数里直接写数据库操作,逻辑简单还好,但员工管理这块涉及分页、多条件筛选、跨表校验,直接堆在路由函数里会很臃肿,所以拆出了一层"服务层":路由只负责接收请求、调用服务、包装响应,具体的业务逻辑(查询条件拼接、存在性校验、异常处理)都下沉到 services/yuangong.py 里的 BuMen、YuanGong、DangAn 三个类中,每个类用 @staticmethod 组织一组相关方法。这样路由文件读起来非常清爽:
@yuangong_router.post("/bumen_add", summary="添加部门")
async def bumen_add(bumen: BuMenadd):
await BuMen.add_bumen(bumen)
return {"code": "200", "msg": "添加部门成功"}
具体的"部门是否已存在"这类判断,都封装进了 BuMen.add_bumen() 内部,路由层完全不用关心。
二:五大模块
2.1 登录:账号密码硬编码校验
需求里登录只要求用户名 admin、密码 123456 能登录成功,属于最简单的一种登录场景。项目里沿用了之前 Day03 的用户模型思路:按用户名查库,再比对密码是否一致,密码错误和用户不存在分开给出提示。这种"教学级"的登录逻辑重点在于走通登录流程,实际生产环境密码肯定不能明文存储和比对,得配合哈希加密(比如 bcrypt)。
2.2 部门管理:单表 CRUD + 一个关键的删除保护
部门模型很简单,就是名称、位置、电话几个字段:
class Department(models.Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=50, unique=True)
location = fields.CharField(max_length=100, null=True)
phone = fields.CharField(max_length=20, null=True)
需求里要求部门列表要显示"员工数量",这就需要在查询时带上关联的员工数据:
@staticmethod
async def bumenlist():
bumen = await models.Department.all().prefetch_related("employees_department")
bumen_list = []
for i in bumen:
count = 0
for j in i.employees_department:
if j.status != "离职":
count += 1
bumen_list.append({..., "elements_count": count})
return bumen_list
这里有个小细节:员工数量统计的是"在职"人数,不是简单的 len(),需要遍历判断 status 字段再计数,因为离职员工的记录还留在库里,只是状态变了。
删除部门时按需求要做保护——部门下还有员工就不能删:
@staticmethod
async def delete_bumen(id: int):
bumens = await models.Department.filter(id=id).first()
if not bumens:
raise Exception("部门不存在")
yuangong = await models.Employee.filter(department_id=id).first()
if yuangong:
raise Exception("部门下有员工,无法删除")
await bumens.delete()
return 200
这里用了 raise Exception() 而不是直接 return 错误字典,是服务层和路由层职责分离后的一种写法——服务层专心做业务判断,把"错误"作为异常抛出,具体怎么包装成 HTTP 响应交给上层处理(或者配合 FastAPI 的异常处理器统一转换成标准错误格式)。
2.3 员工管理:分页 + 多条件筛选是重点
员工是这次改动最大的模块,模型里除了基础字段,还有一个指向部门的外键(多对一关系):
class Employee(models.Model):
...
department = fields.ForeignKeyField("models.Department", related_name="employees_department")
status = fields.CharField(max_length=4, default="在职")
需求要求支持按姓名模糊搜索、按部门筛选、按状态筛选,并且要分页,每页 10 条。查询方法把这几个条件按需叠加:
@staticmethod
async def yuangonglist(name, department, status, page, size):
offset = (page – 1) * size
query = models.Employee.all().prefetch_related("department").offset(offset).limit(size)
if name:
query = query.filter(name__icontains=name)
if department:
query = query.filter(department_id=department)
if status:
query = query.filter(status=status)
yuangongs = await query
...
分页的核心就是 offset 和 limit 两个方法:offset = (page – 1) * size 算出要跳过多少条,limit(size) 限制每页取多少条。路由层用 FastAPI 的 Query 给分页参数加上默认值和最小值校验:
async def yuangong_list(name: str = None, department: int = None, status: str = None,
page: int = Query(1, ge=1),
size: int = Query(5, ge=5)):
值得一提的是所有筛选条件都用 if xxx: 判断了一下是否传值,避免把 None 当成筛选条件传给 filter(),不然会误判成"筛选出字段为空的记录"。
新增员工时同样要先校验工号唯一、部门是否存在,跟前几天文章模块校验外键的思路一模一样:
department = await models.Department.filter(id=yuangong.department).first()
if not department:
raise Exception("部门不存在")
删除员工,需求特别提到"同时删除关联的档案"。因为 Employee_Profiles 对 Employee 是 OneToOneField,数据库层面的外键约束本身就是 ON DELETE CASCADE,所以只要删掉 Employee 记录,关联的档案会被数据库自动级联清理,服务层不需要手动多写一步删除档案的代码。
2.4 员工档案:一对一关系的"只能有一份"约束
档案模型用 OneToOneField 关联员工:
class Employee_Profiles(models.Model):
employee = fields.OneToOneField("models.Employee", related_name="employee_profiles_employee")
id_card = fields.CharField(max_length=18, null=True)
...
需求要求"每个员工只能有一份档案,已创建的不能重复创建",这个约束需要在业务逻辑里主动检查,因为一对一关系在数据库层面虽然天然保证了唯一性(重复插入会报错),但更友好的做法是提前查一遍,给出清晰的错误提示而不是让数据库异常直接抛出来:
@staticmethod
async def add_dangan(dangan: DangAnadd, id: int):
yuangong = await models.Employee.filter(id=id).first()
if not yuangong:
raise Exception("员工不存在")
dangans = await models.Employee_Profiles.filter(employee_id=id).first()
if dangans:
raise Exception("员工已存在档案")
await models.Employee_Profiles.create(employee_id=id, **dangan.dict())
查询和修改档案的逻辑跟之前 User_Profiles 的思路完全一致,都是先确认员工存在、再确认档案存在,最后用 exclude_unset=True + setattr() 做局部更新。
2.5 首页统计:一个轻量的聚合接口
统计接口需求不复杂——总员工数、在职员工数、部门数:
async def tongji():
yuangong = await models.Employee.all()
bumen = await models.Department.all()
shu = 0
for i in yuangong:
if i.status != "离职":
shu += 1
return {
"yuangong_count": len(yuangong),
"bumen_count": len(bumen),
"yuangong_status_count": shu
}
这里是把全部数据取出来在 Python 层面做计数,数据量小的时候没问题;如果员工数据量很大,更合理的方式是用 ORM 自带的聚合方法(比如 annotate 配合 Count)直接在数据库层面算出结果,减少数据传输和内存占用,这是接下来准备深入的方向。
三、这次实践的几点收获
一是服务层拆分带来的好处:路由文件只剩"收参数、调服务、包装返回"三步,业务逻辑集中在 service 类里,测试和维护都更方便。二是无论是部门删除保护还是档案唯一性校验,本质上都是同一种模式——在真正执行数据库写操作前,先用一次查询确认前置条件是否满足,这种"先查后写"的防御性写法在关联表场景里几乎是标配。三是分页和多条件筛选的组合写法也算是摸清楚了套路:所有可选条件统一用 if 判断后再叠加到 query 上,最后再 await 一次性执行。
接下来准备把统计接口换成数据库聚合函数的写法,顺便补上事务管理,确保像"删除部门连带校验"这种多步操作在异常情况下也能保持数据一致性。
