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__ = "orders"id = 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": f"Bearer {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": f"Bearer {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": f"Bearer {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，形成审计日志。