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, OrderStatus
class OrderService:
@staticmethod
def 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}
)
@staticmethod
def 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_status
if technician_id:
order.technician_id = technician_id
db.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 OrderService
router = 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 == 201
order_id = response.json()[id]
# 3. 尝试非法流转：直接从 pending 跳到 completed
invalid_response = client.patch(f/api/v1/orders/{order_id}/status, json={
status: completed
}, headers={Authorization: fBearer {admin_token}})
assert invalid_response.status_code == 400
assert 非法状态流转 in invalid_response.json()[detail]
# 4. 合法流转：pending - assigned
valid_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 职责分离。
状态机设计：用枚举和校验逻辑保证业务一致性。
防御性编程：在入口层和逻辑层双重校验，防止非法操作。
测试驱动：用单元测试覆盖边界情况，提升代码信心。
很多新人觉得项目难，是因为想一口吃成胖子。其实，把复杂系统拆解成一个个小的、可验证的模块，逐个击破，难度就降下来了。
你在项目里踩过这个坑吗？比如状态流转混乱，或者权限校验漏掉？评论区聊聊，咱们一起复盘。 
长河君 · 2026/9/22 8:06:01 · 阅读 1.2k 3天手写实现报修系统，告别教程依赖症
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, OrderStatus
class OrderService:
@staticmethod
def 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}
)
@staticmethod
def 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_status
if technician_id:
order.technician_id = technician_id
db.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 OrderService
router = 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 == 201
order_id = response.json()[id]
# 3. 尝试非法流转：直接从 pending 跳到 completed
invalid_response = client.patch(f/api/v1/orders/{order_id}/status, json={
status: completed
}, headers={Authorization: fBearer {admin_token}})
assert invalid_response.status_code == 400
assert 非法状态流转 in invalid_response.json()[detail]
# 4. 合法流转：pending - assigned
valid_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 职责分离。
状态机设计：用枚举和校验逻辑保证业务一致性。
防御性编程：在入口层和逻辑层双重校验，防止非法操作。
测试驱动：用单元测试覆盖边界情况，提升代码信心。
很多新人觉得项目难，是因为想一口吃成胖子。其实，把复杂系统拆解成一个个小的、可验证的模块，逐个击破，难度就降下来了。
你在项目里踩过这个坑吗？比如状态流转混乱，或者权限校验漏掉？评论区聊聊，咱们一起复盘。