3步搞定加班黑名单实战项目，告别环境配置卡半天
配置环境就卡半天，代码跑不通，报错满屏飞，这是无数开发者在接手新任务时的噩梦。尤其是当你为了赶一个实战项目，盯着“加班黑名单”这个看似简单的功能模块，却在依赖安装、权限配置上耗费了整整三天。
很多新人觉得这只是个简单的增删改查（CRUD）功能，但真正落地时，你会发现“加班黑名单”不仅仅是数据的存储，更涉及复杂的权限隔离、状态同步以及高并发下的数据一致性。这篇文章不玩虚的，直接带你从零搭建一个可复现的“加班黑名单”管理系统。我们会用 Python 和 FastAPI 框架，结合 SQLite 数据库（生产环境建议换 PostgreSQL），完整走通从目录结构到核心代码实现，再到运行测试的全过程。
项目目标
在动手写代码前，我们必须明确这个“加班黑名单”实战项目到底要解决什么问题。
在传统的考勤或工时管理中，“黑名单”通常指那些因为频繁加班、工时异常或违反合规要求而被系统标记的员工列表。这个模块的核心目标有三个：
- 数据持久化：安全地存储员工 ID、被加入黑名单的原因、操作人、时间戳。
- 状态隔离：确保只有 HR 或特定权限的管理员能查看和修改黑名单，普通员工只能查询自己的状态。
- 审计追踪：每一次添加或删除操作，都必须留下不可篡改的日志，以便后续审计。
很多初学者容易犯的错误是，把“黑名单”做成一个简单的布尔值字段（如 is_blacklisted = True）。这在实战项目中是大忌。因为业务需求会变，今天你可能只需要一个标记，明天就需要记录“为什么被拉黑”，后天还需要支持“申诉后自动移除”。因此，我们需要一个独立的表结构来承载这些逻辑，而不是在用户表中加个字段了事。
目录结构
工欲善其事，必先利其器。一个清晰的目录结构能帮你避免后期代码乱成一团。我们采用标准的 FastAPI 项目结构，但针对“加班黑名单”模块做了专门的分层。
overtime_blacklist/
├── app/
│ ├── __init__.py
│ ├── main.py # 应用入口，挂载路由
│ ├── core/
│ │ ├── config.py # 配置管理，读取环境变量
│ │ └── security.py # 权限校验逻辑
│ ├── db/
│ │ ├── base.py # 数据库引擎和会话管理
│ │ └── models.py # SQLAlchemy ORM 模型定义
│ ├── schemas/
│ │ └── blacklist.py # Pydantic 数据验证模式
│ └── api/
│ └── v1/
│ ├── endpoints/
│ │ └── blacklist.py # 核心 API 路由
│ └── deps.py # 依赖注入，获取当前用户等
├── tests/
│ └── test_blacklist.py # 单元测试
├── requirements.txt
└── README.md
关键点解析：
db/models.py：这里定义我们的数据表。千万不要把业务逻辑写在模型里，模型只负责映射数据库结构。schemas/blacklist.py：FastAPI 的核心优势在于数据验证。所有的输入输出数据都要通过 Pydantic 模型进行严格校验，防止脏数据进入数据库。api/v1/endpoints/：路由文件。我们将所有与“加班黑名单”相关的接口都放在这里，保持高内聚低耦合。
这种结构在 Stack Overflow 上被许多高票答案推荐，因为它在小型实战项目中既足够灵活，又不会像企业级微服务架构那样过于臃肿。
核心代码实现
接下来是重头戏，我们将逐步实现核心代码。
1. 定义数据模型 (ORM)
首先，我们在 app/db/models.py 中定义 BlacklistRecord 模型。注意，我们使用 SQLAlchemy 2.0 风格。
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text
from sqlalchemy.orm import relationship
from app.db.base import Base
from datetime import datetimeclass BlacklistRecord(Base):__tablename__ = "blacklist_records"id = Column(Integer, primary_key=True, index=True)employee_id = Column(Integer, ForeignKey("employees.id"), index=True, nullable=False)reason = Column(Text, nullable=False) # 拉黑原因created_by = Column(Integer, ForeignKey("users.id"), nullable=False) # 操作人created_at = Column(DateTime, default=datetime.utcnow)is_active = Column(Integer, default=1) # 1表示生效，0表示已移除# 关系映射，方便查询时直接关联员工信息employee = relationship("Employee", back_populates="blacklist_history")operator = relationship("User", back_populates="blacklist_actions")
逐行讲解：
index=True：在employee_id上建立索引。因为我们会频繁查询某个员工是否在黑名单中，没有索引会导致全表扫描，数据量大时性能会急剧下降。is_active：使用整型而非布尔型，是为了方便后续做历史数据归档。如果我们直接删除记录，就失去了审计价值；将is_active置为 0，相当于软删除。
2. 定义 Pydantic Schema
在 app/schemas/blacklist.py 中，我们定义输入和输出的数据结构。
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optionalclass BlacklistCreate(BaseModel):employee_id: int = Field(..., gt=0, description="员工ID必须大于0")reason: str = Field(..., min_length=5, max_length=255, description="原因至少5个字")class BlacklistResponse(BaseModel):id: intemployee_id: intreason: strcreated_at: datetimeis_active: intclass Config:from_attributes = True
避坑指南：
很多新手在 Pydantic 中忘记设置 from_attributes = True（旧版本是 orm_mode = True），导致在将 SQLAlchemy 对象转换为 JSON 时报错 AttributeError。这是 FastAPI 开发中最常见的报错之一，务必检查。
3. 实现核心 API 接口
在 app/api/v1/endpoints/blacklist.py 中，我们实现添加和查询黑名单的接口。
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import Listfrom app.db.base import get_db
from app.db.models import BlacklistRecord, Employee
from app.schemas.blacklist import BlacklistCreate, BlacklistResponse
from app.api.v1.deps import get_current_active_user, get_current_hrrouter = APIRouter()@router.post("/", response_model=BlacklistResponse)
def add_to_blacklist(record_in: BlacklistCreate,db: Session = Depends(get_db),current_user = Depends(get_current_hr) # 依赖注入，确保只有HR能调用
):# 1. 检查员工是否存在employee = db.query(Employee).filter(Employee.id == record_in.employee_id).first()if not employee:raise HTTPException(status_code=404, detail="员工不存在")# 2. 检查是否已经在黑名单中（防止重复添加）existing_record = db.query(BlacklistRecord).filter(BlacklistRecord.employee_id == record_in.employee_id,BlacklistRecord.is_active == 1).first()if existing_record:raise HTTPException(status_code=400, detail="该员工已在黑名单中")# 3. 创建新记录db_record = BlacklistRecord(employee_id=record_in.employee_id,reason=record_in.reason,created_by=current_user.id,is_active=1)db.add(db_record)db.commit()db.refresh(db_record)return db_record@router.get("/check/{employee_id}", response_model=bool)
def check_blacklist_status(employee_id: int,db: Session = Depends(get_db)
):# 查询是否存在生效的黑名单记录result = db.query(BlacklistRecord).filter(BlacklistRecord.employee_id == employee_id,BlacklistRecord.is_active == 1).first()return result is not None
代码细节拆解：
- 依赖注入
get_current_hr：这是安全性的核心。我们假设get_current_hr是一个装饰器或函数，它验证当前请求的用户是否具有 HR 角色。如果权限不足，FastAPI 会自动返回 403 Forbidden。 - 事务处理：
db.commit()必须放在所有操作成功后。如果中间抛出异常，FastAPI 的异常处理器会捕获它，事务不会提交，保证了数据的一致性。 - 幂等性考虑：我们在添加前检查了
existing_record，这保证了接口的幂等性。即使前端因网络抖动重复发送请求，也不会产生重复数据。
运行与测试
代码写完只是第一步，能跑起来并且符合预期才是关键。
1. 初始化数据库
运行项目前，确保数据库表已创建。在 app/db/base.py 中，我们通常会有这样的初始化逻辑：
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settingsSQLALCHEMY_DATABASE_URL = settings.DATABASE_URL
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()def init_db():Base.metadata.create_all(bind=engine)
在 main.py 中启动时调用 init_db()。
2. 使用 HTTPie 或 Postman 测试
假设项目已运行在 http://localhost:8000。
测试添加黑名单：
http POST :8000/api/v1/blacklist/ \Authorization:Bearer <YOUR_TOKEN> \employee_id:=101 \reason:="连续3个月加班超过200小时，违反劳动法"
预期返回：
{"id": 1,"employee_id": 101,"reason": "连续3个月加班超过200小时，违反劳动法","created_at": "2023-10-27T10:00:00","is_active": 1
}
测试查询状态：
http GET :8000/api/v1/blacklist/check/101
预期返回：
true
常见报错排查：
如果在测试中遇到 401 Unauthorized，请检查你的 Token 是否过期或请求头格式是否正确。Stack Overflow 上有大量关于 FastAPI 认证失败的讨论，90% 的问题都出在 Authorization 头的拼写上（必须是 Bearer 加空格，而不是 bearer）。
3. 单元测试
在 tests/test_blacklist.py 中，我们可以使用 pytest 和 httpx 进行简单的集成测试。
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.db.base import init_db, engine# 测试前重置数据库
init_db()
client = TestClient(app)def test_add_blacklist_success():# 模拟登录获取 Token 的逻辑省略headers = {"Authorization": "Bearer test_token"}response = client.post("/api/v1/blacklist/",headers=headers,json={"employee_id": 999, "reason": "测试原因"})assert response.status_code == 200assert response.json()["employee_id"] == 999
优化扩展
基础功能跑通后，我们需要考虑生产环境的挑战。
1. 性能优化：添加缓存
对于“查询员工是否在黑名单”这种高频读、低频写的操作，我们可以引入 Redis 缓存。
import redis
from app.core.config import settingsr = redis.Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT)def check_blacklist_with_cache(employee_id: int, db: Session):cache_key = f"blacklist:employee:{employee_id}"# 先查缓存cached_result = r.get(cache_key)if cached_result is not None:return cached_result == b"1"# 缓存未命中，查数据库result = db.query(BlacklistRecord).filter(BlacklistRecord.employee_id == employee_id,BlacklistRecord.is_active == 1).first()is_blacklisted = result is not None# 写入缓存，设置过期时间 5 分钟r.setex(cache_key, 300, "1" if is_blacklisted else "0")return is_blacklisted
注意：当黑名单状态发生变化时（添加或删除），必须主动删除对应的 Redis 缓存，否则会导致数据不一致。
2. 扩展：支持申诉流程
在实际业务中，员工可能会对“被拉黑”提出申诉。我们可以扩展模型，增加一个 appeal_status 字段，或者创建一个独立的 Appeal 表，关联 BlacklistRecord。
当申诉成立时，触发一个事件，将 is_active 置为 0，并记录申诉通过的时间。这种设计使得系统具备了可扩展性，未来如果增加“自动解除黑名单”规则，只需修改业务逻辑，无需改动核心数据结构。
3. 日志审计
除了数据库记录，我们还应该记录应用日志。使用 Python 的 logging 模块，在关键操作点打印日志。
import logging
logger = logging.getLogger(__name__)# 在添加黑名单成功后
logger.info(f"Employee {record_in.employee_id} added to blacklist by user {current_user.id}. Reason: {record_in.reason}")
这些日志对于排查线上问题和合规审计至关重要。
小结
搭建这个“加班黑名单”实战项目，我们不仅实现了基本的 CRUD 功能，更重要的是掌握了以下几个核心工程化思维：
- 数据建模的灵活性：通过独立的记录表而非简单字段，支持了复杂的业务场景（如审计、历史追溯）。
- 安全性设计：通过依赖注入和权限校验，确保了接口的访问控制。
- 性能考量：通过索引优化和缓存策略，提升了系统的响应速度。
- 可测试性：清晰的目录结构和模块化代码，使得单元测试和集成测试变得容易。
这个知识点你面试被问过吗？特别是关于“如何设计一个支持审计追踪的状态变更系统”或者“FastAPI 中如何处理高并发下的数据一致性问题”。留言说说你的答案，或者你遇到过什么坑，我们一起交流。
