基于FastAPI与订单状态机的虚拟商品自动发货系统实践-北京尧图网络

1. 项目定位与整体设计思路先说结论这个项目解决的是“没有营业执照、没有企业资质、也不想走第三方支付平台审核”的卖家如何低成本搭建一个能自动发货、能管理订单的虚拟商品交易系统。我做这个系统时最核心的取舍就是不接微信/支付宝官方支付接口而是采用线下转账人工确认的模式。很多人一听“线下支付”就觉得low但实际在圈子里跑过的都知道个人开发者做虚拟资源交易第一步就被卡在支付接口上——要么资质不够要么审核周期长要么交易类目直接被拒。线下支付虽然多了一步人工干预但胜在零门槛、零费率、随时可上线。技术栈选 FastAPI 而不是 Django 或 Flask原因有三FastAPI 原生支持异步虚拟资源交易往往涉及卡密查询、库存扣减、通知回调这些 I/O 操作用 async 能吃到并发红利Pydantic 做参数校验写接口省掉一大半防御性代码接第三方通知时安全性更好把控自动生成 OpenAPI 文档后端接口写完就是现成的联调文档不用单独维护。系统整体按“轻量”来设计没有搞微服务一个 FastAPI 进程 MySQL Redis 就能跑起来。管理后台不单独做前端项目用 Jinja2 模板渲染几个页面就够用核心逻辑全在订单状态机上。1.1 核心需求拆解一个虚拟资源电商系统表面上只需“商品展示下单支付发货”四个环节但落到实际业务需求会拆分得很细商品侧虚拟商品分两类一类是卡密类如软件激活码、会员兑换码需要库存管理卖完自动下架另一类是自动发货类如网盘链接、授权文件不需要库存下单即发。订单侧订单生命周期必须明确从创建、待支付、已确认、已发货到完成每个状态变更都要留痕这是虚拟交易对账的基础。支付侧线下支付的核心是让买家知道“把钱转到哪里、转多少、怎么证明”还要防止买家转账金额填错、漏填订单号导致对不上账。后台侧管理员需要一个订单确认页输入实付金额就能匹配订单、点确认后触发发货。这里最容易出现的问题是人工漏操作所以必须有“待确认”角标提醒。1.2 为什么选择“线下支付后台确认”模式这个模式的关键词其实是“信任兜底”。平台不参与资金托管卖家靠的是买家主动转账买家靠的是卖家收到钱后发货。对个人卖家而言这是唯一能在没有企业资质的情况下跑通交易闭环的方式。从下单到发货整个路径是买家选择商品系统生成一笔“待支付”订单买家看到收款二维码和订单金额线下转账买家回到页面填写“转账单号”或“转账金额尾号”用于对账管理员在后台看到待确认订单核实到账后点击“确认收款”系统自动发货返回卡密或网盘链接并短信/邮件通知买家。这个模式的优势是账目清晰、系统改动小、不需要审核而且售后纠纷处理时有明确的时间节点支付时间、发货时间都在系统里有记录。缺点是每一笔订单多多少少要人工介入但控制在每秒几单的规模内完全够用。2. 数据模型与订单状态机设计数据表的设计决定整个系统能不能跑稳。我第一版草率地把商品和卡密混在一张表里结果上架新商品时总是遇到奇怪的兼容问题最后老老实实拆成五张表用户表、商品表、卡密库存表、订单表、订单状态流水表。2.1 核心表结构拆解商品表主要字段是id、商品名、商品类型0卡密类1自动发货类、价格、是否上架、封面图、商品描述。这里有个重要的细节就是卡密类商品必须冗余一个“库存剩余量”字段不要下单时临时去 count 卡密表否则高并发下会拖垮数据库。卡密库存表字段是id、商品id、卡密内容、状态0未售出1锁定中2已售出、创建时间。加“锁定中”这个状态是必须的因为订单流程里有“买家拍下但还没付完钱”的中间态如果不锁定同一张卡密可能被两个订单同时抢走。订单表核心字段订单号、商品id、商品快照商品名、单价、购买数量、总金额、买家联系方式手机号/邮箱发货用、付款状态、订单状态、支付单号买家填的转账信息、支付时间、发货时间。商品快照字段非常重要——虚拟商品改价、改名是家常便饭如果订单里去 join 商品表拿名称和价格历史订单的数据会被改掉对账时全乱套。订单状态流水表只记录一行订单id、旧状态、新状态、操作人、操作时间。这张表平时用不上但遇到客服纠纷、要查“这单到底怎么变成这个状态的”时候它就是铁证。2.2 订单状态机设计订单状态是整个系统的中枢我设计了五个状态状态值含义可流转到0待支付1、41已确认(待发货)2、42已发货33已完成无4已取消无流程图不画了描述一下关键流转逻辑买家下单创建一笔状态为 0 的订单管理员在后台确认收款后状态从 0 变为 1系统执行发货动作扣库存、发卡密后状态从 1 变为 2买家点击“确认收货”虚拟资源一般不需要但保留这个动作可以自动完结订单后状态从 2 变为 3。状态为 0 的订单超过 30 分钟未支付系统自动置为 4# 订单状态机核心逻辑 from enum import IntEnum class OrderStatus(IntEnum): PENDING 0 # 待支付 CONFIRMED 1 # 已确认待发货 SHIPPED 2 # 已发货 COMPLETED 3 # 已完成 CANCELLED 4 # 已取消 VALID_TRANSITIONS { OrderStatus.PENDING: {OrderStatus.CONFIRMED, OrderStatus.CANCELLED}, OrderStatus.CONFIRMED: {OrderStatus.SHIPPED, OrderStatus.CANCELLED}, OrderStatus.SHIPPED: {OrderStatus.COMPLETED}, OrderStatus.COMPLETED: set(), OrderStatus.CANCELLED: set(), } def transition_order(order, new_status): if new_status not in VALID_TRANSITIONS[order.status]: raise ValueError(f非法状态流转: {order.status} - {new_status}) old_status order.status order.status new_status # 记录流水 log_order_status(order.id, old_status, new_status)这里有个实战心得状态流转一定要做合法性校验不要开发时图省事直接 update后面线上出问题会非常难排查。快递发货可以回退但虚拟资源一旦发出你没法把发给买家的网盘链接收回来所以“已发货”这个状态一旦进入就不允许回退。3. FastAPI 工程搭建与核心接口实现3.1 项目环境准备这里重点讲环境搭建因为很多人在 FastAPI 入门阶段就卡在“装不上”和“不热更新”两个问题上。我在 Windows 上做开发时用过原生环境后来换成 uv 管理体验差别巨大。# 使用 uv 创建虚拟环境并安装 FastAPI推荐 uv init fastapi-shop cd fastapi-shop uv add fastapi uvicorn[standard] sqlalchemy asyncmy alembic pydantic-settings redis为什么推荐 uv因为uv add会自动创建.venv并在项目根目录生成pyproject.toml每次新增依赖都自动锁版本。相比pip installrequirements.txt的方式uv 的解析速度快至少一个数量级而且不会动不动就把系统全局环境搞脏。在 PyCharm 里打开项目时解释器直接选.venv下的 python.exe 即可不需要手动source。关于 PyCharm 安装 FastAPI 失败的问题绝大多数情况是默认的 PyPI 源在国内访问超时。解决办法是给 uv 配置国内镜像# 配置 uv 使用清华镜像源 uv add fastapi --default-index https://pypi.tuna.tsinghua.edu.cn/simple启动热更新问题一定要在 uvicorn 命令里显式加--reloaduvicorn app.main:app --host 0.0.0.0 --port 8000 --reload如果加了--reload还不管用先确认项目文件是通过 Open 打开映射到本地文件系统的而不是通过 SSH 远程挂载再确认启动目录和 app 导入路径一致。我遇到一次热更新失效是因为把app.main:app写成了main:app导致启动目录不对文件变更监听失效。3.2 项目目录结构与数据库初始化我习惯按功能模块分目录而不是传统 Django 的按 app 分fastapi-shop/ ├── app/ │ ├── main.py # 应用入口 │ ├── config.py # 配置管理 │ ├── models.py # SQLAlchemy 模型 │ ├── schemas.py # Pydantic 模型 │ ├── api/ │ │ ├── product.py # 商品接口 │ │ ├── order.py # 订单接口 │ │ └── admin.py # 后台管理接口 │ ├── services/ │ │ ├── order_service.py # 订单业务逻辑 │ │ └── deliver_service.py # 发货逻辑 │ └── templates/ # Jinja2 模板 ├── migrations/ # alembic 迁移脚本 └── pyproject.toml数据库连接用异步 SQLAlchemy这里提一个异步连接池的参数问题。很多人用create_async_engine时报 “TimeoutError”多半是没配连接池大小。虚拟交易系统单机并发量不大连接池配置保守一点不出问题# config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str mysqlasyncmy://root:passwordlocalhost:3306/fastapi_shop redis_url: str redis://localhost:6379/0 admin_token: str your-secret-token order_expire_minutes: int 30 class Config: env_file .env settings Settings() # database.py from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker engine create_async_engine( settings.database_url, echoFalse, pool_size5, max_overflow10, pool_pre_pingTrue, pool_recycle3600, ) SessionLocal async_sessionmaker(engine, expire_on_commitFalse)这里面的pool_pre_pingTrue很关键这是防止 MySQL 连接被 SWITCH 后重启导致连接池里的“僵尸连接”报MySQL server has gone away的一行配置。pool_recycle3600是告诉 SQLAlchemy 一小时强制回收一次连接避免 sleep 超时被服务端杀掉。3.3 商品与下单接口实现商品接口没什么特别的主要是加了缓存避免每次访问都查数据库。缓存用 Redis 存商品详情 JSON缓存时间设 60 秒。更新商品信息后要手动 delete 缓存否则前台展示滞后。下单接口是整个系统的重点。卡密类商品的下单要加三步事务操作检查订单表中是否已有相同“商品id 买家标识”的待支付订单如果有直接复用旧订单避免重复下单占用库存锁定卡密库存——把对应商品下状态为“未售出”的卡密标记为“锁定中”并记录到订单上创建订单状态为待支付。# services/order_service.py async def create_order(db: AsyncSession, product_id: int, buyer_contact: str, qty: int 1): # 查询商品 product await db.get(Product, product_id) if not product or not product.is_active: raise HTTPException(404, 商品不存在或已下架) # 幂等检查同一买家同一商品存在待支付订单则复用 existing await db.execute( select(Order).where( Order.product_id product_id, Order.buyer_contact buyer_contact, Order.status OrderStatus.PENDING ) ) if existing.scalar_one_or_none(): return existing.scalar_one() # 卡密类商品需要锁定库存 if product.type PRODUCT_TYPE_CARD: cards await lock_card_stock(db, product_id, qty) if len(cards) qty: await db.rollback() raise HTTPException(400, 库存不足) else: cards [] # 创建订单 order_no generate_order_no() order Order( order_noorder_no, product_idproduct.id, product_nameproduct.name, product_priceproduct.price, qtyqty, total_amountproduct.price * qty, buyer_contactbuyer_contact, statusOrderStatus.PENDING, ) db.add(order) await db.commit() await db.refresh(order) return order注意生成订单号的逻辑不要用数据库自增 id 直接当订单号给买家看容易暴露销量。我用时间戳 6位随机数字拼接重复概率极低且在并发下也能保证唯一。3.4 支付对接页面实现线下支付需要一个页面告诉买家“转多少钱、转到哪、怎么填单”。这里最容易出问题的不是代码而是“买家不知道要填什么”。我在下单成功页展示三个信息并用醒目字体提醒收款方式支付宝/微信收款码图片存静态目录或 OSS 收款账户名应付金额精确到分比如 ¥19.90转账填单规范转账时备注“订单号后四位商品简称”如“3912 卡密”然后回到页面填入“转账单号交易/流水号”提交。页面填单后会调到POST /api/order/{order_no}/pay-proof接口把买家的转账单号入库更新pay_proof字段。这一步相当重要后台确认时多了一个数字可以比对。# schemas.py class PayProofSubmit(BaseModel): order_no: str transaction_no: str # 买家转账的交易流水号 router.post(/order/{order_no}/pay-proof) async def submit_pay_proof(order_no: str, payload: PayProofSubmit, db: AsyncSession Depends(get_db)): order await query_order_by_no(db, order_no) if not order or order.status ! OrderStatus.PENDING: raise HTTPException(400, 订单不存在或状态不允许) order.transaction_no payload.transaction_no order.pay_proof_submitted_at datetime.now() await db.commit() return {code: 0, msg: 已提交等待管理员确认}4. 后台确认机制与自动发货实现后台确认是整个系统的操作核心。管理员每天要面对很多订单所以确认页面的设计要“一屏搞定”列表里只展示等待确认的订单每条显示订单号、商品名、金额、买家填写的转账信息、下单时间。管理员核对后点按钮系统弹出金额比对确认框避免手滑把 90 看成 900。4.1 后台确认接口确认接口做三件事校验订单状态必须是“待支付”防止重复确认校验管理员传入的实付金额与订单金额是否一致误差在 1 元内允许通过防止虚拟商品满减之类的特殊情况更新订单状态为“已确认”并触发发货任务。注意一定要用数据库行锁。两个管理员同时打开同一个订单都点确认如果没有行锁会触发两次发货虽然状态机校验了状态但并发场景下先读后写可能出现竞态。# services/deliver_service.py async def confirm_order(db: AsyncSession, order_id: int, paid_amount: float): # 使用 SELECT FOR UPDATE 锁定订单行 result await db.execute( select(Order).where(Order.id order_id).with_for_update() ) order result.scalar_one_or_none() if not order: raise HTTPException(404, 订单不存在) if order.status ! OrderStatus.PENDING: raise HTTPException(400, 订单状态不允许确认) # 金额比对误差1元内 if abs(paid_amount - order.total_amount) 1: raise HTTPException(400, f实付金额与订单金额不符: {paid_amount}) order.status OrderStatus.CONFIRMED order.paid_amount paid_amount order.confirmed_at datetime.now() await db.commit() # 触发异步发货 asyncio.create_task(deliver_order(order.id)) return {code: 0, msg: 确认成功开始发货}4.2 自动发货逻辑发货任务根据商品类型走两个分支。卡密类商品发货最简单从卡密表里把“锁定中”且属于当前订单的卡密查出来标记为“已售出”然后组装内容发给买家。自动发货类商品更简单把商品详情里的发货内容网盘链接提取码/授权文件URL直接发给买家。实际场景中发货后要发短信或邮件通知买家取货。这里牵扯到第三方短信/邮件服务如果没接备选方案是在订单详情页展示“卡密信息”区块发货后订单详情页直接可见。这也是很多个人虚拟交易系统采用的方案——不需要额外触达买家回到页面或查订单列表就能看到发货内容。# services/deliver_service.py async def deliver_order(order_id: int): async with SessionLocal() as db: order await db.get(Order, order_id) product await db.get(Product, order.product_id) if product.type PRODUCT_TYPE_CARD: # 查询本单锁定的卡密 result await db.execute( select(Card).where( Card.order_id order.id, Card.status CARD_STATUS_LOCKED ) ) cards result.scalars().all() if not cards: # 理论不会出现但兜底 await notify_admin(order.id, 订单无锁定卡密需人工处理) return for card in cards: card.status CARD_STATUS_SOLD card.sold_at datetime.now() cards_content \n.join([c.content for c in cards]) order.deliver_content cards_content else: # 自动发货类发货内容在商品表冗余 order.deliver_content product.auto_deliver_content order.status OrderStatus.SHIPPED order.shipped_at datetime.now() await db.commit()这里有个我在生产环境踩过的大坑不能直接用asyncio.create_task在请求里开异步任务。FastAPI 的请求处理完事件循环没退出任务一般能跑完但如果任务里发生未捕获异常整个请求会连带报错。更稳的做法是用 Celery 或 Arq 做任务队列不过这个系统追求轻量我用了asyncio.create_task并包了一层 try-except 确保不泄漏同时也接受它“不保证任务一定执行”的缺点——因为发货操作在后面有补偿机制订单状态卡在已确认但未发货时后台可以手动点击“重新发货”。4.3 后台管理页面体验优化后台我不建议直接用 FastAPI 的接口裸奔管理员每天看那么多单子体验很重要。我用 Jinja2 渲染了三个页面待确认订单页只展示状态为“待支付且已提交支付凭证”的订单并在导航栏用红点显示待确认数量订单管理页按状态筛选支持按订单号/买家联系方式搜索点开有完整时间线和状态流水发货异常页专门展示“已确认但未发货超过10分钟”的订单方便人工补发货。页面接口的鉴权不能和买家接口共用我在 FastAPI 里用了一个简单的 Bearer Token 中间件后台接口要求请求头带X-Admin-Token比对配置里的admin_token。登录页面是静态 HTML 表单提交后跳转接口设置 Cookie。这个方案不搞密码哈希、不做多用户但对单人运营卖家来说足够。5. 常见问题与排查技巧实录这部分写在文档里也查不到都是实打实跑出来的坑。5.1 状态流转与并发问题问题表现两个管理员同时点“确认收款”系统发了两遍卡密。原因分析两个请求同时读到了订单的“待支付”状态都执行了下发逻辑。第一版我只在代码 if 里判断状态数据库层面没有行锁所以并发击穿了校验。解决方式在确认订单和发货逻辑里都使用SELECT ... FOR UPDATE锁定订单行。另外在卡密库存表里维护了一个“已售出”状态一张卡密被第二笔订单查到锁定状态时直接报错返回作为兜底防线。5.2 支付金额比对的小数精度问题问题表现管理员在后台确认一笔 19.90 元的订单输入 19.9系统提示“实付金额与订单金额不符”。原因分析Python 的 float 计算有精度问题19.9 0.1 不等于 20。订单金额我在数据库存的是 Decimal传到前端后可能被 JSON 序列化成 float再原样返回就产生了误差。解决方式金额字段统一用 Decimal 类型API 层用 Pydantic 校验时用Decimal而不是floatfrom decimal import Decimal from pydantic import BaseModel, condecimal class ConfirmRequest(BaseModel): order_id: int paid_amount: condecimal(max_digits10, decimal_places2)同时金额比对不再用abs(a-b) 1而是改成if abs(Decimal(str(paid_amount)) - order.total_amount) Decimal(1.00): raise HTTPException(400, 金额偏差过大)5.3 后台页面登录失效与 Cookie 问题问题表现管理员浏览器访问后台页面偶尔会出现“登录状态丢失”刷新后又能访问。原因分析我用 JWT Token 存在 Cookie 里但忘记设置secureFalse在本地 HTTP 环境下 Chrome 强制拦截了非同源 Cookie 请求。后来本地用localhost访问时通常没问题但用局域网 IP 访问就会偶发。解决方式后台页面的 Cookie 设置加了samesitelax并且明确指定secureFalse。整体来看如果只是在本地或者内网使用这个方案够用要上公网建议把后台单独放到一个子路径并配置 HTTPS或者直接用 Basic Auth 也能省不少事。5.4 MySQL 连接被重置问题表现系统跑几天后接口偶发报Lost connection to MySQL server during query。原因分析MySQL 默认wait_timeout是 8 小时异步连接池里的连接空闲太久会被服务端断开SQLAlchemy 不知情继续用。解决方式连接串加pool_pre_pingTrue和pool_recycle3600前面代码里已经写了。这个经验不是 FastAPI 特有的任何用连接池访问 MySQL 的语言都会遇到换 Django、Flask 也一样。5.5 定时任务清理过期订单线下支付的人工确认通常是隔一段时间看一次如果买家拍下后不付钱待支付订单一直占着库存对卡密类商品影响较大。我加了个定时任务每 5 分钟扫一次状态为“待支付”且创建时间超过 30 分钟的订单把它们置为“已取消”并把锁定的卡密释放回“未售出”状态。# tasks.py 使用 asyncio.create_task 实现简单定时任务 import asyncio from datetime import datetime, timedelta async def cancel_expired_orders(): while True: await asyncio.sleep(300) async with SessionLocal() as db: now datetime.now() expire_time now - timedelta(minutes30) result await db.execute( select(Order).where( Order.status OrderStatus.PENDING, Order.created_at expire_time ) ) expired_orders result.scalars().all() for order in expired_orders: order.status OrderStatus.CANCELLED # 释放锁定的卡密 await release_cards_by_order(db, order.id) await db.commit()这个任务的注册方式是在main.py的 startup 事件里启动asynccontextmanager async def lifespan(app: FastAPI): asyncio.create_task(cancel_expired_orders()) yield定时任务不能像 Celery 那样持久化、保证精确执行但对这个系统场景30 分钟才执行一次清理完全够用。6. 安全与体验细节补充虚拟交易系统最怕两件事订单信息被篡改、后台被人撞库。下单接口我补充了两个安全措施一是创建订单时校验商品状态和上下架状态二是提交支付凭证的接口加了简单的频率限制同一 IP 一分钟最多提交 10 次防止被脚本刷接口。商品价格这个字段在前端下单请求里绝对不能传。服务端创建订单时只从数据库读价格不接受任何前端传参的price、amount字段。这是新手最容易犯的错误——前端表单带了个价格字段后端不校验直接用结果买家改一下 POST 请求体就能 1 分钱下单。我的做法是在 Pydantic 模型里直接不定义这个字段请求体里出现未知字段默认被丢弃。另外关于买家的联系方式存储我做了掩码展示后台列表只显示中间四位脱敏的邮箱/手机号点进详情才能看到完整联系方式防止运营页面被截图外泄带来的信息泄露虽然是自己用但养成这个习惯没坏处。写在最后的一点体会跑这个系统的几个月里最大的感触是轻量级系统不等于功能可以糊弄尤其涉及钱和货的流转状态机、库存锁定、定时清理这些“看起来麻烦”的设计恰恰是整个系统的地基。你永远不知道哪一笔订单会出纠纷但状态流水表和数据一致性能让你在混乱里迅速定位问题。最后再分享一个小技巧后台页面我加了一个“导出当日订单 CSV”的按钮用的是 FastAPI StreamingResponse几行代码的事情。表面上是多了一个导出功能实际价值是每天晚上对账的时候能拿 Excel 打开看一眼“卖了多少单、实收多少、哪些订单状态卡住了”。这个习惯帮我发现了不少漏确认的订单也算这个系统里最实用的隐藏功能了。

基于FastAPI与订单状态机的虚拟商品自动发货系统实践

相关推荐

Ventoy启动盘无法引导？3步排查修复，保留ISO文件的更新方法

基于均值-方差模型的资产配置：Python实现有效前沿与协方差估计

G-Helper：华硕笔记本性能模式、风扇调校与 GPU 切换，一个 exe 搞定

更多资讯

CLI-Anything 3MF 网格几何编辑器：圆柱孔检测、缩放与 3D 打印文件修复实战指南

Python包标准化实践：WeClaw框架发布全流程

SystemInformer 多语言切换：3 分钟看懂界面语言，附汉化教程

ESP32-P4 USB开发实战：从协议原理到枚举排坑

光谱样机开发选型指南：成品光谱仪、模块还是定制光路？

Ollama 无法使用 GPU 的完整排查思路：从驱动到模型加载全链路

今日更新

Linux 内核 ARM 虚拟内存布局全解析：从 4GB 地址空间划分到源码级验证

AIGC降重工具测评与教育应用指南

数据容灾核心指标与实战方案解析

本周精选

超人会飞不算本事：系统稳定依赖清晰规则与边界设计

超人VS蜘蛛侠：拆解超级IP的影响力与传播方法论

基于CNN的调制信号识别：MATLAB实现时频图分类实战

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？