十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

3步搞定加班黑名单实战项目,告别环境配置卡半天

3步搞定加班黑名单实战项目,告别环境配置卡半天 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_recordsid = 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 = fblacklist:employee:{employee_id}# 先查缓存cached_result = r.get(cache_key)if cached_result is not None:return cached_result == b1# 缓存未命中,查数据库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(fEmployee {record_in.employee_id} added to blacklist by user {current_user.id}. Reason: {record_in.reason})这些日志对于排查线上问题和合规审计至关重要。 小结 搭建这个“加班黑名单”实战项目,我们不仅实现了基本的 CRUD 功能,更重要的是掌握了以下几个核心工程化思维:数据建模的灵活性:通过独立的记录表而非简单字段,支持了复杂的业务场景(如审计、历史追溯)。 安全性设计:通过依赖注入和权限校验,确保了接口的访问控制。 性能考量:通过索引优化和缓存策略,提升了系统的响应速度。 可测试性:清晰的目录结构和模块化代码,使得单元测试和集成测试变得容易。这个知识点你面试被问过吗?特别是关于“如何设计一个支持审计追踪的状态变更系统”或者“FastAPI 中如何处理高并发下的数据一致性问题”。留言说说你的答案,或者你遇到过什么坑,我们一起交流。
返回列表