FastAPI的项目结构--员工管理系统
一、项目结构长这样
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="添加部门")asyncdefbumen_add(bumen:BuMenadd):awaitBuMen.add_bumen(bumen)return{"code":"200","msg":"添加部门成功"}具体的"部门是否已存在"这类判断,都封装进了BuMen.add_bumen()内部,路由层完全不用关心。
二:五大模块
2.1 登录:账号密码硬编码校验
需求里登录只要求用户名admin、密码123456能登录成功,属于最简单的一种登录场景。项目里沿用了之前 Day03 的用户模型思路:按用户名查库,再比对密码是否一致,密码错误和用户不存在分开给出提示。这种"教学级"的登录逻辑重点在于走通登录流程,实际生产环境密码肯定不能明文存储和比对,得配合哈希加密(比如 bcrypt)。
2.2 部门管理:单表 CRUD + 一个关键的删除保护
部门模型很简单,就是名称、位置、电话几个字段:
classDepartment(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)需求里要求部门列表要显示"员工数量",这就需要在查询时带上关联的员工数据:
@staticmethodasyncdefbumenlist():bumen=awaitmodels.Department.all().prefetch_related("employees_department")bumen_list=[]foriinbumen:count=0forjini.employees_department:ifj.status!="离职":count+=1bumen_list.append({...,"elements_count":count})returnbumen_list这里有个小细节:员工数量统计的是"在职"人数,不是简单的len(),需要遍历判断status字段再计数,因为离职员工的记录还留在库里,只是状态变了。
删除部门时按需求要做保护——部门下还有员工就不能删:
@staticmethodasyncdefdelete_bumen(id:int):bumens=awaitmodels.Department.filter(id=id).first()ifnotbumens:raiseException("部门不存在")yuangong=awaitmodels.Employee.filter(department_id=id).first()ifyuangong:raiseException("部门下有员工,无法删除")awaitbumens.delete()return200这里用了raise Exception()而不是直接return错误字典,是服务层和路由层职责分离后的一种写法——服务层专心做业务判断,把"错误"作为异常抛出,具体怎么包装成 HTTP 响应交给上层处理(或者配合 FastAPI 的异常处理器统一转换成标准错误格式)。
2.3 员工管理:分页 + 多条件筛选是重点
员工是这次改动最大的模块,模型里除了基础字段,还有一个指向部门的外键(多对一关系):
classEmployee(models.Model):...department=fields.ForeignKeyField("models.Department",related_name="employees_department")status=fields.CharField(max_length=4,default="在职")需求要求支持按姓名模糊搜索、按部门筛选、按状态筛选,并且要分页,每页 10 条。查询方法把这几个条件按需叠加:
@staticmethodasyncdefyuangonglist(name,department,status,page,size):offset=(page-1)*size query=models.Employee.all().prefetch_related("department").offset(offset).limit(size)ifname:query=query.filter(name__icontains=name)ifdepartment:query=query.filter(department_id=department)ifstatus:query=query.filter(status=status)yuangongs=awaitquery...分页的核心就是offset和limit两个方法:offset = (page - 1) * size算出要跳过多少条,limit(size)限制每页取多少条。路由层用 FastAPI 的Query给分页参数加上默认值和最小值校验:
asyncdefyuangong_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=awaitmodels.Department.filter(id=yuangong.department).first()ifnotdepartment:raiseException("部门不存在")删除员工,需求特别提到"同时删除关联的档案"。因为Employee_Profiles对Employee是OneToOneField,数据库层面的外键约束本身就是ON DELETE CASCADE,所以只要删掉Employee记录,关联的档案会被数据库自动级联清理,服务层不需要手动多写一步删除档案的代码。
2.4 员工档案:一对一关系的"只能有一份"约束
档案模型用OneToOneField关联员工:
classEmployee_Profiles(models.Model):employee=fields.OneToOneField("models.Employee",related_name="employee_profiles_employee")id_card=fields.CharField(max_length=18,null=True)...需求要求"每个员工只能有一份档案,已创建的不能重复创建",这个约束需要在业务逻辑里主动检查,因为一对一关系在数据库层面虽然天然保证了唯一性(重复插入会报错),但更友好的做法是提前查一遍,给出清晰的错误提示而不是让数据库异常直接抛出来:
@staticmethodasyncdefadd_dangan(dangan:DangAnadd,id:int):yuangong=awaitmodels.Employee.filter(id=id).first()ifnotyuangong:raiseException("员工不存在")dangans=awaitmodels.Employee_Profiles.filter(employee_id=id).first()ifdangans:raiseException("员工已存在档案")awaitmodels.Employee_Profiles.create(employee_id=id,**dangan.dict())查询和修改档案的逻辑跟之前 User_Profiles 的思路完全一致,都是先确认员工存在、再确认档案存在,最后用exclude_unset=True+setattr()做局部更新。
2.5 首页统计:一个轻量的聚合接口
统计接口需求不复杂——总员工数、在职员工数、部门数:
asyncdeftongji():yuangong=awaitmodels.Employee.all()bumen=awaitmodels.Department.all()shu=0foriinyuangong:ifi.status!="离职":shu+=1return{"yuangong_count":len(yuangong),"bumen_count":len(bumen),"yuangong_status_count":shu}这里是把全部数据取出来在 Python 层面做计数,数据量小的时候没问题;如果员工数据量很大,更合理的方式是用 ORM 自带的聚合方法(比如annotate配合Count)直接在数据库层面算出结果,减少数据传输和内存占用,这是接下来准备深入的方向。
三、这次实践的几点收获
一是服务层拆分带来的好处:路由文件只剩"收参数、调服务、包装返回"三步,业务逻辑集中在 service 类里,测试和维护都更方便。二是无论是部门删除保护还是档案唯一性校验,本质上都是同一种模式——在真正执行数据库写操作前,先用一次查询确认前置条件是否满足,这种"先查后写"的防御性写法在关联表场景里几乎是标配。三是分页和多条件筛选的组合写法也算是摸清楚了套路:所有可选条件统一用if判断后再叠加到query上,最后再await一次性执行。
接下来准备把统计接口换成数据库聚合函数的写法,顺便补上事务管理,确保像"删除部门连带校验"这种多步操作在异常情况下也能保持数据一致性。