3天手写实现报修系统，告别教程依赖症 看了一堆教程还是不会写项目？这是无数初学者的痛点。别慌，今天咱们不玩虚的，直接上手手写实现一个实用的报修系统。 很多新人卡在“看了很多，动手就废”的瓶颈期。原因很简单：教程往往只讲局部，没讲全链路。一个完整的报修系统，涉及用户登录、工单创建、状态流转、后台管理，缺了任何一环都跑不通。 咱们抛开那些花里胡哨的前端特效，聚焦核心业务逻辑。用 Python + FastAPI + SQLite 这套轻量级组合，手写实现从 0 到 1 的完整流程。代码不多，但每一步都踩在实处，看完就能跑通。 项目目标与核心逻辑 在敲第一行代码前，先理清业务边界。一个最小可行产品（MVP）的报修系统，必须包含三个角色：普通用户、维修师傅、系统管理员。 核心流程闭环如下：用户端：提交报修申请，包含故障描述、位置、紧急程度。 调度端：管理员或系统自动分配任务给空闲师傅。 执行端：师傅接单、到达现场、完成维修、上传凭证。 反馈端：用户评价，工单归档。很多新手容易陷入“过度设计”，上来就搞微服务、消息队列。对于练手项目，手写实现单体应用才是正道。数据一致性比架构炫酷更重要。 我们定义四个核心实体：User：包含用户、师傅、管理员三种角色。 Order：报修工单，核心状态机载体。 ServiceItem：维修项目，用于计费参考。 Review：用户评价，闭环最后一环。状态流转是报修系统的灵魂。工单状态必须严格受控，禁止跳跃式变更。比如，不能从“待分配”直接跳到“已完成”，中间必须经过“已接单”和“处理中”。这种状态机的严谨性，在开发者文档中通常被强调为业务一致性的基石，但在实际编码中，往往被新人忽略。 目录结构规划 清晰的目录结构是工程化的第一步。别把代码全堆在 main.py 里，那是脚本，不是项目。 推荐采用标准的 FastAPI 项目结构： repair-system/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口 │ ├── config.py # 配置管理 │ ├── database.py # 数据库连接 │ ├── models/ # SQLAlchemy 模型 │ │ ├── __init__.py │ │ ├── user.py │ │ └── order.py │ ├── schemas/ # Pydantic 数据校验 │ │ ├── __init__.py │ │ ├── user.py │ │ └── order.py │ ├── api/ # 路由层 │ │ ├── __init__.py │ │ ├── deps.py # 依赖注入 │ │ └── v1/ │ │ ├── auth.py │ │ └── orders.py │ └── services/ # 业务逻辑层 │ ├── __init__.py │ └── order_service.py ├── tests/ # 单元测试 │ ├── __init__.py │ └── test_orders.py ├── requirements.txt └── README.md关键点解析：分层架构：API 层只负责参数接收和响应返回，业务逻辑下沉到 services 层。这样以后改数据库或加逻辑，不用动路由代码。 Schema 与 Model 分离：models 是数据库结构，schemas 是接口数据格式。两者解耦，防止数据库字段直接暴露给前端，也是安全规范的基本要求。 依赖注入：deps.py 存放获取当前用户、数据库会话等通用逻辑，避免在每个接口里重复写 get_db()。这种结构虽然初期搭建稍显繁琐，但手写实现过程中，你会深刻体会到模块化带来的可维护性提升。 核心代码实现 接下来进入硬核环节。我们将分模块展示关键代码，并逐行讲解坑点。 1. 数据库模型定义 使用 SQLAlchemy 2.0 风格，确保类型安全。 # app/models/order.py from sqlalchemy import Column, Integer, String, Float, DateTime, Enum, ForeignKey from sqlalchemy.orm import relationship from datetime import datetime from app.database import Base# 定义工单状态枚举，严禁使用魔法字符串 class OrderStatus(str, Enum):PENDING = pending # 待分配ASSIGNED = assigned # 已分配IN_PROGRESS = in_progress # 处理中COMPLETED = completed # 已完成CANCELLED = cancelled # 已取消class Order(Base):__tablename__ = ordersid = Column(Integer, primary_key=True, index=True)title = Column(String(100), nullable=False)description = Column(String(500))location = Column(String(200), nullable=False)status = Column(Enum(OrderStatus), default=OrderStatus.PENDING)priority = Column(Integer, default=1) # 1普通 2紧急created_at = Column(DateTime, default=datetime.utcnow)updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)# 关联关系user_id = Column(Integer, ForeignKey(users.id))user = relationship(User, back_populates=orders)technician_id = Column(Integer, ForeignKey(users.id))technician = relationship(User, backref=assigned_orders)def to_dict(self):转换为字典，便于 JSON 序列化return {id: self.id,title: self.title,status: self.status.value,location: self.location,created_at: self.created_at.isoformat(),technician_name: self.technician.full_name if self.technician else None}避坑指南：Enum 类型：很多新手喜欢用字符串 pending 表示状态。一旦拼错，系统就崩了。使用 Python 的 Enum 类，可以在 IDE 中自动补全，也能在数据库层面约束值域。 to_dict 方法：Pydantic 的 model_dump 虽然好用，但处理嵌套关系（如 technician）时容易报错。手动定义序列化方法，能更精细地控制输出字段，避免循环引用。2. 业务逻辑层：状态机控制 这是手写实现报修系统最核心的部分。所有状态变更必须经过服务层校验。 # app/services/order_service.py from fastapi import HTTPException, status from app.models.order import Order, OrderStatusclass OrderService:@staticmethoddef validate_status_transition(current_status: OrderStatus, new_status: OrderStatus):校验状态流转合法性这是防止业务逻辑漏洞的关键屏障allowed_transitions = {OrderStatus.PENDING: [OrderStatus.ASSIGNED, OrderStatus.CANCELLED],OrderStatus.ASSIGNED: [OrderStatus.IN_PROGRESS, OrderStatus.CANCELLED],OrderStatus.IN_PROGRESS: [OrderStatus.COMPLETED],OrderStatus.COMPLETED: [], # 终态，不可变OrderStatus.CANCELLED: [] # 终态，不可变}if new_status not in allowed_transitions.get(current_status, []):raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail=f非法状态流转: {current_status.value} - {new_status.value})@staticmethoddef update_order_status(db, order_id: int, new_status: OrderStatus, technician_id: int = None):order = db.query(Order).filter(Order.id == order_id).first()if not order:raise HTTPException(status_code=404, detail=工单不存在)# 1. 校验状态流转OrderService.validate_status_transition(order.status, new_status)# 2. 更新数据order.status = new_statusif technician_id:order.technician_id = technician_iddb.commit()db.refresh(order)return order为什么要在服务层做校验？ 如果在 API 层直接修改数据库，攻击者可以绕过前端限制，直接发送 POST 请求把“待分配”的工单改为“已完成”，导致财务损失。服务层的状态机校验，是开发者文档中推荐的最佳实践，它将业务规则与入口解耦，确保无论请求来自哪里，逻辑都是统一的。 3. API 路由层 保持接口轻薄，只做参数解析和权限检查。 # app/api/v1/orders.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.database import get_db from app.api.deps import get_current_user from app.schemas.order import OrderCreate, OrderStatusUpdate from app.services.order_service import OrderServicerouter = APIRouter()@router.post(/orders, response_model=dict, status_code=status.HTTP_201_CREATED) def create_order(order_in: OrderCreate, db: Session = Depends(get_db), current_user=Depends(get_current_user)):# 1. 创建工单对象new_order = Order(title=order_in.title,description=order_in.description,location=order_in.location,priority=order_in.priority,user_id=current_user.id)# 2. 入库db.add(new_order)db.commit()db.refresh(new_order)return new_order.to_dict()@router.patch(/orders/{order_id}/status) def update_status(order_id: int, status_in: OrderStatusUpdate, db: Session = Depends(get_db), current_user=Depends(get_current_user)):# 权限检查：只有管理员或指定师傅能改状态if current_user.role != admin and current_user.role != technician:raise HTTPException(status_code=403, detail=权限不足)# 调用服务层处理核心逻辑updated_order = OrderService.update_order_status(db=db,order_id=order_id,new_status=status_in.status,technician_id=current_user.id if current_user.role == technician else None)return updated_order.to_dict()运行与测试 代码写完，别急着觉得大功告成。没有测试的代码等于没有写。 1. 环境配置与启动 # 安装依赖 pip install fastapi uvicorn sqlalchemy pydantic python-jose passlib bcrypt# 启动服务 uvicorn app.main:app --reload访问 http://127.0.01:8000/docs，你会看到 Swagger UI 界面。这是 FastAPI 的默认功能，极大降低了前后端联调成本。 2. 编写单元测试 使用 pytest + httpx 进行接口测试。重点测试状态流转的边界情况。 # tests/test_orders.py import pytest from fastapi.testclient import TestClient from app.main import app from app.database import Base, engine# 每次测试前重建表，确保数据隔离 Base.metadata.drop_all(bind=engine) Base.metadata.create_all(bind=engine)client = TestClient(app)def test_order_lifecycle():# 1. 登录获取 Token (假设已有登录接口)# ... 省略登录逻辑，直接构造一个模拟用户# 2. 创建工单response = client.post(/api/v1/orders, json={title: 空调漏水,description: 客厅空调滴水,location: A栋101,priority: 1}, headers={Authorization: fBearer {token}})assert response.status_code == 201order_id = response.json()[id]# 3. 尝试非法流转：直接从 pending 跳到 completedinvalid_response = client.patch(f/api/v1/orders/{order_id}/status, json={status: completed}, headers={Authorization: fBearer {admin_token}})assert invalid_response.status_code == 400assert 非法状态流转 in invalid_response.json()[detail]# 4. 合法流转：pending - assignedvalid_response = client.patch(f/api/v1/orders/{order_id}/status, json={status: assigned}, headers={Authorization: fBearer {admin_token}})assert valid_response.status_code == 200测试价值： 这段测试代码揭示了手写实现中极易出现的 Bug：状态校验缺失。如果没有服务层的 validate_status_transition，第 3 步的非法请求会直接成功，导致数据混乱。 优化扩展方向 基础功能跑通后，如何让它更像生产级系统？并发控制： 在高并发场景下，两个师傅同时抢单可能导致重复分配。解决方案是使用数据库的 SELECT ... FOR UPDATE 行级锁，或者引入 Redis 分布式锁。 异步任务： 发送短信通知、生成 PDF 账单等操作，不要阻塞主线程。引入 Celery 或 Arq 进行异步处理。 数据权限： 师傅只能看分配给自己的工单，用户只能看自己的工单。在查询时动态拼接 WHERE user_id = :current_user_id，这是安全审计的重点。 日志与监控： 接入 Sentry 或 Prometheus。报修系统涉及线下服务，任何异常都需要快速定位。记录每个状态变更的操作人、时间、IP，形成审计日志。小结 通过手写实现这个报修系统，你不仅完成了一个 Demo，更掌握了后端开发的通用范式：分层架构：API、Service、Model 职责分离。 状态机设计：用枚举和校验逻辑保证业务一致性。 防御性编程：在入口层和逻辑层双重校验，防止非法操作。 测试驱动：用单元测试覆盖边界情况，提升代码信心。很多新人觉得项目难，是因为想一口吃成胖子。其实，把复杂系统拆解成一个个小的、可验证的模块，逐个击破，难度就降下来了。 你在项目里踩过这个坑吗？比如状态流转混乱，或者权限校验漏掉？评论区聊聊，咱们一起复盘。