近期有不少做技术服务、任务代办类业务的朋友在问同一个问题一个订单动不动就横跨好几个版本阶段客户问“做到哪一步了”的时候只能翻聊天记录手数订单越多越乱。这篇文章就围绕这个真实痛点展开把「任务型订单管理」从数据建模、状态机设计到接口和看板页面完整实现一遍。适合有 Python 基础、想动手做一个内部管理系统的开发者也适合正在用表格人工跟单、准备把流程工具化的同学。读完之后你能掌握三件事怎么把一张“看着简单”的订单拆成可计算的数据结构、怎么用状态机约束业务流转防止脏数据、怎么用 Flask SQLite 在半小时内跑起一套能用的订单看板。文中所有代码都可以直接复制运行遇到报错也能在排查章节里找到对应解法。1. 背景与核心概念1.1 订单型业务真正缺的不是表格而是结构化先说清楚这篇文章要解决的问题。类似“某版本全包已完成”“本单已交付感谢信任”这样的通知本质上是订单生命周期走到了终点。麻烦的地方在于这类订单往往不是一次点击就结束的动作而是由若干阶段性目标串起来的前期确认需求、中期分批交付、后期验收归档。如果订单信息只存在于聊天记录和脑子里就会出现三个典型问题进度无法量化、责任无法追溯、历史无法复用。用 Excel 跟单在订单量小于十单时确实够用但一旦订单并发上来你会发现自己在做的是“人肉状态同步”。客户问进度你要去看聊天记录同事交接你要口述半小时出了问题想复盘找不到任何操作痕迹。这些问题的共同根源是业务信息没有被结构化状态变化没有被记录下来。所以这篇文章不讲“如何做业务”只讲“如何用代码把业务装进一套可计算的结构里”。具体做法是把一条订单拆成四层对象客户、服务包、订单、任务节点用数据库表把它们固定下来再用一个受约束的状态机控制订单怎么往前走最后用一个看板页面把进度可视化出来。这套结构不依赖任何特定行业做内容交付、设计外包、设备巡检、课程陪跑都能套用。1.2 四个核心概念客户、服务包、订单、任务节点在动手建表之前必须先把概念边界划清楚。很多同学写出来的订单系统最后变成一坨 JSON 字段堆在数据库里就是因为没有把这几层分开。**客户customer**是付费主体也是长期资产。它的关键属性是联系方式与备注一个客户可以有多条订单订单之间可以互相参考。把客户独立成表而不是在订单里存一个字符串名字好处是客户改名时历史订单不用批量改也方便后续做“复购统计”。**服务包package**是对“卖什么”的抽象。它描述的是可复用的产品定义比如“单阶段基础版”“多阶段全包版”包含价格、预估工时、以及一份任务模板。服务包是模板订单是实例这个区分非常重要订单创建时把模板复制一份成任务节点之后修改服务包不会影响已经在跑的订单。**订单order**是一次具体的交易与交付过程它关联一个客户和一个服务包拥有独立的状态、报价、实收金额和时间戳。订单号是对外沟通的唯一凭证必须唯一且可读。**任务节点order_task**是订单内部的执行单元带有序号、标题、权重和状态。进度百分比不是靠人工填的而是从任务节点聚合算出来的。这一点是整套设计的关键后文会专门展开。把这四层的关系画成文字就是一个客户对应多条订单一个服务包对应多条订单一条订单对应多条任务节点。理解了这句后面的表结构就是自然推导出来的结果。1.3 状态机为什么不能用一个字段随便改初学者最常见的写法是给订单一个status字段然后哪个接口想改就改。这种做法在小规模下看不出问题但一旦有了并发或者多人协作就会出现“已完成的订单被改回进行中”“已取消的订单还能继续推进”这类脏数据而且很难查是哪一步出的错。正确做法是把状态流转规则集中定义成一张“允许表”任何一次状态变更都必须先校验当前状态是否允许跳到目标状态。这在软件工程里叫状态机State Machine它的价值不在于概念好听而在于把散落在各个接口里的 if 判断收拢到一处改规则只改一个地方。本文实现的订单状态包含六个PENDING待确认、CONFIRMED已确认、IN_PROGRESS进行中、PENDING_ACCEPT待验收、DONE已完成、CANCELLED已取消。流转关系是有向的待确认只能到已确认或已取消进行中提交验收时系统会额外校验所有任务节点是否都已完成待验收可以退回进行中这是为了应对“客户提出补充修改”的真实场景。终态DONE和CANCELLED不再允许任何流转这样历史就不会被篡改。顺带提一句状态机还有一个容易被忽略的收益它天然生成了一份业务日志。每次流转都往日志表写一条记录谁在什么时候把订单从什么状态改成了什么状态一目了然。后面章节的审计能力就是建立在这个基础之上的。1.4 技术边界本文实现什么、不实现什么为了让文章聚焦这里明确一下范围。本文实现的是一个内部使用的订单管理后端与只读看板包括建表、接口、进度计算、状态校验、操作日志和页面渲染。不实现的部分包括支付对接、多租户、复杂权限体系、消息推送、前端工程化。这些内容每一个都值得单独写一篇硬塞进来只会让代码变得不可运行。另外必须强调一条边界如果业务涉及第三方平台的账号与虚拟财产操作请务必先确认对应产品与平台的用户协议是否允许本文讨论的仅仅是订单数据如何管理不涉及任何账号代操作、自动化脚本、绕过风控或者规避平台规则的技术手段。任何系统的第一优先级都是合规技术方案再好前提不合规也没有意义。2. 环境准备与版本说明2.1 运行环境与技术选型本文选择 Python Flask SQLite 的组合原因很实际零外部服务依赖装完 Python 就能跑非常适合内部工具和中小规模业务。如果你的团队已经在用 Java 或者 Node思路完全一样把数据访问层换掉即可表结构和状态机设计是可以直接复用的。环境要求如下版本请根据你的实际环境调整本文重点演示配置思路而不是锁死版本号操作系统Windows 10/11、macOS、主流 Linux 发行版均可Python3.10 及以上用到了contextlib和 f-string3.10 更稳妥Flask3.x 系列数据库SQLite 3Python 标准库自带无需单独安装开发工具VS Code、PyCharm 或任意编辑器接口调试curl、Postman 或浏览器开发者工具一句话解释选型理由Flask 足够轻路由写起来直观SQLite 单文件存储备份就是复制一个文件对本量级的订单系统完全够用而且它支持事务和外键能让我们把数据一致性讲清楚。如果日后订单量涨到需要多人高并发写入再迁移到 PostgreSQL 或 MySQLSQL 语句的改动量很小。2.2 项目目录结构在开始写代码之前先把目录建好。结构清晰的目录能让你在后面排查问题时少走很多弯路建议严格按下面这个结构组织order-studio/ ├── app.py # Flask 应用与全部接口 ├── db.py # 数据库连接、建表脚本、事务封装 ├── seed.py # 初始化示例数据 ├── requirements.txt # 依赖清单 ├── data/ │ └── studio.db # SQLite 数据文件首次运行自动生成 └── templates/ └── index.html # 订单看板页面这里的data/目录故意和代码分开因为它是唯一需要备份的东西。生产环境里你会希望把数据文件挂到独立磁盘上而不是和代码混在一起这样发布新版本时直接替换代码目录就行不会误删数据。2.3 依赖安装与虚拟环境强烈建议使用虚拟环境避免污染全局 Python 环境。下面三条命令按顺序执行即可# 1. 创建虚拟环境 python -m venv .venv # 2. 激活虚拟环境Windows .venv\Scripts\activate # 激活虚拟环境macOS / Linux source .venv/bin/activate # 3. 安装依赖 pip install -r requirements.txtrequirements.txt的内容如下Flask3.0.3安装完成后执行pip list确认 Flask 已经出现在列表里。如果激活虚拟环境后命令提示符前面没有出现(.venv)说明激活失败通常是在 Windows 上用了 PowerShell 但执行策略有限制可以在 PowerShell 里用.venv\Scripts\Activate.ps1或者干脆切到 CMD 执行activate.bat。这一步看似琐碎但它是后面所有报错排查的地基环境不对后面全是玄学问题。3. 数据模型与表结构设计3.1 客户表与服务包表客户表的设计原则是“宁多勿少”把可能需要检索的字段都留出来。contact字段用来存联系方式不做格式约束因为现实中它可能是手机号、微信号、邮箱甚至一个内部工单链接。note用来存业务备注比如“长期合作结算周期月结”。CREATE TABLE IF NOT EXISTS customer ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, contact TEXT NOT NULL, note TEXT DEFAULT , created_at TEXT NOT NULL );服务包表多了三个关键字段code是业务编码必须唯一方便口头沟通时直接说编码price_cents用整数存“分”这个设计后面会专门解释原因active是上下架开关用软下架代替删除保证历史订单还能正确关联到服务包信息。CREATE TABLE IF NOT EXISTS package ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, price_cents INTEGER NOT NULL DEFAULT 0, active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL );再往下是服务包的任务模板表。这张表是整套设计里最容易被省略、但收益最大的一张它让“创建一个订单”从手工填十个节点变成选一个服务包自动生成。模板节点的seq决定顺序weight决定它在总进度里的权重。CREATE TABLE IF NOT EXISTS package_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, package_id INTEGER NOT NULL, seq INTEGER NOT NULL, title TEXT NOT NULL, weight INTEGER NOT NULL DEFAULT 1, FOREIGN KEY (package_id) REFERENCES package (id) );3.2 订单主表与任务节点表订单主表承载的是“这笔交易”的全部事实谁买的、买的什么、多少钱、当前什么状态、什么时候开始什么时候结束。注意quoted_cents报价和paid_cents实收是两个字段这是从实践中得出的经验因为报价和实收不一致是常态有优惠、有追加、有尾款合成一个字段最后一定会出问题。CREATE TABLE IF NOT EXISTS order_main ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT NOT NULL UNIQUE, customer_id INTEGER NOT NULL, package_id INTEGER NOT NULL, status TEXT NOT NULL DEFAULT PENDING, quoted_cents INTEGER NOT NULL DEFAULT 0, paid_cents INTEGER NOT NULL DEFAULT 0, started_at TEXT, finished_at TEXT, created_at TEXT NOT NULL, FOREIGN KEY (customer_id) REFERENCES customer (id), FOREIGN KEY (package_id) REFERENCES package (id) );任务节点表是进度的数据来源。它的字段不多但每一个都承担明确职责seq保证展示顺序稳定weight支持加权进度status只允许四个值TODO、DOING、DONE、BLOCKED。BLOCKED这个状态非常实用它把“卡住了”这种只能靠嘴说的情况变成了可统计的数据一个订单里出现多个BLOCKED节点时管理者应该立刻介入。CREATE TABLE IF NOT EXISTS order_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, seq INTEGER NOT NULL, title TEXT NOT NULL, status TEXT NOT NULL DEFAULT TODO, weight INTEGER NOT NULL DEFAULT 1, updated_at TEXT NOT NULL, FOREIGN KEY (order_id) REFERENCES order_main (id) );3.3 操作日志表与索引设计日志表的设计遵循一条铁律只允许插入不允许更新和删除。它记录每一次状态流转和任务变更字段里from_status和to_status让变更方向可读operator记录操作人detail存补充信息。这张表日后会成为排查纠纷、分析效率的主要依据。CREATE TABLE IF NOT EXISTS order_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, action TEXT NOT NULL, from_status TEXT, to_status TEXT, operator TEXT DEFAULT system, detail TEXT DEFAULT , created_at TEXT NOT NULL, FOREIGN KEY (order_id) REFERENCES order_main (id) );索引方面别小看这几行。订单列表页最常见的查询是按状态筛选和按客户查历史日志表则永远是按订单号反查。没有索引时数据量上千就开始明显变慢加上之后基本无感CREATE INDEX IF NOT EXISTS idx_order_status ON order_main (status); CREATE INDEX IF NOT EXISTS idx_order_customer ON order_main (customer_id); CREATE INDEX IF NOT EXISTS idx_task_order ON order_task (order_id); CREATE INDEX IF NOT EXISTS idx_log_order ON order_log (order_id);3.4 完整建表脚本落地到 db.py把上面所有 SQL 拼进一个脚本再加上连接封装就得到了完整的db.py。这里用contextmanager封装事务是为了让业务代码只需要写with transaction() as conn:不用关心 commit、rollback 和 close避免“忘记提交”或“异常时连接泄漏”这两类经典事故。# 文件路径db.py import sqlite3 from contextlib import contextmanager from pathlib import Path BASE_DIR Path(__file__).resolve().parent DB_PATH BASE_DIR / data / studio.db SCHEMA CREATE TABLE IF NOT EXISTS customer ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, contact TEXT NOT NULL, note TEXT DEFAULT , created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS package ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, price_cents INTEGER NOT NULL DEFAULT 0, active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS package_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, package_id INTEGER NOT NULL, seq INTEGER NOT NULL, title TEXT NOT NULL, weight INTEGER NOT NULL DEFAULT 1, FOREIGN KEY (package_id) REFERENCES package (id) ); CREATE TABLE IF NOT EXISTS order_main ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT NOT NULL UNIQUE, customer_id INTEGER NOT NULL, package_id INTEGER NOT NULL, status TEXT NOT NULL DEFAULT PENDING, quoted_cents INTEGER NOT NULL DEFAULT 0, paid_cents INTEGER NOT NULL DEFAULT 0, started_at TEXT, finished_at TEXT, created_at TEXT NOT NULL, FOREIGN KEY (customer_id) REFERENCES customer (id), FOREIGN KEY (package_id) REFERENCES package (id) ); CREATE TABLE IF NOT EXISTS order_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, seq INTEGER NOT NULL, title TEXT NOT NULL, status TEXT NOT NULL DEFAULT TODO, weight INTEGER NOT NULL DEFAULT 1, updated_at TEXT NOT NULL, FOREIGN KEY (order_id) REFERENCES order_main (id) ); CREATE TABLE IF NOT EXISTS order_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, action TEXT NOT NULL, from_status TEXT, to_status TEXT, operator TEXT DEFAULT system, detail TEXT DEFAULT , created_at TEXT NOT NULL, FOREIGN KEY (order_id) REFERENCES order_main (id) ); CREATE INDEX IF NOT EXISTS idx_order_status ON order_main (status); CREATE INDEX IF NOT EXISTS idx_order_customer ON order_main (customer_id); CREATE INDEX IF NOT EXISTS idx_task_order ON order_task (order_id); CREATE INDEX IF NOT EXISTS idx_log_order ON order_log (order_id); def get_conn(): 创建连接并开启外键约束与 WAL 模式。 DB_PATH.parent.mkdir(parentsTrue, exist_okTrue) conn sqlite3.connect(DB_PATH, timeout10) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) conn.execute(PRAGMA journal_mode WAL) return conn contextmanager def transaction(): 事务上下文正常退出提交异常回滚最后一定关闭连接。 conn get_conn() try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def query_all(sql, params()): conn get_conn() try: return [dict(r) for r in conn.execute(sql, params).fetchall()] finally: conn.close() def query_one(sql, params()): conn get_conn() try: row conn.execute(sql, params).fetchone() return dict(row) if row else None finally: conn.close() def init_db(): with transaction() as conn: conn.executescript(SCHEMA) print(f[ok] 数据库已就绪{DB_PATH}) if __name__ __main__: init_db()这里有几个细节值得展开。第一sqlite3.Row让查询结果可以按列名访问dict(row)转成普通字典后 JSON 序列化不会报错这是很多人第一次用 SQLite 时踩的坑。第二PRAGMA foreign_keys ON必须显式开启SQLite 默认不检查外键如果忘了这行你插入一条指向不存在客户的订单也会成功等到查列表时才发现一堆孤儿数据。第三journal_mode WAL允许读写并发避免开发时一边开着页面一边跑脚本就出现database is locked。4. 核心逻辑实现状态机与进度计算4.1 把流转规则集中成一张表状态机的实现有很多种方式最轻量也最好维护的是“字典 校验函数”。字典的键是当前状态值是允许跳转到的目标状态集合。任何一次变更先查字典不允许就直接返回 409把非法流转挡在业务逻辑之外。# 文件路径app.py节选 TRANSITIONS { PENDING: {CONFIRMED, CANCELLED}, CONFIRMED: {IN_PROGRESS, CANCELLED}, IN_PROGRESS: {PENDING_ACCEPT, CANCELLED}, PENDING_ACCEPT: {DONE, IN_PROGRESS}, DONE: set(), CANCELLED: set(), }这样写的好处是规则可视。你一眼就能看出“已完成的订单不能改回去”也能看出“待验收可以退回进行中”。如果以后业务要求增加“挂起”状态只改这个字典和对应的前端展示即可所有接口自动生效不需要满项目找 if 判断。4.2 进度用加权计算而不是数个数假设一个订单有五个节点需求确认、清单梳理、第一批交付、第二批交付、归档确认。如果简单按“完成数 / 总数”算五个节点各占 20%但实际工作量显然不是这样两批交付占了八成精力。用计数法算出来的进度会严重虚高客户看到 60% 以为快好了结果还要等一周体验反而更差。加权计算解决了这个问题每个节点带一个weight进度等于“已完成节点的权重和 / 全部节点权重和”。模板里给交付类节点更大权重进度曲线就贴近真实工作量。SQL 层面可以用两个子查询一次算完SELECT o.id, o.order_no, o.status, c.name AS customer_name, p.code AS package_code, p.name AS package_name, (SELECT COALESCE(SUM(t.weight), 0) FROM order_task t WHERE t.order_id o.id) AS total_weight, (SELECT COALESCE(SUM(t.weight), 0) FROM order_task t WHERE t.order_id o.id AND t.status DONE) AS done_weight FROM order_main o JOIN customer c ON c.id o.customer_id JOIN package p ON p.id o.package_id ORDER BY o.id DESC LIMIT 100;COALESCE不能省因为当一个订单还没有任何任务节点时SUM会返回NULL直接除会抛异常。拿到total_weight和done_weight之后在 Python 里做一次除法并处理分母为零的情况即可。4.3 订单号生成与唯一性保障订单号是对外沟通的凭证必须满足三个要求唯一、可读、能看出日期。本文采用SO yyyymmdd 三位序号的格式例如SO20250612001。生成方式是先统计当天已有订单数再加一。def gen_order_no(conn): prefix datetime.now().strftime(SO%Y%m%d) row conn.execute( SELECT COUNT(*) AS c FROM order_main WHERE order_no LIKE ?, (prefix %,), ).fetchone() return f{prefix}{row[c] 1:03d}必须说明的是这个实现在单进程、低并发下是安全的但在多进程部署时存在极小的竞态窗口两个请求可能拿到同一个序号。真正的兜底是数据库上的UNIQUE约束即使生成了重复号插入也会失败并抛异常不会产生脏数据。如果你需要严格无冲突更稳妥的做法是用一个独立的序列表配合事务加锁或者直接用 UUID 做内部主键、订单号只作为展示字段。这一点在生产环境里要提前想清楚。5. 完整实战从零跑通一套订单管理系统5.1 编写初始化脚本先把示例数据和模板准备好这样启动后就能立刻看到效果。脚本用INSERT OR IGNORE保证重复执行不会插入重复数据方便反复调试。# 文件路径seed.py from datetime import datetime from db import init_db, transaction def now(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def main(): init_db() with transaction() as conn: conn.execute( INSERT OR IGNORE INTO customer(id, name, contact, note, created_at) VALUES (?,?,?,?,?), (1, 示例客户A, wechat:demo_a, 长期合作月结, now()), ) conn.execute( INSERT OR IGNORE INTO package(id, code, name, price_cents, active, created_at) VALUES (?,?,?,?,?,?), (1, PKG-MULTI-FULL, 多阶段全包服务, 128000, 1, now()), ) conn.execute(DELETE FROM package_task WHERE package_id 1) tasks [ (需求确认与资料核对, 1), (版本内容清单梳理, 3), (第一阶段交付与初验, 2), (第二阶段交付与终验, 3), (收尾归档与客户确认, 1), ] for seq, (title, weight) in enumerate(tasks, start1): conn.execute( INSERT INTO package_task(package_id, seq, title, weight) VALUES (?,?,?,?), (1, seq, title, weight), ) print([ok] 示例数据写入完成) if __name__ __main__: main()执行python seed.py看到两行[ok]输出就说明建表和初始化都成功了。此时data/studio.db已经生成你可以用任意 SQLite 客户端打开它检查也可以直接用sqlite3 data/studio.db .tables在命令行确认。5.2 编写接口层接下来是核心的应用文件。下面这份代码是可以直接运行的完整版本包含客户创建、订单创建自动复制模板、状态流转、任务状态更新、订单列表五个接口以及一个看板页面入口。# 文件路径app.py from datetime import datetime from flask import Flask, jsonify, render_template, request from db import query_all, query_one, transaction app Flask(__name__) TRANSITIONS { PENDING: {CONFIRMED, CANCELLED}, CONFIRMED: {IN_PROGRESS, CANCELLED}, IN_PROGRESS: {PENDING_ACCEPT, CANCELLED}, PENDING_ACCEPT: {DONE, IN_PROGRESS}, DONE: set(), CANCELLED: set(), } TASK_STATUS {TODO, DOING, DONE, BLOCKED} def now_str(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def gen_order_no(conn): prefix datetime.now().strftime(SO%Y%m%d) row conn.execute( SELECT COUNT(*) AS c FROM order_main WHERE order_no LIKE ?, (prefix %,), ).fetchone() return f{prefix}{row[c] 1:03d} def write_log(conn, order_id, action, from_status, to_status, operator, detail): conn.execute( INSERT INTO order_log(order_id, action, from_status, to_status, operator, detail, created_at) VALUES (?,?,?,?,?,?,?), (order_id, action, from_status, to_status, operator, detail, now_str()), ) app.get(/) def index(): return render_template(index.html) app.post(/api/customers) def create_customer(): data request.get_json(silentTrue) or {} name (data.get(name) or ).strip() contact (data.get(contact) or ).strip() if not name or not contact: return jsonify(code400, msgname 与 contact 均为必填), 400 with transaction() as conn: cur conn.execute( INSERT INTO customer(name, contact, note, created_at) VALUES (?,?,?,?), (name, contact, data.get(note, ), now_str()), ) cid cur.lastrowid return jsonify(code0, data{id: cid}), 201 app.post(/api/orders) def create_order(): data request.get_json(silentTrue) or {} customer_id data.get(customer_id) package_id data.get(package_id) operator data.get(operator, system) if not customer_id or not package_id: return jsonify(code400, msgcustomer_id 与 package_id 均为必填), 400 with transaction() as conn: cust conn.execute( SELECT id FROM customer WHERE id ?, (customer_id,) ).fetchone() pkg conn.execute( SELECT * FROM package WHERE id ? AND active 1, (package_id,) ).fetchone() if cust is None: return jsonify(code404, msg客户不存在), 404 if pkg is None: return jsonify(code404, msg服务包不存在或已下架), 404 order_no gen_order_no(conn) cur conn.execute( INSERT INTO order_main(order_no, customer_id, package_id, status, quoted_cents, paid_cents, created_at) VALUES (?,?,?,?,?,?,?), ( order_no, customer_id, package_id, PENDING, int(data.get(quoted_cents, pkg[price_cents])), 0, now_str(), ), ) order_id cur.lastrowid tpl conn.execute( SELECT title, weight FROM package_task WHERE package_id ? ORDER BY seq, (package_id,), ).fetchall() for seq, t in enumerate(tpl, start1): conn.execute( INSERT INTO order_task(order_id, seq, title, status, weight, updated_at) VALUES (?,?,?,?,?,?), (order_id, seq, t[title], TODO, t[weight], now_str()), ) write_log(conn, order_id, CREATE, None, PENDING, operator, fpackage{pkg[code]}) return jsonify(code0, data{id: order_id, order_no: order_no}), 201 app.post(/api/orders/int:order_id/transition) def transition(order_id): data request.get_json(silentTrue) or {} target (data.get(to) or ).upper() operator data.get(operator, system) with transaction() as conn: row conn.execute( SELECT * FROM order_main WHERE id ?, (order_id,) ).fetchone() if row is None: return jsonify(code404, msg订单不存在), 404 current row[status] if target not in TRANSITIONS.get(current, set()): return jsonify(code409, msgf非法流转{current} - {target}), 409 if current IN_PROGRESS and target PENDING_ACCEPT: left conn.execute( SELECT COUNT(*) AS c FROM order_task WHERE order_id ? AND status DONE, (order_id,), ).fetchone()[c] if left 0: return jsonify(code409, msgf仍有 {left} 个节点未完成不能提交验收), 409 started_at row[started_at] if target IN_PROGRESS and not started_at: started_at now_str() finished_at row[finished_at] if target in (DONE, CANCELLED): finished_at now_str() conn.execute( UPDATE order_main SET status ?, started_at ?, finished_at ? WHERE id ?, (target, started_at, finished_at, order_id), ) write_log(conn, order_id, TRANSITION, current, target, operator, data.get(detail, )) return jsonify(code0, data{id: order_id, status: target}) app.post(/api/orders/int:order_id/tasks/int:task_id/status) def update_task(order_id, task_id): data request.get_json(silentTrue) or {} status (data.get(status) or ).upper() if status not in TASK_STATUS: return jsonify(code400, msgfstatus 必须是 {sorted(TASK_STATUS)} 之一), 400 with transaction() as conn: task conn.execute( SELECT * FROM order_task WHERE id ? AND order_id ?, (task_id, order_id), ).fetchone() if task is None: return jsonify(code404, msg任务节点不存在或不属于该订单), 404 conn.execute( UPDATE order_task SET status ?, updated_at ? WHERE id ?, (status, now_str(), task_id), ) write_log(conn, order_id, TASK, None, None, data.get(operator, system), ftask#{task_id} {task[status]} - {status}) return jsonify(code0, data{task_id: task_id, status: status}) app.get(/api/orders) def list_orders(): rows query_all( SELECT o.id, o.order_no, o.status, o.quoted_cents, o.paid_cents, o.created_at, c.name AS customer_name, p.code AS package_code, p.name AS package_name, (SELECT COALESCE(SUM(t.weight), 0) FROM order_task t WHERE t.order_id o.id) AS total_weight, (SELECT COALESCE(SUM(t.weight), 0) FROM order_task t WHERE t.order_id o.id AND t.status DONE) AS done_weight FROM order_main o JOIN customer c ON c.id o.customer_id JOIN package p ON p.id o.package_id ORDER BY o.id DESC LIMIT 100 ) for r in rows: total r[total_weight] or 0 done r[done_weight] or 0 r[progress] round(done * 100 / total) if total else 0 return jsonify(code0, datarows) if __name__ __main__: app.run(host127.0.0.1, port5000, debugTrue)代码里有几处值得停下来看一下。第一所有写操作都包在transaction()里创建订单时“插订单 复制模板任务 写日志”要么全成功要么全失败不会出现有订单没任务节点的半成品。第二提交验收前的那次节点检查必须在事务内完成否则检查和更新之间存在时间窗口可能被其他请求插队。第三任务状态更新接口同时校验task_id和order_id防止有人拿 A 订单的任务 ID 去改 B 订单的数据。5.3 编写看板页面前端不需要框架一个fetch加模板字符串就够了。页面包含订单列表、进度条和“推进”按钮按钮会根据当前状态自动映射下一个合法状态非法状态直接置灰从界面上就避免了误操作。!-- 文件路径templates/index.html -- !DOCTYPE html html langzh-CN head meta charsetutf-8 / meta nameviewport contentwidthdevice-width, initial-scale1 / title订单进度看板/title style body { font-family: -apple-system, Microsoft YaHei, sans-serif; margin: 32px; color: #222; } h1 { font-size: 20px; } table { width: 100%; border-collapse: collapse; margin-top: 16px; } th, td { border-bottom: 1px solid #eee; padding: 10px 8px; font-size: 14px; text-align: left; } th { background: #fafafa; font-weight: 600; } .bar { display: inline-block; width: 120px; height: 8px; background: #eee; border-radius: 4px; overflow: hidden; vertical-align: middle; } .bar span { display: block; height: 100%; background: #2f7cf6; } button { padding: 4px 10px; border: 1px solid #2f7cf6; background: #fff; color: #2f7cf6; border-radius: 4px; cursor: pointer; } button:disabled { border-color: #ccc; color: #bbb; cursor: not-allowed; } /style /head body h1订单进度看板/h1 table idgrid thead tr th订单号/thth客户/thth服务包/thth状态/thth进度/thth创建时间/thth操作/th /tr /thead tbody/tbody /table script const NEXT { PENDING: CONFIRMED, CONFIRMED: IN_PROGRESS, IN_PROGRESS: PENDING_ACCEPT, PENDING_ACCEPT: DONE }; async function load() { const res await fetch(/api/orders); const body await res.json(); const tbody document.querySelector(#grid tbody); tbody.innerHTML (body.data || []).map(r { const next NEXT[r.status] || ; return tr td${r.order_no}/td td${r.customer_name}/td td${r.package_code}/td td${r.status}/td tdspan classbarspan stylewidth:${r.progress}%/span/span ${r.progress}%/td td${r.created_at}/td tdbutton>python seed.py # 初始化数据库和示例数据 python app.py # 启动服务默认 http://127.0.0.1:5000然后打开浏览器访问http://127.0.0.1:5000你会看到一个空列表。接着用 curl 创建一条订单curl -X POST http://127.0.0.1:5000/api/orders \ -H Content-Type: application/json \ -d {customer_id: 1, package_id: 1, operator: alice}预期返回{code: 0, data: {id: 1, order_no: SO20250612001}}刷新页面订单就会出现在列表里状态是PENDING进度 0%。接下来把第一个任务节点标记为完成curl -X POST http://127.0.0.1:5000/api/orders/1/tasks/1/status \ -H Content-Type: application/json \ -d {status: DONE, operator: alice}再刷新页面进度会跳到 10%因为总权重是 1323110而第一个节点权重为 1。这个数字是算出来的不是填进去的这正是整套设计的价值所在。5.5 结果说明与状态流转演示继续按顺序推进PENDING → CONFIRMED → IN_PROGRESS每一步都能在页面按钮上点出来。此时如果你尝试直接点“推进到 PENDING_ACCEPT”会收到 409 和提示“仍有 4 个节点未完成不能提交验收”。把所有任务节点都标记为DONE之后再推进就能顺利进入待验收最后到DONE。curl -X POST http://127.0.0.1:5000/api/orders/1/transition \ -H Content-Type: application/json \ -d {to: CONFIRMED, operator: alice}此时订单的started_at会在进入IN_PROGRESS时自动写入finished_at会在进入DONE或CANCELLED时自动写入。你可以直接查数据库验证SELECT order_no, status, started_at, finished_at FROM order_main WHERE id 1;同时order_log表里已经累积了一串记录包含每一次TRANSITION和TASK变更。这份日志在出现争议时能直接拿给对方看比回忆靠谱得多。6. 常见问题与排查思路6.1 高频报错速查表先给一张表覆盖本文代码在运行过程中最可能遇到的几类问题。遇到报错时按表格对照能省掉大量猜测时间。问题现象常见原因解决思路sqlite3.OperationalError: database is locked多个进程同时写库或上一个连接未关闭启用 WAL缩短事务范围确认异常分支也执行了 close插入订单报FOREIGN KEY constraint failed客户或服务包 ID 不存在或外键检查未开启先查 customer/package 是否存在确认PRAGMA foreign_keys ON订单列表progress显示 0任务节点全为TODO或模板未复制成功查order_task是否有该订单的数据提交验收一直返回 409存在TODO、DOING或BLOCKED的节点查未完成节点数量逐个推进TypeError: sqlite3.Row object is not subscriptable未转成字典就按 JSON 返回查询结果统一dict(row)接口 500 但日志无信息debugTrue未开或异常被吞打开调试模式查看终端完整堆栈页面刷新后数据不更新浏览器缓存了fetch结果加时间戳参数或设置Cache-Control订单号重复插入失败并发创建时序号撞车依赖UNIQUE约束兜底或改用独立序列表6.2 外键失败的完整排查路径FOREIGN KEY constraint failed是新手最容易卡住的报错之一因为提示信息里没有告诉你哪张表、哪个字段出问题。按下面四步走基本能定位第一步确认报错发生在哪条 SQL。在transaction()外层加except sqlite3.Error as e: print(e, sql)或者打开 
长河君 · 2026/10/26 2:17:26 · 阅读 1.2k 任务型订单管理系统实战：状态机、加权进度与Flask看板
近期有不少做技术服务、任务代办类业务的朋友在问同一个问题一个订单动不动就横跨好几个版本阶段客户问“做到哪一步了”的时候只能翻聊天记录手数订单越多越乱。这篇文章就围绕这个真实痛点展开把「任务型订单管理」从数据建模、状态机设计到接口和看板页面完整实现一遍。适合有 Python 基础、想动手做一个内部管理系统的开发者也适合正在用表格人工跟单、准备把流程工具化的同学。读完之后你能掌握三件事怎么把一张“看着简单”的订单拆成可计算的数据结构、怎么用状态机约束业务流转防止脏数据、怎么用 Flask SQLite 在半小时内跑起一套能用的订单看板。文中所有代码都可以直接复制运行遇到报错也能在排查章节里找到对应解法。1. 背景与核心概念1.1 订单型业务真正缺的不是表格而是结构化先说清楚这篇文章要解决的问题。类似“某版本全包已完成”“本单已交付感谢信任”这样的通知本质上是订单生命周期走到了终点。麻烦的地方在于这类订单往往不是一次点击就结束的动作而是由若干阶段性目标串起来的前期确认需求、中期分批交付、后期验收归档。如果订单信息只存在于聊天记录和脑子里就会出现三个典型问题进度无法量化、责任无法追溯、历史无法复用。用 Excel 跟单在订单量小于十单时确实够用但一旦订单并发上来你会发现自己在做的是“人肉状态同步”。客户问进度你要去看聊天记录同事交接你要口述半小时出了问题想复盘找不到任何操作痕迹。这些问题的共同根源是业务信息没有被结构化状态变化没有被记录下来。所以这篇文章不讲“如何做业务”只讲“如何用代码把业务装进一套可计算的结构里”。具体做法是把一条订单拆成四层对象客户、服务包、订单、任务节点用数据库表把它们固定下来再用一个受约束的状态机控制订单怎么往前走最后用一个看板页面把进度可视化出来。这套结构不依赖任何特定行业做内容交付、设计外包、设备巡检、课程陪跑都能套用。1.2 四个核心概念客户、服务包、订单、任务节点在动手建表之前必须先把概念边界划清楚。很多同学写出来的订单系统最后变成一坨 JSON 字段堆在数据库里就是因为没有把这几层分开。**客户customer**是付费主体也是长期资产。它的关键属性是联系方式与备注一个客户可以有多条订单订单之间可以互相参考。把客户独立成表而不是在订单里存一个字符串名字好处是客户改名时历史订单不用批量改也方便后续做“复购统计”。**服务包package**是对“卖什么”的抽象。它描述的是可复用的产品定义比如“单阶段基础版”“多阶段全包版”包含价格、预估工时、以及一份任务模板。服务包是模板订单是实例这个区分非常重要订单创建时把模板复制一份成任务节点之后修改服务包不会影响已经在跑的订单。**订单order**是一次具体的交易与交付过程它关联一个客户和一个服务包拥有独立的状态、报价、实收金额和时间戳。订单号是对外沟通的唯一凭证必须唯一且可读。**任务节点order_task**是订单内部的执行单元带有序号、标题、权重和状态。进度百分比不是靠人工填的而是从任务节点聚合算出来的。这一点是整套设计的关键后文会专门展开。把这四层的关系画成文字就是一个客户对应多条订单一个服务包对应多条订单一条订单对应多条任务节点。理解了这句后面的表结构就是自然推导出来的结果。1.3 状态机为什么不能用一个字段随便改初学者最常见的写法是给订单一个status字段然后哪个接口想改就改。这种做法在小规模下看不出问题但一旦有了并发或者多人协作就会出现“已完成的订单被改回进行中”“已取消的订单还能继续推进”这类脏数据而且很难查是哪一步出的错。正确做法是把状态流转规则集中定义成一张“允许表”任何一次状态变更都必须先校验当前状态是否允许跳到目标状态。这在软件工程里叫状态机State Machine它的价值不在于概念好听而在于把散落在各个接口里的 if 判断收拢到一处改规则只改一个地方。本文实现的订单状态包含六个PENDING待确认、CONFIRMED已确认、IN_PROGRESS进行中、PENDING_ACCEPT待验收、DONE已完成、CANCELLED已取消。流转关系是有向的待确认只能到已确认或已取消进行中提交验收时系统会额外校验所有任务节点是否都已完成待验收可以退回进行中这是为了应对“客户提出补充修改”的真实场景。终态DONE和CANCELLED不再允许任何流转这样历史就不会被篡改。顺带提一句状态机还有一个容易被忽略的收益它天然生成了一份业务日志。每次流转都往日志表写一条记录谁在什么时候把订单从什么状态改成了什么状态一目了然。后面章节的审计能力就是建立在这个基础之上的。1.4 技术边界本文实现什么、不实现什么为了让文章聚焦这里明确一下范围。本文实现的是一个内部使用的订单管理后端与只读看板包括建表、接口、进度计算、状态校验、操作日志和页面渲染。不实现的部分包括支付对接、多租户、复杂权限体系、消息推送、前端工程化。这些内容每一个都值得单独写一篇硬塞进来只会让代码变得不可运行。另外必须强调一条边界如果业务涉及第三方平台的账号与虚拟财产操作请务必先确认对应产品与平台的用户协议是否允许本文讨论的仅仅是订单数据如何管理不涉及任何账号代操作、自动化脚本、绕过风控或者规避平台规则的技术手段。任何系统的第一优先级都是合规技术方案再好前提不合规也没有意义。2. 环境准备与版本说明2.1 运行环境与技术选型本文选择 Python Flask SQLite 的组合原因很实际零外部服务依赖装完 Python 就能跑非常适合内部工具和中小规模业务。如果你的团队已经在用 Java 或者 Node思路完全一样把数据访问层换掉即可表结构和状态机设计是可以直接复用的。环境要求如下版本请根据你的实际环境调整本文重点演示配置思路而不是锁死版本号操作系统Windows 10/11、macOS、主流 Linux 发行版均可Python3.10 及以上用到了contextlib和 f-string3.10 更稳妥Flask3.x 系列数据库SQLite 3Python 标准库自带无需单独安装开发工具VS Code、PyCharm 或任意编辑器接口调试curl、Postman 或浏览器开发者工具一句话解释选型理由Flask 足够轻路由写起来直观SQLite 单文件存储备份就是复制一个文件对本量级的订单系统完全够用而且它支持事务和外键能让我们把数据一致性讲清楚。如果日后订单量涨到需要多人高并发写入再迁移到 PostgreSQL 或 MySQLSQL 语句的改动量很小。2.2 项目目录结构在开始写代码之前先把目录建好。结构清晰的目录能让你在后面排查问题时少走很多弯路建议严格按下面这个结构组织order-studio/ ├── app.py # Flask 应用与全部接口 ├── db.py # 数据库连接、建表脚本、事务封装 ├── seed.py # 初始化示例数据 ├── requirements.txt # 依赖清单 ├── data/ │ └── studio.db # SQLite 数据文件首次运行自动生成 └── templates/ └── index.html # 订单看板页面这里的data/目录故意和代码分开因为它是唯一需要备份的东西。生产环境里你会希望把数据文件挂到独立磁盘上而不是和代码混在一起这样发布新版本时直接替换代码目录就行不会误删数据。2.3 依赖安装与虚拟环境强烈建议使用虚拟环境避免污染全局 Python 环境。下面三条命令按顺序执行即可# 1. 创建虚拟环境 python -m venv .venv # 2. 激活虚拟环境Windows .venv\Scripts\activate # 激活虚拟环境macOS / Linux source .venv/bin/activate # 3. 安装依赖 pip install -r requirements.txtrequirements.txt的内容如下Flask3.0.3安装完成后执行pip list确认 Flask 已经出现在列表里。如果激活虚拟环境后命令提示符前面没有出现(.venv)说明激活失败通常是在 Windows 上用了 PowerShell 但执行策略有限制可以在 PowerShell 里用.venv\Scripts\Activate.ps1或者干脆切到 CMD 执行activate.bat。这一步看似琐碎但它是后面所有报错排查的地基环境不对后面全是玄学问题。3. 数据模型与表结构设计3.1 客户表与服务包表客户表的设计原则是“宁多勿少”把可能需要检索的字段都留出来。contact字段用来存联系方式不做格式约束因为现实中它可能是手机号、微信号、邮箱甚至一个内部工单链接。note用来存业务备注比如“长期合作结算周期月结”。CREATE TABLE IF NOT EXISTS customer ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, contact TEXT NOT NULL, note TEXT DEFAULT , created_at TEXT NOT NULL );服务包表多了三个关键字段code是业务编码必须唯一方便口头沟通时直接说编码price_cents用整数存“分”这个设计后面会专门解释原因active是上下架开关用软下架代替删除保证历史订单还能正确关联到服务包信息。CREATE TABLE IF NOT EXISTS package ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, price_cents INTEGER NOT NULL DEFAULT 0, active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL );再往下是服务包的任务模板表。这张表是整套设计里最容易被省略、但收益最大的一张它让“创建一个订单”从手工填十个节点变成选一个服务包自动生成。模板节点的seq决定顺序weight决定它在总进度里的权重。CREATE TABLE IF NOT EXISTS package_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, package_id INTEGER NOT NULL, seq INTEGER NOT NULL, title TEXT NOT NULL, weight INTEGER NOT NULL DEFAULT 1, FOREIGN KEY (package_id) REFERENCES package (id) );3.2 订单主表与任务节点表订单主表承载的是“这笔交易”的全部事实谁买的、买的什么、多少钱、当前什么状态、什么时候开始什么时候结束。注意quoted_cents报价和paid_cents实收是两个字段这是从实践中得出的经验因为报价和实收不一致是常态有优惠、有追加、有尾款合成一个字段最后一定会出问题。CREATE TABLE IF NOT EXISTS order_main ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT NOT NULL UNIQUE, customer_id INTEGER NOT NULL, package_id INTEGER NOT NULL, status TEXT NOT NULL DEFAULT PENDING, quoted_cents INTEGER NOT NULL DEFAULT 0, paid_cents INTEGER NOT NULL DEFAULT 0, started_at TEXT, finished_at TEXT, created_at TEXT NOT NULL, FOREIGN KEY (customer_id) REFERENCES customer (id), FOREIGN KEY (package_id) REFERENCES package (id) );任务节点表是进度的数据来源。它的字段不多但每一个都承担明确职责seq保证展示顺序稳定weight支持加权进度status只允许四个值TODO、DOING、DONE、BLOCKED。BLOCKED这个状态非常实用它把“卡住了”这种只能靠嘴说的情况变成了可统计的数据一个订单里出现多个BLOCKED节点时管理者应该立刻介入。CREATE TABLE IF NOT EXISTS order_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, seq INTEGER NOT NULL, title TEXT NOT NULL, status TEXT NOT NULL DEFAULT TODO, weight INTEGER NOT NULL DEFAULT 1, updated_at TEXT NOT NULL, FOREIGN KEY (order_id) REFERENCES order_main (id) );3.3 操作日志表与索引设计日志表的设计遵循一条铁律只允许插入不允许更新和删除。它记录每一次状态流转和任务变更字段里from_status和to_status让变更方向可读operator记录操作人detail存补充信息。这张表日后会成为排查纠纷、分析效率的主要依据。CREATE TABLE IF NOT EXISTS order_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, action TEXT NOT NULL, from_status TEXT, to_status TEXT, operator TEXT DEFAULT system, detail TEXT DEFAULT , created_at TEXT NOT NULL, FOREIGN KEY (order_id) REFERENCES order_main (id) );索引方面别小看这几行。订单列表页最常见的查询是按状态筛选和按客户查历史日志表则永远是按订单号反查。没有索引时数据量上千就开始明显变慢加上之后基本无感CREATE INDEX IF NOT EXISTS idx_order_status ON order_main (status); CREATE INDEX IF NOT EXISTS idx_order_customer ON order_main (customer_id); CREATE INDEX IF NOT EXISTS idx_task_order ON order_task (order_id); CREATE INDEX IF NOT EXISTS idx_log_order ON order_log (order_id);3.4 完整建表脚本落地到 db.py把上面所有 SQL 拼进一个脚本再加上连接封装就得到了完整的db.py。这里用contextmanager封装事务是为了让业务代码只需要写with transaction() as conn:不用关心 commit、rollback 和 close避免“忘记提交”或“异常时连接泄漏”这两类经典事故。# 文件路径db.py import sqlite3 from contextlib import contextmanager from pathlib import Path BASE_DIR Path(__file__).resolve().parent DB_PATH BASE_DIR / data / studio.db SCHEMA CREATE TABLE IF NOT EXISTS customer ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, contact TEXT NOT NULL, note TEXT DEFAULT , created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS package ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, price_cents INTEGER NOT NULL DEFAULT 0, active INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS package_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, package_id INTEGER NOT NULL, seq INTEGER NOT NULL, title TEXT NOT NULL, weight INTEGER NOT NULL DEFAULT 1, FOREIGN KEY (package_id) REFERENCES package (id) ); CREATE TABLE IF NOT EXISTS order_main ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT NOT NULL UNIQUE, customer_id INTEGER NOT NULL, package_id INTEGER NOT NULL, status TEXT NOT NULL DEFAULT PENDING, quoted_cents INTEGER NOT NULL DEFAULT 0, paid_cents INTEGER NOT NULL DEFAULT 0, started_at TEXT, finished_at TEXT, created_at TEXT NOT NULL, FOREIGN KEY (customer_id) REFERENCES customer (id), FOREIGN KEY (package_id) REFERENCES package (id) ); CREATE TABLE IF NOT EXISTS order_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, seq INTEGER NOT NULL, title TEXT NOT NULL, status TEXT NOT NULL DEFAULT TODO, weight INTEGER NOT NULL DEFAULT 1, updated_at TEXT NOT NULL, FOREIGN KEY (order_id) REFERENCES order_main (id) ); CREATE TABLE IF NOT EXISTS order_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, action TEXT NOT NULL, from_status TEXT, to_status TEXT, operator TEXT DEFAULT system, detail TEXT DEFAULT , created_at TEXT NOT NULL, FOREIGN KEY (order_id) REFERENCES order_main (id) ); CREATE INDEX IF NOT EXISTS idx_order_status ON order_main (status); CREATE INDEX IF NOT EXISTS idx_order_customer ON order_main (customer_id); CREATE INDEX IF NOT EXISTS idx_task_order ON order_task (order_id); CREATE INDEX IF NOT EXISTS idx_log_order ON order_log (order_id); def get_conn(): 创建连接并开启外键约束与 WAL 模式。 DB_PATH.parent.mkdir(parentsTrue, exist_okTrue) conn sqlite3.connect(DB_PATH, timeout10) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) conn.execute(PRAGMA journal_mode WAL) return conn contextmanager def transaction(): 事务上下文正常退出提交异常回滚最后一定关闭连接。 conn get_conn() try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def query_all(sql, params()): conn get_conn() try: return [dict(r) for r in conn.execute(sql, params).fetchall()] finally: conn.close() def query_one(sql, params()): conn get_conn() try: row conn.execute(sql, params).fetchone() return dict(row) if row else None finally: conn.close() def init_db(): with transaction() as conn: conn.executescript(SCHEMA) print(f[ok] 数据库已就绪{DB_PATH}) if __name__ __main__: init_db()这里有几个细节值得展开。第一sqlite3.Row让查询结果可以按列名访问dict(row)转成普通字典后 JSON 序列化不会报错这是很多人第一次用 SQLite 时踩的坑。第二PRAGMA foreign_keys ON必须显式开启SQLite 默认不检查外键如果忘了这行你插入一条指向不存在客户的订单也会成功等到查列表时才发现一堆孤儿数据。第三journal_mode WAL允许读写并发避免开发时一边开着页面一边跑脚本就出现database is locked。4. 核心逻辑实现状态机与进度计算4.1 把流转规则集中成一张表状态机的实现有很多种方式最轻量也最好维护的是“字典 校验函数”。字典的键是当前状态值是允许跳转到的目标状态集合。任何一次变更先查字典不允许就直接返回 409把非法流转挡在业务逻辑之外。# 文件路径app.py节选 TRANSITIONS { PENDING: {CONFIRMED, CANCELLED}, CONFIRMED: {IN_PROGRESS, CANCELLED}, IN_PROGRESS: {PENDING_ACCEPT, CANCELLED}, PENDING_ACCEPT: {DONE, IN_PROGRESS}, DONE: set(), CANCELLED: set(), }这样写的好处是规则可视。你一眼就能看出“已完成的订单不能改回去”也能看出“待验收可以退回进行中”。如果以后业务要求增加“挂起”状态只改这个字典和对应的前端展示即可所有接口自动生效不需要满项目找 if 判断。4.2 进度用加权计算而不是数个数假设一个订单有五个节点需求确认、清单梳理、第一批交付、第二批交付、归档确认。如果简单按“完成数 / 总数”算五个节点各占 20%但实际工作量显然不是这样两批交付占了八成精力。用计数法算出来的进度会严重虚高客户看到 60% 以为快好了结果还要等一周体验反而更差。加权计算解决了这个问题每个节点带一个weight进度等于“已完成节点的权重和 / 全部节点权重和”。模板里给交付类节点更大权重进度曲线就贴近真实工作量。SQL 层面可以用两个子查询一次算完SELECT o.id, o.order_no, o.status, c.name AS customer_name, p.code AS package_code, p.name AS package_name, (SELECT COALESCE(SUM(t.weight), 0) FROM order_task t WHERE t.order_id o.id) AS total_weight, (SELECT COALESCE(SUM(t.weight), 0) FROM order_task t WHERE t.order_id o.id AND t.status DONE) AS done_weight FROM order_main o JOIN customer c ON c.id o.customer_id JOIN package p ON p.id o.package_id ORDER BY o.id DESC LIMIT 100;COALESCE不能省因为当一个订单还没有任何任务节点时SUM会返回NULL直接除会抛异常。拿到total_weight和done_weight之后在 Python 里做一次除法并处理分母为零的情况即可。4.3 订单号生成与唯一性保障订单号是对外沟通的凭证必须满足三个要求唯一、可读、能看出日期。本文采用SO yyyymmdd 三位序号的格式例如SO20250612001。生成方式是先统计当天已有订单数再加一。def gen_order_no(conn): prefix datetime.now().strftime(SO%Y%m%d) row conn.execute( SELECT COUNT(*) AS c FROM order_main WHERE order_no LIKE ?, (prefix %,), ).fetchone() return f{prefix}{row[c] 1:03d}必须说明的是这个实现在单进程、低并发下是安全的但在多进程部署时存在极小的竞态窗口两个请求可能拿到同一个序号。真正的兜底是数据库上的UNIQUE约束即使生成了重复号插入也会失败并抛异常不会产生脏数据。如果你需要严格无冲突更稳妥的做法是用一个独立的序列表配合事务加锁或者直接用 UUID 做内部主键、订单号只作为展示字段。这一点在生产环境里要提前想清楚。5. 完整实战从零跑通一套订单管理系统5.1 编写初始化脚本先把示例数据和模板准备好这样启动后就能立刻看到效果。脚本用INSERT OR IGNORE保证重复执行不会插入重复数据方便反复调试。# 文件路径seed.py from datetime import datetime from db import init_db, transaction def now(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def main(): init_db() with transaction() as conn: conn.execute( INSERT OR IGNORE INTO customer(id, name, contact, note, created_at) VALUES (?,?,?,?,?), (1, 示例客户A, wechat:demo_a, 长期合作月结, now()), ) conn.execute( INSERT OR IGNORE INTO package(id, code, name, price_cents, active, created_at) VALUES (?,?,?,?,?,?), (1, PKG-MULTI-FULL, 多阶段全包服务, 128000, 1, now()), ) conn.execute(DELETE FROM package_task WHERE package_id 1) tasks [ (需求确认与资料核对, 1), (版本内容清单梳理, 3), (第一阶段交付与初验, 2), (第二阶段交付与终验, 3), (收尾归档与客户确认, 1), ] for seq, (title, weight) in enumerate(tasks, start1): conn.execute( INSERT INTO package_task(package_id, seq, title, weight) VALUES (?,?,?,?), (1, seq, title, weight), ) print([ok] 示例数据写入完成) if __name__ __main__: main()执行python seed.py看到两行[ok]输出就说明建表和初始化都成功了。此时data/studio.db已经生成你可以用任意 SQLite 客户端打开它检查也可以直接用sqlite3 data/studio.db .tables在命令行确认。5.2 编写接口层接下来是核心的应用文件。下面这份代码是可以直接运行的完整版本包含客户创建、订单创建自动复制模板、状态流转、任务状态更新、订单列表五个接口以及一个看板页面入口。# 文件路径app.py from datetime import datetime from flask import Flask, jsonify, render_template, request from db import query_all, query_one, transaction app Flask(__name__) TRANSITIONS { PENDING: {CONFIRMED, CANCELLED}, CONFIRMED: {IN_PROGRESS, CANCELLED}, IN_PROGRESS: {PENDING_ACCEPT, CANCELLED}, PENDING_ACCEPT: {DONE, IN_PROGRESS}, DONE: set(), CANCELLED: set(), } TASK_STATUS {TODO, DOING, DONE, BLOCKED} def now_str(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def gen_order_no(conn): prefix datetime.now().strftime(SO%Y%m%d) row conn.execute( SELECT COUNT(*) AS c FROM order_main WHERE order_no LIKE ?, (prefix %,), ).fetchone() return f{prefix}{row[c] 1:03d} def write_log(conn, order_id, action, from_status, to_status, operator, detail): conn.execute( INSERT INTO order_log(order_id, action, from_status, to_status, operator, detail, created_at) VALUES (?,?,?,?,?,?,?), (order_id, action, from_status, to_status, operator, detail, now_str()), ) app.get(/) def index(): return render_template(index.html) app.post(/api/customers) def create_customer(): data request.get_json(silentTrue) or {} name (data.get(name) or ).strip() contact (data.get(contact) or ).strip() if not name or not contact: return jsonify(code400, msgname 与 contact 均为必填), 400 with transaction() as conn: cur conn.execute( INSERT INTO customer(name, contact, note, created_at) VALUES (?,?,?,?), (name, contact, data.get(note, ), now_str()), ) cid cur.lastrowid return jsonify(code0, data{id: cid}), 201 app.post(/api/orders) def create_order(): data request.get_json(silentTrue) or {} customer_id data.get(customer_id) package_id data.get(package_id) operator data.get(operator, system) if not customer_id or not package_id: return jsonify(code400, msgcustomer_id 与 package_id 均为必填), 400 with transaction() as conn: cust conn.execute( SELECT id FROM customer WHERE id ?, (customer_id,) ).fetchone() pkg conn.execute( SELECT * FROM package WHERE id ? AND active 1, (package_id,) ).fetchone() if cust is None: return jsonify(code404, msg客户不存在), 404 if pkg is None: return jsonify(code404, msg服务包不存在或已下架), 404 order_no gen_order_no(conn) cur conn.execute( INSERT INTO order_main(order_no, customer_id, package_id, status, quoted_cents, paid_cents, created_at) VALUES (?,?,?,?,?,?,?), ( order_no, customer_id, package_id, PENDING, int(data.get(quoted_cents, pkg[price_cents])), 0, now_str(), ), ) order_id cur.lastrowid tpl conn.execute( SELECT title, weight FROM package_task WHERE package_id ? ORDER BY seq, (package_id,), ).fetchall() for seq, t in enumerate(tpl, start1): conn.execute( INSERT INTO order_task(order_id, seq, title, status, weight, updated_at) VALUES (?,?,?,?,?,?), (order_id, seq, t[title], TODO, t[weight], now_str()), ) write_log(conn, order_id, CREATE, None, PENDING, operator, fpackage{pkg[code]}) return jsonify(code0, data{id: order_id, order_no: order_no}), 201 app.post(/api/orders/int:order_id/transition) def transition(order_id): data request.get_json(silentTrue) or {} target (data.get(to) or ).upper() operator data.get(operator, system) with transaction() as conn: row conn.execute( SELECT * FROM order_main WHERE id ?, (order_id,) ).fetchone() if row is None: return jsonify(code404, msg订单不存在), 404 current row[status] if target not in TRANSITIONS.get(current, set()): return jsonify(code409, msgf非法流转{current} - {target}), 409 if current IN_PROGRESS and target PENDING_ACCEPT: left conn.execute( SELECT COUNT(*) AS c FROM order_task WHERE order_id ? AND status DONE, (order_id,), ).fetchone()[c] if left 0: return jsonify(code409, msgf仍有 {left} 个节点未完成不能提交验收), 409 started_at row[started_at] if target IN_PROGRESS and not started_at: started_at now_str() finished_at row[finished_at] if target in (DONE, CANCELLED): finished_at now_str() conn.execute( UPDATE order_main SET status ?, started_at ?, finished_at ? WHERE id ?, (target, started_at, finished_at, order_id), ) write_log(conn, order_id, TRANSITION, current, target, operator, data.get(detail, )) return jsonify(code0, data{id: order_id, status: target}) app.post(/api/orders/int:order_id/tasks/int:task_id/status) def update_task(order_id, task_id): data request.get_json(silentTrue) or {} status (data.get(status) or ).upper() if status not in TASK_STATUS: return jsonify(code400, msgfstatus 必须是 {sorted(TASK_STATUS)} 之一), 400 with transaction() as conn: task conn.execute( SELECT * FROM order_task WHERE id ? AND order_id ?, (task_id, order_id), ).fetchone() if task is None: return jsonify(code404, msg任务节点不存在或不属于该订单), 404 conn.execute( UPDATE order_task SET status ?, updated_at ? WHERE id ?, (status, now_str(), task_id), ) write_log(conn, order_id, TASK, None, None, data.get(operator, system), ftask#{task_id} {task[status]} - {status}) return jsonify(code0, data{task_id: task_id, status: status}) app.get(/api/orders) def list_orders(): rows query_all( SELECT o.id, o.order_no, o.status, o.quoted_cents, o.paid_cents, o.created_at, c.name AS customer_name, p.code AS package_code, p.name AS package_name, (SELECT COALESCE(SUM(t.weight), 0) FROM order_task t WHERE t.order_id o.id) AS total_weight, (SELECT COALESCE(SUM(t.weight), 0) FROM order_task t WHERE t.order_id o.id AND t.status DONE) AS done_weight FROM order_main o JOIN customer c ON c.id o.customer_id JOIN package p ON p.id o.package_id ORDER BY o.id DESC LIMIT 100 ) for r in rows: total r[total_weight] or 0 done r[done_weight] or 0 r[progress] round(done * 100 / total) if total else 0 return jsonify(code0, datarows) if __name__ __main__: app.run(host127.0.0.1, port5000, debugTrue)代码里有几处值得停下来看一下。第一所有写操作都包在transaction()里创建订单时“插订单 复制模板任务 写日志”要么全成功要么全失败不会出现有订单没任务节点的半成品。第二提交验收前的那次节点检查必须在事务内完成否则检查和更新之间存在时间窗口可能被其他请求插队。第三任务状态更新接口同时校验task_id和order_id防止有人拿 A 订单的任务 ID 去改 B 订单的数据。5.3 编写看板页面前端不需要框架一个fetch加模板字符串就够了。页面包含订单列表、进度条和“推进”按钮按钮会根据当前状态自动映射下一个合法状态非法状态直接置灰从界面上就避免了误操作。!-- 文件路径templates/index.html -- !DOCTYPE html html langzh-CN head meta charsetutf-8 / meta nameviewport contentwidthdevice-width, initial-scale1 / title订单进度看板/title style body { font-family: -apple-system, Microsoft YaHei, sans-serif; margin: 32px; color: #222; } h1 { font-size: 20px; } table { width: 100%; border-collapse: collapse; margin-top: 16px; } th, td { border-bottom: 1px solid #eee; padding: 10px 8px; font-size: 14px; text-align: left; } th { background: #fafafa; font-weight: 600; } .bar { display: inline-block; width: 120px; height: 8px; background: #eee; border-radius: 4px; overflow: hidden; vertical-align: middle; } .bar span { display: block; height: 100%; background: #2f7cf6; } button { padding: 4px 10px; border: 1px solid #2f7cf6; background: #fff; color: #2f7cf6; border-radius: 4px; cursor: pointer; } button:disabled { border-color: #ccc; color: #bbb; cursor: not-allowed; } /style /head body h1订单进度看板/h1 table idgrid thead tr th订单号/thth客户/thth服务包/thth状态/thth进度/thth创建时间/thth操作/th /tr /thead tbody/tbody /table script const NEXT { PENDING: CONFIRMED, CONFIRMED: IN_PROGRESS, IN_PROGRESS: PENDING_ACCEPT, PENDING_ACCEPT: DONE }; async function load() { const res await fetch(/api/orders); const body await res.json(); const tbody document.querySelector(#grid tbody); tbody.innerHTML (body.data || []).map(r { const next NEXT[r.status] || ; return tr td${r.order_no}/td td${r.customer_name}/td td${r.package_code}/td td${r.status}/td tdspan classbarspan stylewidth:${r.progress}%/span/span ${r.progress}%/td td${r.created_at}/td tdbutton>python seed.py # 初始化数据库和示例数据 python app.py # 启动服务默认 http://127.0.0.1:5000然后打开浏览器访问http://127.0.0.1:5000你会看到一个空列表。接着用 curl 创建一条订单curl -X POST http://127.0.0.1:5000/api/orders \ -H Content-Type: application/json \ -d {customer_id: 1, package_id: 1, operator: alice}预期返回{code: 0, data: {id: 1, order_no: SO20250612001}}刷新页面订单就会出现在列表里状态是PENDING进度 0%。接下来把第一个任务节点标记为完成curl -X POST http://127.0.0.1:5000/api/orders/1/tasks/1/status \ -H Content-Type: application/json \ -d {status: DONE, operator: alice}再刷新页面进度会跳到 10%因为总权重是 1323110而第一个节点权重为 1。这个数字是算出来的不是填进去的这正是整套设计的价值所在。5.5 结果说明与状态流转演示继续按顺序推进PENDING → CONFIRMED → IN_PROGRESS每一步都能在页面按钮上点出来。此时如果你尝试直接点“推进到 PENDING_ACCEPT”会收到 409 和提示“仍有 4 个节点未完成不能提交验收”。把所有任务节点都标记为DONE之后再推进就能顺利进入待验收最后到DONE。curl -X POST http://127.0.0.1:5000/api/orders/1/transition \ -H Content-Type: application/json \ -d {to: CONFIRMED, operator: alice}此时订单的started_at会在进入IN_PROGRESS时自动写入finished_at会在进入DONE或CANCELLED时自动写入。你可以直接查数据库验证SELECT order_no, status, started_at, finished_at FROM order_main WHERE id 1;同时order_log表里已经累积了一串记录包含每一次TRANSITION和TASK变更。这份日志在出现争议时能直接拿给对方看比回忆靠谱得多。6. 常见问题与排查思路6.1 高频报错速查表先给一张表覆盖本文代码在运行过程中最可能遇到的几类问题。遇到报错时按表格对照能省掉大量猜测时间。问题现象常见原因解决思路sqlite3.OperationalError: database is locked多个进程同时写库或上一个连接未关闭启用 WAL缩短事务范围确认异常分支也执行了 close插入订单报FOREIGN KEY constraint failed客户或服务包 ID 不存在或外键检查未开启先查 customer/package 是否存在确认PRAGMA foreign_keys ON订单列表progress显示 0任务节点全为TODO或模板未复制成功查order_task是否有该订单的数据提交验收一直返回 409存在TODO、DOING或BLOCKED的节点查未完成节点数量逐个推进TypeError: sqlite3.Row object is not subscriptable未转成字典就按 JSON 返回查询结果统一dict(row)接口 500 但日志无信息debugTrue未开或异常被吞打开调试模式查看终端完整堆栈页面刷新后数据不更新浏览器缓存了fetch结果加时间戳参数或设置Cache-Control订单号重复插入失败并发创建时序号撞车依赖UNIQUE约束兜底或改用独立序列表6.2 外键失败的完整排查路径FOREIGN KEY constraint failed是新手最容易卡住的报错之一因为提示信息里没有告诉你哪张表、哪个字段出问题。按下面四步走基本能定位第一步确认报错发生在哪条 SQL。在transaction()外层加except sqlite3.Error as e: print(e, sql)或者打开