告别配置地狱：员工管理系统保姆级教程，3步跑通

配置环境就卡半天，是不是你的常态？依赖冲突、端口占用、数据库连不上，一个下午就没了。别慌，今天这篇保姆级教程，不讲虚的，直接带你从零搭建一个能用的员工管理系统。

项目目标与场景拆解

很多技术博主一上来就甩架构图，但对于刚起步的团队，最痛点其实是“数据散落在Excel里”。我们要做的系统，核心就解决三个问题：人员信息入库、权限分级查看、操作日志留痕。

这里有个容易被忽视的合规细节。根据《个人信息保护法》，员工信息属于敏感个人数据，系统必须做到最小权限原则。也就是说，HR能看全量数据，但普通经理只能看本部门。这不仅是功能需求，更是法律底线。在后续代码中，我们会专门针对这一层做拦截，而不是靠前端隐藏按钮，因为前端隐藏是骗不了抓包的。

我们的技术选型很克制：后端用 Python FastAPI，前端用 Vue3，数据库选 PostgreSQL。为什么选这个组合？因为 FastAPI 自带类型提示，配合 Pydantic 模型，数据校验几乎零成本；PostgreSQL 的 JSONB 类型能灵活存储员工的各种扩展字段，比如“紧急联系人”、“技能标签”，不用频繁改表结构。

目录结构与设计思路

好的工程结构是维护性的基础。很多新手喜欢把所有代码扔在 main.py 里，那是灾难的开始。我们采用分层架构，严格分离职责。

project-root/
├── app/
│   ├── api/
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── deps.py      # 依赖注入：获取DB会话、当前用户
│   │   │   └── endpoints/
│   │   │       ├── __init__.py
│   │   │       └── employees.py # 员工相关API
│   ├── core/
│   │   ├── config.py        # 配置管理，读取.env
│   │   └── security.py      # JWT生成与验证
│   ├── db/
│   │   ├── base.py          # 数据库连接
│   │   ├── init_db.py       # 初始化表结构
│   │   └── session.py
│   ├── models/
│   │   └── employee.py      # SQLAlchemy ORM模型
│   ├── schemas/
│   │   └── employee.py      # Pydantic请求/响应模型
│   └── main.py              # FastAPI应用入口
├── tests/
│   └── test_employees.py
├── .env.example             # 环境变量模板
├── requirements.txt
└── README.md

注意 deps.py 文件。这是 FastAPI 的精髓所在。我们将“获取数据库连接”和“获取当前登录用户”封装成依赖函数，这样在每一个接口中，只需要在参数里加一个 db: Session = Depends(get_db)，逻辑就清晰多了。这种解耦方式，后期想换数据库或者改认证方式，只需改依赖，不用动业务逻辑。

核心代码实现详解

1. 数据模型定义

先看模型。这里我们不仅定义字段，还定义了关系。员工表需要关联部门表，这是多对一关系。

# app/models/employee.py
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, func
from app.db.base import Baseclass Employee(Base):__tablename__ = "employees"id = Column(Integer, primary_key=True, index=True)name = Column(String(50), index=True, nullable=False)email = Column(String(100), unique=True, index=True, nullable=False)department_id = Column(Integer, ForeignKey("departments.id"))role = Column(String(20), default="employee") # 角色：employee, manager, hrcreated_at = Column(DateTime(timezone=True), server_default=func.now())# 关系定义，懒加载department = relationship("Department", back_populates="employees")

对应的 Pydantic 模型，用于数据校验。注意 Config 类中的 from_orm = True，这是为了让 Pydantic 直接从 ORM 对象转换，避免手动赋值。

# app/schemas/employee.py
from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optionalclass EmployeeBase(BaseModel):name: stremail: EmailStrdepartment_id: introle: str = "employee"class EmployeeCreate(EmployeeBase):passclass EmployeeResponse(EmployeeBase):id: intcreated_at: datetimeclass Config:from_orm = True

2. 依赖注入与权限控制

这是最关键的避坑点。很多教程只讲怎么查数据，不讲怎么防止越权。我们在 deps.py 中实现了一个 get_current_user 依赖，它解析 JWT Token，返回当前用户对象。

# app/api/v1/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.core.security import verify_token
from app.models.user import Useroauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User:credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Could not validate credentials",headers={"WWW-Authenticate": "Bearer"},)try:payload = verify_token(token)user_id: int = payload.get("sub")if user_id is None:raise credentials_exceptionexcept Exception:raise credentials_exceptionuser = db.query(User).filter(User.id == user_id).first()if user is None:raise credentials_exceptionreturn user

在员工接口中，我们加入权限判断。如果是 HR，可以看全部；如果是 Manager，只能看自己部门的。

# app/api/v1/endpoints/employees.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from app.api.v1.deps import get_db, get_current_user
from app.models.employee import Employee
from app.schemas.employee import EmployeeCreate, EmployeeResponse
from app.models.user import Userrouter = APIRouter()@router.get("/employees", response_model=List[EmployeeResponse])
def read_employees(db: Session = Depends(get_db),current_user: User = Depends(get_current_user)
):query = db.query(Employee)# 核心逻辑：权限过滤if current_user.role == "hr":pass # HR看全部elif current_user.role == "manager":# 假设当前用户是某个部门的经理，这里简化处理# 实际项目中应关联 user -> department 关系query = query.filter(Employee.department_id == current_user.department_id)else:raise HTTPException(status_code=403, detail="Insufficient permissions")return query.all()@router.post("/employees", response_model=EmployeeResponse)
def create_employee(employee_in: EmployeeCreate,db: Session = Depends(get_db),current_user: User = Depends(get_current_user)
):# 只有HR可以创建if current_user.role != "hr":raise HTTPException(status_code=403, detail="Only HR can create employees")db_employee = Employee(**employee_in.dict())db.add(db_employee)db.commit()db.refresh(db_employee)return db_employee

运行与测试验证

代码写完，别急着跑，先测。使用 httpx 配合 pytest 是 FastAPI 的标准测试姿势。

# tests/test_employees.py
import pytest
from fastapi.testclient import TestClient
from app.main import appclient = TestClient(app)def test_create_employee_with_hr_token():# 1. 获取HR的Token (模拟登录)response = client.post("/token", data={"username": "hr_test", "password": "pwd"})assert response.status_code == 200token = response.json()["access_token"]# 2. 创建员工headers = {"Authorization": f"Bearer {token}"}data = {"name": "Zhang San","email": "zhangsan@example.com","department_id": 1,"role": "employee"}response = client.post("/employees", json=data, headers=headers)assert response.status_code == 200assert response.json()["name"] == "Zhang San"def test_create_employee_with_normal_user_token():# 1. 获取普通员工Tokenresponse = client.post("/token", data={"username": "user_test", "password": "pwd"})token = response.json()["access_token"]headers = {"Authorization": f"Bearer {token}"}# 2. 尝试创建，应返回403data = {"name": "Li Si", "email": "lisi@example.com", "department_id": 1}response = client.post("/employees", json=data, headers=headers)assert response.status_code == 403

运行测试命令：pytest -v。如果看到绿色对勾，说明核心逻辑通了。如果卡在数据库连接，检查 .env 文件中的 DATABASE_URL 是否正确，特别是本地开发时，确保 PostgreSQL 服务已启动且端口未被占用。

优化扩展与避坑指南

系统能跑起来只是第一步，稳定运行才是关键。这里有几个实战中踩过的坑，务必注意。

1. 数据库连接池配置

FastAPI 是异步框架，但 SQLAlchemy 同步引擎在高并发下会阻塞事件循环。虽然本文为了简化使用了同步写法，但在生产环境，建议迁移到 asyncpg 并使用异步引擎。配置连接池参数至关重要：

# app/db/session.py (异步版本示例)
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from app.core.config import settingsengine = create_async_engine(settings.DATABASE_URL,echo=False,pool_size=20,       # 连接池大小max_overflow=10,    # 最大溢出连接数pool_timeout=30,    # 获取连接超时时间pool_recycle=1800   # 连接回收时间，防止数据库超时断开
)

2. 日志与审计

员工数据的修改必须留痕。不要只打 print，使用 Python 标准的 logging 模块，并配置 RotatingFileHandler，防止日志文件无限增大撑爆磁盘。对于敏感操作（如删除员工、修改薪资字段），建议单独记录一张审计日志表，包含操作人、操作时间、变更前后的 JSON 快照。

3. 敏感数据脱敏

API 返回员工手机号、身份证号时，必须脱敏。在 Pydantic 模型中使用 @validator 或 Field 的 repr 属性，或者在后端序列化前手动处理。例如，手机号 13800138000 返回为 138****8000。这不仅是安全规范，也是合规要求。参考 OWASP 的《ASVS》（应用安全验证标准）中关于数据泄露防护的建议，前端展示层和后端存储层都要做双重校验。

4. 缓存策略

部门列表、岗位字典这类低频变动的数据，不要每次请求都查库。使用 Redis 缓存，设置合理的 TTL（过期时间），比如 1 小时。当部门信息变更时，主动清除相关 Key。这能将接口响应时间从 50ms 降低到 5ms 以内。