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、4 |
| 1 | 已确认(待发货) | 2、4 |
| 2 | 已发货 | 3 |
| 3 | 已完成 | 无 |
| 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 install+requirements.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 命令里显式加--reload：
uvicorn 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 = "mysql+asyncmy://root:password@localhost: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, echo=False, pool_size=5, max_overflow=10, pool_pre_ping=True, pool_recycle=3600, ) SessionLocal = async_sessionmaker(engine, expire_on_commit=False)这里面的pool_pre_ping=True很关键，这是防止 MySQL 连接被 SWITCH 后重启导致连接池里的“僵尸连接”报MySQL server has gone away的一行配置。pool_recycle=3600是告诉 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_no=order_no, product_id=product.id, product_name=product.name, product_price=product.price, qty=qty, total_amount=product.price * qty, buyer_contact=buyer_contact, status=OrderStatus.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而不是float：
from decimal import Decimal from pydantic import BaseModel, condecimal class ConfirmRequest(BaseModel): order_id: int paid_amount: condecimal(max_digits=10, decimal_places=2)同时金额比对不再用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 里，但忘记设置secure=False，在本地 HTTP 环境下 Chrome 强制拦截了非同源 Cookie 请求。后来本地用localhost访问时通常没问题，但用局域网 IP 访问就会偶发。
解决方式：后台页面的 Cookie 设置加了samesite="lax"，并且明确指定secure=False。整体来看，如果只是在本地或者内网使用，这个方案够用；要上公网建议把后台单独放到一个子路径并配置 HTTPS，或者直接用 Basic Auth 也能省不少事。
5.4 MySQL 连接被重置
问题表现：系统跑几天后，接口偶发报Lost connection to MySQL server during query。
原因分析：MySQL 默认wait_timeout是 8 小时，异步连接池里的连接空闲太久会被服务端断开，SQLAlchemy 不知情继续用。
解决方式：连接串加pool_pre_ping=True和pool_recycle=3600（前面代码里已经写了）。这个经验不是 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(minutes=30) 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 打开看一眼“卖了多少单、实收多少、哪些订单状态卡住了”。这个习惯帮我发现了不少漏确认的订单，也算这个系统里最实用的隐藏功能了。
