FastAPI的项目结构--员工管理系统

发布时间:2026/7/22 3:39:41
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):idfields.IntField(pkTrue)namefields.CharField(max_length50,uniqueTrue)locationfields.CharField(max_length100,nullTrue)phonefields.CharField(max_length20,nullTrue)需求里要求部门列表要显示员工数量这就需要在查询时带上关联的员工数据staticmethodasyncdefbumenlist():bumenawaitmodels.Department.all().prefetch_related(employees_department)bumen_list[]foriinbumen:count0forjini.employees_department:ifj.status!离职:count1bumen_list.append({...,elements_count:count})returnbumen_list这里有个小细节员工数量统计的是在职人数不是简单的len()需要遍历判断status字段再计数因为离职员工的记录还留在库里只是状态变了。删除部门时按需求要做保护——部门下还有员工就不能删staticmethodasyncdefdelete_bumen(id:int):bumensawaitmodels.Department.filter(idid).first()ifnotbumens:raiseException(部门不存在)yuangongawaitmodels.Employee.filter(department_idid).first()ifyuangong:raiseException(部门下有员工无法删除)awaitbumens.delete()return200这里用了raise Exception()而不是直接return错误字典是服务层和路由层职责分离后的一种写法——服务层专心做业务判断把错误作为异常抛出具体怎么包装成 HTTP 响应交给上层处理或者配合 FastAPI 的异常处理器统一转换成标准错误格式。2.3 员工管理分页 多条件筛选是重点员工是这次改动最大的模块模型里除了基础字段还有一个指向部门的外键多对一关系classEmployee(models.Model):...departmentfields.ForeignKeyField(models.Department,related_nameemployees_department)statusfields.CharField(max_length4,default在职)需求要求支持按姓名模糊搜索、按部门筛选、按状态筛选并且要分页每页 10 条。查询方法把这几个条件按需叠加staticmethodasyncdefyuangonglist(name,department,status,page,size):offset(page-1)*size querymodels.Employee.all().prefetch_related(department).offset(offset).limit(size)ifname:queryquery.filter(name__icontainsname)ifdepartment:queryquery.filter(department_iddepartment)ifstatus:queryquery.filter(statusstatus)yuangongsawaitquery...分页的核心就是offset和limit两个方法offset (page - 1) * size算出要跳过多少条limit(size)限制每页取多少条。路由层用 FastAPI 的Query给分页参数加上默认值和最小值校验asyncdefyuangong_list(name:strNone,department:intNone,status:strNone,page:intQuery(1,ge1),size:intQuery(5,ge5)):值得一提的是所有筛选条件都用if xxx:判断了一下是否传值避免把None当成筛选条件传给filter()不然会误判成筛选出字段为空的记录。新增员工时同样要先校验工号唯一、部门是否存在跟前几天文章模块校验外键的思路一模一样departmentawaitmodels.Department.filter(idyuangong.department).first()ifnotdepartment:raiseException(部门不存在)删除员工需求特别提到同时删除关联的档案。因为Employee_Profiles对Employee是OneToOneField数据库层面的外键约束本身就是ON DELETE CASCADE所以只要删掉Employee记录关联的档案会被数据库自动级联清理服务层不需要手动多写一步删除档案的代码。2.4 员工档案一对一关系的只能有一份约束档案模型用OneToOneField关联员工classEmployee_Profiles(models.Model):employeefields.OneToOneField(models.Employee,related_nameemployee_profiles_employee)id_cardfields.CharField(max_length18,nullTrue)...需求要求每个员工只能有一份档案已创建的不能重复创建这个约束需要在业务逻辑里主动检查因为一对一关系在数据库层面虽然天然保证了唯一性重复插入会报错但更友好的做法是提前查一遍给出清晰的错误提示而不是让数据库异常直接抛出来staticmethodasyncdefadd_dangan(dangan:DangAnadd,id:int):yuangongawaitmodels.Employee.filter(idid).first()ifnotyuangong:raiseException(员工不存在)dangansawaitmodels.Employee_Profiles.filter(employee_idid).first()ifdangans:raiseException(员工已存在档案)awaitmodels.Employee_Profiles.create(employee_idid,**dangan.dict())查询和修改档案的逻辑跟之前 User_Profiles 的思路完全一致都是先确认员工存在、再确认档案存在最后用exclude_unsetTruesetattr()做局部更新。2.5 首页统计一个轻量的聚合接口统计接口需求不复杂——总员工数、在职员工数、部门数asyncdeftongji():yuangongawaitmodels.Employee.all()bumenawaitmodels.Department.all()shu0foriinyuangong:ifi.status!离职:shu1return{yuangong_count:len(yuangong),bumen_count:len(bumen),yuangong_status_count:shu}这里是把全部数据取出来在 Python 层面做计数数据量小的时候没问题如果员工数据量很大更合理的方式是用 ORM 自带的聚合方法比如annotate配合Count直接在数据库层面算出结果减少数据传输和内存占用这是接下来准备深入的方向。三、这次实践的几点收获一是服务层拆分带来的好处路由文件只剩收参数、调服务、包装返回三步业务逻辑集中在 service 类里测试和维护都更方便。二是无论是部门删除保护还是档案唯一性校验本质上都是同一种模式——在真正执行数据库写操作前先用一次查询确认前置条件是否满足这种先查后写的防御性写法在关联表场景里几乎是标配。三是分页和多条件筛选的组合写法也算是摸清楚了套路所有可选条件统一用if判断后再叠加到query上最后再await一次性执行。接下来准备把统计接口换成数据库聚合函数的写法顺便补上事务管理确保像删除部门连带校验这种多步操作在异常情况下也能保持数据一致性。