王宇宏实战：5个步骤一文搞懂劳务系统搭建

版本升级后 API 全变了？别慌，老规矩，咱们不整虚的，直接上代码。

做开发这么多年，最怕的就是接手一个老项目，或者自己项目升级框架版本，结果发现连个简单的查询接口都跑不通。特别是涉及到像【王宇宏】这样具体业务场景的系统，底层数据结构一变，上层逻辑全得重写。

今天这篇，我就以“王宇宏”这个具体案例为引子，带大家从零搭建一个典型的劳务班组管理后端服务。别被名字吓到，这其实是一个标准的 RESTful API 开发流程。我们会用 Python 和 FastAPI 框架，因为它的开发效率高，且官方文档对异步支持讲得非常透彻。

咱们的目标很明确：搭建一个能跑、能测、能扩展的最小可行产品（MVP）。重点解决三个痛点：

目录结构混乱：新手写代码往往是一个大文件到底，改一处崩全身。
API 变动无感：缺乏统一的版本管理和错误处理机制。
业务逻辑耦合：数据库操作和业务逻辑混在一起，维护成本极高。

下面咱们一步步来，保证你看完能直接在本地跑通。

项目目标与核心边界

在动手之前，先搞清楚“王宇宏”在这个系统里到底指代什么？在实际的劳务班组管理中，“王宇宏”通常是一个具体的劳务班组负责人或核心技术人员。

我们的系统需要覆盖他的日常职责边界：

人员管理：班组内工人的入职、离职、技能认证状态。
考勤记录：每日打卡数据的录入与汇总。
材料申报：劳务分包材料的提交与审核状态跟踪。

这里有一个关键的业务规则需要硬编码进逻辑： 证书有效期与年审机制。根据行业惯例，特种作业操作证每3年复审一次，安全员证书每2年复审。如果证书过期，系统必须自动标记该人员为“不可上岗”状态，并在API返回中明确提示。

这不是简单的 CRUD，这是带有状态机的业务逻辑。很多新手容易忽略这一点，导致后期数据清洗成本极高。我们要在数据模型设计阶段就把这个状态字段预留好。

目录结构：工程化的第一步

很多博主喜欢直接甩代码，但我强烈建议你先把目录结构搭好。一个清晰的结构，是项目长期可维护的基石。

以下是我们本次实战的目录结构：

wanghai-project/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用入口
│   ├── config.py        # 配置管理
│   ├── database.py      # 数据库连接
│   ├── models/          # 数据模型
│   │   ├── __init__.py
│   │   └── worker.py    # 劳务人员模型
│   ├── schemas/         # Pydantic 校验模型
│   │   ├── __init__.py
│   │   └── worker.py
│   ├── services/        # 业务逻辑层
│   │   ├── __init__.py
│   │   └── worker_service.py
│   └── routers/         # API 路由
│       ├── __init__.py
│       └── workers.py
├── tests/
│   ├── __init__.py
│   └── test_workers.py
├── requirements.txt
└── README.md

为什么要这样分？

Models vs Schemas：models 是 SQLAlchemy 的 ORM 模型，对应数据库表结构；schemas 是 Pydantic 模型，用于数据校验和序列化。两者分离，避免数据库结构变动直接污染 API 契约。
Services 层：这是核心。把业务逻辑（比如判断证书是否过期）从 Router 中剥离出来。Router 只负责接收请求和返回响应，Service 负责处理逻辑。这样，如果以后你要把 API 改成 GraphQL，或者加一个命令行工具调用同一套逻辑，你只需要复用 Service 层即可。
Config 独立：环境变量、数据库 URL、密钥等敏感信息，绝不硬编码在代码里。

核心代码实现：逐行拆解

接下来是重头戏。我们将实现“查询王宇宏所在班组人员列表，并自动过滤证书过期人员”的功能。

1. 数据模型定义 (models/worker.py)

from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime
from app.database import Baseclass Worker(Base):__tablename__ = "workers"id = Column(Integer, primary_key=True, index=True)name = Column(String(50), nullable=False, index=True)  # 姓名，如王宇宏role = Column(String(20), nullable=False)  # 角色：负责人/技术工/普工cert_type = Column(String(50))  # 证书类型cert_expiry_date = Column(DateTime)  # 证书有效期is_active = Column(Boolean, default=True)  # 是否在岗created_at = Column(DateTime, default=datetime.utcnow)# 关系映射，后续扩展班组属性用# group_id = Column(Integer, ForeignKey("groups.id"))# group = relationship("Group")def is_cert_valid(self):"""核心业务逻辑：判断证书是否有效注意：这里不能只判断 is_active，必须结合时间"""if not self.cert_expiry_date:return Falsereturn self.cert_expiry_date >= datetime.utcnow()

关键点讲解：

is_cert_valid 方法直接定义在 Model 上。虽然有些架构派反对在 Model 里写业务逻辑，但对于这种简单的状态判断，放在 Model 里最方便，且符合 DRY 原则。
datetime.utcnow 用于获取当前 UTC 时间。务必统一时区处理，否则在跨时区部署时会出现“早上正常，晚上报错”的灵异现象。

2. Schema 定义 (schemas/worker.py)

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional, Listclass WorkerBase(BaseModel):name: str = Field(..., max_length=50)role: strcert_type: Optional[str] = Nonecert_expiry_date: Optional[datetime] = Noneclass WorkerCreate(WorkerBase):passclass WorkerResponse(WorkerBase):id: intis_active: boolcert_status: str  # 新增字段：证书状态（有效/过期/无）class Config:from_attributes = True  # 允许从 ORM 模型直接转换

注意： cert_status 是一个计算字段，它不在数据库里，而是在序列化时动态生成的。这要求我们在 Service 层处理好这个逻辑，而不是让前端去算。

3. Service 层逻辑 (services/worker_service.py)

from sqlalchemy.orm import Session
from app.models.worker import Worker
from app.schemas.worker import WorkerResponse
from datetime import datetime
from typing import Listclass WorkerService:def __init__(self, db: Session):self.db = dbdef get_worker_list(self, filter_expired: bool = True) -> List[WorkerResponse]:"""获取人员列表:param filter_expired: 是否过滤掉证书过期的人"""query = self.db.query(Worker)# 基础过滤：只查在岗人员query = query.filter(Worker.is_active == True)# 如果需要过滤证书过期的if filter_expired:# 这里使用 Python 的 filter 在内存中过滤，或者使用 SQL 的 func.now()# 为了演示简洁，先查出所有，再过滤pass workers = query.all()results = []for w in workers:# 构建响应对象resp = WorkerResponse(id=w.id,name=w.name,role=w.role,cert_type=w.cert_type,cert_expiry_date=w.cert_expiry_date,is_active=w.is_active,cert_status="valid" if w.is_cert_valid() else "expired")# 二次过滤：如果要求过滤过期，且当前过期，则跳过if filter_expired and resp.cert_status == "expired":continueresults.append(resp)return results

避坑指南：

N+1 问题：上面的代码在数据量小的时候没问题。如果 Worker 表有 10 万条记录，且每条记录都需要查询关联的 Group 表，这样写会发起 10 万次 SQL 查询，直接拖垮数据库。
解决方案：在 query.all() 之前，使用 joinedload 或 subqueryload 进行预加载。在本例中，因为只是简单字段，暂时没体现，但你在实战中必须警惕。

4. 路由与 API 端点 (routers/workers.py)

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import get_db
from app.services.worker_service import WorkerService
from app.schemas.worker import WorkerResponse
from typing import Listrouter = APIRouter(prefix="/api/v1/workers", tags=["workers"])@router.get("/", response_model=List[WorkerResponse])
def list_workers(filter_expired: bool = True,db: Session = Depends(get_db)
):"""获取劳务班组人员列表示例：GET /api/v1/workers?filter_expired=true"""service = WorkerService(db)# 业务校验：如果数据库连接失败，这里会抛异常try:return service.get_worker_list(filter_expired=filter_expired)except Exception as e:# 生产环境建议记录日志，而不是直接返回原始错误raise HTTPException(status_code=500, detail="Failed to fetch workers")

版本控制的重要性： 注意 URL 中的 /api/v1/。这就是解决“版本升级后 API 全变了”痛点的核心手段之一。当未来业务逻辑变更，比如证书年审规则从 3 年改为 2 年，或者需要返回新的字段 penalty_status 时，你可以新增 /api/v2/workers 路由，而不影响旧版 /api/v1 的客户端。

运行与测试：确保稳定性

代码写完不测试，等于没写。我们使用 pytest 和 httpx 进行接口测试。

1. 初始化数据库 (database.py)

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker# 使用 SQLite 便于本地测试，生产环境请换 PostgreSQL
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()def get_db():db = SessionLocal()try:yield dbfinally:db.close()

2. 编写测试用例 (tests/test_workers.py)

import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.database import engine, Base
from app.models.worker import Worker
from datetime import datetime, timedelta# 每次测试前重建表
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)client = TestClient(app)def test_list_workers_with_expired_filter():# 模拟数据# 1. 王宇宏，证书有效w1 = Worker(name="王宇宏", role="负责人", cert_expiry_date=datetime.utcnow() + timedelta(days=100), is_active=True)# 2. 李四，证书过期w2 = Worker(name="李四", role="普工", cert_expiry_date=datetime.utcnow() - timedelta(days=10), is_active=True)# 插入数据库from app.database import SessionLocaldb = SessionLocal()db.add(w1)db.add(w2)db.commit()db.close()# 测试过滤过期的情况response = client.get("/api/v1/workers?filter_expired=true")assert response.status_code == 200data = response.json()assert len(data) == 1assert data[0]["name"] == "王宇宏"assert data[0]["cert_status"] == "valid"# 测试不过滤的情况response_all = client.get("/api/v1/workers?filter_expired=false")data_all = response_all.json()assert len(data_all) == 2

运行命令：

pip install -r requirements.txt
pytest tests/ -v

如果测试通过，说明你的核心逻辑是健壮的。这时候，你就可以放心地启动服务了：

uvicorn app.main:app --reload

访问 http://127.0.0.1:8000/docs，你会看到 Swagger UI 自动生成的接口文档。这就是 FastAPI 的强大之处，文档即代码。

优化扩展与避坑指南

项目能跑了，但离生产环境还有距离。这里有几个进阶技巧，能让你少走三年弯路。

1. 依赖注入与配置管理

不要硬编码数据库连接。使用 pydantic-settings 加载 .env 文件。

# config.py
from pydantic_settings import BaseSettingsclass Settings(BaseSettings):DATABASE_URL: strAPI_V1_STR: str = "/api/v1"class Config:env_file = ".env"settings = Settings()

这样，开发环境用 SQLite，测试环境用 PostgreSQL，生产环境用 MySQL，只需要改 .env 文件，代码零修改。

2. 异常处理统一化

目前我们的 HTTPException 是散落在各个 Router 里的。建议创建一个全局异常处理器。

# main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponseapp = FastAPI()@app.exception_handler(Exception)
async def custom_exception_handler(request: Request, exc: Exception):return JSONResponse(status_code=500,content={"detail": "Internal Server Error", "error_code": "GENERIC_500"})

这样，无论后端哪里报错，前端收到的 JSON 结构都是统一的，方便前端统一做 Toast 提示。

3. 日志记录

在 WorkerService 中，当检测到证书过期时，打印一条 WARNING 级别的日志。

import logging
logger = logging.getLogger(__name__)# 在 service 中
if w.cert_status == "expired":logger.warning(f"Worker {w.name} certificate expired on {w.cert_expiry_date}")

日志是排查线上问题的唯一线索。没有日志的后端，等于黑盒。

4. 性能优化：索引与缓存

数据库索引：我们在 Worker 模型中给 name 和 cert_expiry_date 加了索引。对于高频查询字段，索引是必须的。
Redis 缓存：如果“查询班组人员”接口被高频调用（比如前端每 5 秒轮询一次），可以考虑将结果缓存到 Redis，设置 30 秒过期时间。但要注意，缓存失效时的并发击穿问题，需要加锁或互斥。