Flask进销存系统实战：核心表结构、库存流水与并发控制

作为一个常年折腾后端业务系统的人，我接到过不少类似的单子：Python + Flask + 进销存。这看起来是个老生常谈的组合，但真正动手做的时候，才发现坑全藏在细节里。这次分享的项目是一个给某精品品牌企业用的进销存系统，品牌方要求叫它Gucci进销存，但实际业务逻辑就是标准的多门店零售进销存——采购入库、门店销售、库存调拨、盘点报损、财务对账一套闭环。我用Flask作为主框架，配合SQLAlchemy做ORM，前端用Jinja2模板加一点 Bootstrap，把一整套系统从零搭了起来。

这几年陆陆续续帮人做了很多个进销存项目，有做服装的、做化妆品代理的、做食品经销的，每一个的业务细节都不一样，但底层模型基本一致。如果你正准备用Flask写进销存系统，或者你在用Django、Spring写但想看看别人怎么设计库存流水，这篇文章应该能给你省不少踩坑时间。我会把表结构设计、事务与锁的细节、权限控制、以及我实际排障时遇到的那些“幽灵库存”问题的处理思路，全部按项目实战的节奏捋一遍。

1. 项目定位与方案选型：进销存系统最怕的不是写不出功能，而是理不清流程

很多人一听说进销存，第一反应就是“不就是CRUD吗”。说实话，如果只是把商品、供应商、订单、库存这几张表建好，然后做增删改查，那确实不难，三天就能写完。但你真放到企业里跑起来，问题就全出来了：仓库和门店的库存怎么分？调拨在途库存算谁的？销售退货之后入库单怎么关联？采购入库时供应商送来的货和单据对不上怎么办？这些都要求你在编码之前，先把业务流程梳理清楚。

1.1 进销存到底解决什么问题

进销存这三个字，拆开就是“进、销、存”，对应的分别是采购入库、销售出库、库存管理。听起来简单，但企业每天真正的业务是这三条链路交叉进行的——采购部下了采购单，仓库收货入库，门店开销售单出货，仓管发现某款库存少了要做盘点报损，财务月底要对账核算毛利。任何一环断了，账就对不上。

就拿这个项目来说，客户是品牌零售企业，多门店、多仓库、统一采购、总部管控。他们要解决的核心痛点有三个：

第一，库存账实不符。以前靠Excel，每次盘点都有一堆差异，还说不清是哪笔单子造成的。第二，销售与采购脱节。总仓不知道门店到底卖了多少，补货全靠感觉。第三，财务对账困难。采购单、销售单、退货单散落在不同表格里，月底核对费劲。

所以这个系统的本质，不是功能多丰富，而是每一笔库存变动都有据可查，账实永远能对上。这个“账实相符”才是进销存系统真正的灵魂。

1.2 为什么选Flask而不是Django或其他框架

我见过不少团队一上来就用Django，因为它自带Admin后台，看起来“开箱即用”。但实际做进销存这种偏业务系统的项目，我反而更推荐Flask。原因有三点：

一是Flask足够轻，学习成本低。整个系统就是一堆视图函数加几个模型类，没有Django那种App级别的强制目录结构，业务逻辑不复杂时写起来更顺手。

二是灵活可控。进销存系统里有很多自定义业务规则，比如“库存不足时是否允许负库存销售”、“调拨单在途库存怎么锁定”，这些逻辑如果用Django的admin，还得绕过它的默认行为，折腾起来反而不如Flask里自己写视图函数直接。

三是部署轻量。很多企业的进销存系统就是内网部署，一台小服务器甚至一台PC就能跑。Flask应用用gunicorn或者uwsgi带起来，配个Nginx反代，非常轻便。

还有一点是生态。Flask配SQLAlchemy是全球Python圈最常见的组合，遇到问题Stack Overflow上基本都有答案，社区资料非常全，比用一些冷门Web框架稳妥得多。

1.3 技术栈清单与工程模块划分

这个项目的技术栈如下，都是比较经典的组合：

• 后端框架：Flask 2.x
• ORM：Flask-SQLAlchemy（基于SQLAlchemy 2.x）
• 数据库：MySQL 5.7+（生产环境），开发环境用的SQLite
• 前端：Jinja2模板 + Bootstrap 5 + 少量原生JavaScript
• 登录与权限：Flask-Session + 自定义用户角色装饰器
• 其他：PyMySQL驱动、Flask-Migrate做数据库迁移、python-dotenv管理配置

模块划分我采用蓝图（Blueprint）机制，按业务域拆成几个模块：

• auth：登录、登出、修改密码
• product：商品管理、品牌分类、条码管理
• purchase：采购单、入库单、供应商管理
• sales：销售单、退货单、客户管理
• stock：库存查询、盘点、调拨、报损
• report：进销存报表、毛利统计、库存预警
• system：用户管理、角色权限、操作日志

每个蓝图就是一个独立目录，里面有views.py（路由）、models.py（模型）、services.py（业务逻辑）、templates/（模板）。这样做得好处是，后边无论是加功能还是排查问题，都能快速定位到对应模块，而不是在一堆堆叠的代码里找。

2. 核心业务拆解与数据库建模：所有坑都在表结构设计阶段埋下的

如果说代码是进销存系统的骨架，那数据库表就是血脉。表设计得不好，后面写再多代码都是打补丁。我见过太多项目，一开始不建库存流水表，等系统上线才发现库存对不上，然后再回头补流水，那就是一场灾难。

2.1 三条核心业务链路

我把进销存系统的业务流程拆成三条主链，所有功能都是围绕这三条链展开的。

采购链路：采购员创建采购单 → 审批（小企业可省略）→ 仓库收货 → 生成入库单 → 库存增加 → 供应商应付账款增加。

销售链路：导购开销售单 → 选择商品 → 计算金额 → 库存扣减 → 生成出库记录 → 客户应收账款增加。

库存调整链路：盘点（盈/亏）、报损、调拨（总仓→门店、门店→门店）、退供应商。每一笔都要求产生库存变动记录。

这三条链路互相交错，比如销售退货会重新走一遍入库逻辑，采购退货则要冲减库存。所以我在设计表结构时，特意让所有库存变动都落在同一张流水表上，这样无论哪条链路发起的变动，最终都能统一对账。

2.2 核心表结构设计

我把核心表列出来，并附上关键字段，这样你对整体模型会有直观认识。

这里有几个设计细节值得细说。

商品表：我要求SKU必须是全局唯一的，而且不能删除，只能下架（status字段控制）。原因很简单，历史单据里引用过这个SKU，如果把商品真删了，所有历史单据都会变成脏数据。很多新手在这个问题上踩坑，被财务问“上个月这张单的商品叫什么”的时候才发现商品没了。

单据表：所有单据都保留status字段，包括草稿、待审核、已确认、已作废等状态。进销存单据是财务做账的依据，一旦确认就不允许直接修改，只能“作废”并生成一笔反向调整单。这是进销存系统的底线逻辑。

冗余设计：在单据明细表里，我故意冗余了price、product_name等字段，而不是靠联表去查。因为商品价格是会变的，单据上记录的是交易发生那一刻的价格，如果只存product_id，日后商品调价，历史单据的价格也会被动改变，这绝对是财务不能接受的。

2.3 库存流水表：进销存系统的“账本”

我始终觉得，一个进销存系统做得好不好，就看一张表——stock_movements，库存流水表。这张表是整个系统的账本，记录每一次库存变动的来龙去脉。

每次库存变动，不管是入库、出库、盘点、调拨还是报损，都必须插入一条流水记录，写入变动的商品、门店、变动类型、变动数量、当时变动前后的库存结余，以及关联的单据号。字段balance_before和balance_after特别重要，它们是审计对账的核心依据。

举个例子，你查到某个SKU现在库存是50，但你觉得不对，就可以直接查流水表：

SELECT ref_type, ref_no, quantity, balance_before, balance_after, created_at, operator_id FROM stock_movements WHERE product_id = 'SKU001' ORDER BY created_at ASC;

这样就能看到这个SKU从第一次入库到现在每一次变动的明细，到底是哪一单、谁操作的、变动前后余量是多少，一目了然。没有这张表，库存对不上时你只能去翻一大堆单据，那基本等于大海捞针。

我还给流水表加了一个type字段，用枚举表示变动类型：INBOUND（入库）、SALE（销售出库）、SALE_RETURN（销售退货）、PURCHASE_RETURN（采购退货）、CHECK_IN（盘盈）、CHECK_OUT（盘亏）、TRANSFER_OUT（调拨出）、TRANSFER_IN（调拨入）、BREAKAGE（报损）。有了类型字段，报表统计时想算“这个月总入库多少”或者“这个月损耗率多高”都特别方便。

3. 核心实现与实操过程：从建表到跑通一条完整入库流程

这块我挑几个关键环节来讲，重点在代码实现和事务处理。

3.1 项目初始化和数据库准备

我先用Flask-SQLAlchemy完成数据库配置，顺便加上简单的Session管理。配置代码一般长这样：

from flask import Flask from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate import os app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', 'mysql+pymysql://root:password@localhost/gucci_erp?charset=utf8mb4') app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'dev-secret-change-me') app.config['SESSION_TYPE'] = 'filesystem' db = SQLAlchemy(app) migrate = Migrate(app, db)

然后定义User模型，带角色字段：

class User(db.Model): __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(64), unique=True, nullable=False, index=True) password_hash = db.Column(db.String(255), nullable=False) role = db.Column(db.String(32), nullable=False, default='clerk') # admin, purchaser, clerk, finance, store_manager store_id = db.Column(db.Integer, db.ForeignKey('stores.id')) status = db.Column(db.SmallInteger, default=1, comment='1-正常 0-停用')

创建完模型以后，用Flask-Migrate执行迁移：

flask db init flask db migrate -m "init tables" flask db upgrade

再用一个种子脚本创建管理员账号和初始门店、示例商品，方便后续测试。这些准备做完，就可以开始写核心业务接口了。

3.2 商品入库功能的实现与事务控制

入库流程是进销存系统里最核心的写操作之一。它的本质是：在事务里同时插入入库单、入库明细、更新库存、插入库存流水。如果这四步有任何一步失败，其他步骤必须全部回滚，否则账就乱了。

我用一个完整的视图函数来展示这个逻辑：

from flask import Blueprint, request, redirect, url_for, flash from extensions import db from models import PurchaseOrder, PurchaseOrderItem, InboundRecord, Product, Stock, StockMovement from utils import get_current_user, require_role purchase_bp = Blueprint('purchase', __name__) @purchase_bp.route('/inbound/confirm/<int:order_id>', methods=['POST']) @require_role('purchaser', 'admin') def confirm_inbound(order_id): order = PurchaseOrder.query.get_or_404(order_id) if order.status != 'approved': flash('当前采购单状态不允许入库确认', 'danger') return redirect(url_for('purchase.detail', order_id=order.id)) # 开启事务 try: for item in order.items: product = Product.query.filter_by(id=item.product_id).with_for_update().first() if not product: raise Exception(f'商品ID {item.product_id} 不存在') # 判断入库目标仓库 store_id = request.form.get('store_id', type=int) if not store_id: raise Exception('请选择入库仓库') stock = Stock.query.filter_by( product_id=product.id, store_id=store_id ).with_for_update().first() if not stock: stock = Stock(product_id=product.id, store_id=store_id, quantity=0, locked_quantity=0) db.session.add(stock) before_quantity = stock.quantity stock.quantity += item.quantity movement = StockMovement( product_id=product.id, store_id=store_id, type='INBOUND', quantity=item.quantity, ref_type='PurchaseOrder', ref_no=order.order_no, balance_before=before_quantity, balance_after=stock.quantity, operator_id=get_current_user().id ) db.session.add(movement) inbound = InboundRecord( order_id=order.id, product_id=product.id, store_id=store_id, quantity=item.quantity, batch_no=f'{order.order_no}-{product.id}', operator_id=get_current_user().id ) db.session.add(inbound) order.status = 'inbound' db.session.commit() flash('入库成功，库存与流水已更新', 'success') except Exception as e: db.session.rollback() flash(f'入库失败：{str(e)}', 'danger') return redirect(url_for('purchase.detail', order_id=order.id))

这段代码里最关键的两行是with_for_update()。它会锁定对应商品和库存行，防止并发情况下两个人同时对同一个SKU做入库操作导致库存被覆盖。做进销存系统，凡是涉及库存数字变动的写操作，一律要加行锁，这是底线。不加锁，后边超卖、库存负数这些问题迟早会找上门。

入库单确认后，采购单状态从approved变为inbound，表示货已入库。如果某张采购单是分批到货的，就需要给采购单增加“已入库数量”字段，每确认一批就累加一次，直到全部到货。

3.3 销售出库与库存扣减：防超卖的关键处理

销售出库的逻辑跟入库基本对称，但有一个著名的坑：超卖。也就是两个门店同时卖同一个SKU，库存只剩1件，两个订单都扣成了0，最后库存变成负数。

超卖的根源在于“先查库存，再扣库存”这两个动作不是原子的，中间有其他请求插了队。解决办法是：在扣库存时用行锁强制串行化。下面是我实际用的销售扣库存代码：

from flask import Blueprint, request, render_template from extensions import db from models import SalesOrder, SalesOrderItem, Stock, StockMovement sales_bp = Blueprint('sales', __name__) @sales_bp.route('/sales/create', methods=['POST']) def create_sale(): store_id = request.form.get('store_id', type=int) items = request.form.getlist('items') # 格式: product_id:quantity if not store_id or not items: flash('缺少门店或商品信息', 'danger') return redirect(url_for('sales.index')) try: total = 0 order_no = generate_sales_order_no() for row in items: product_id, quantity = row.split(':') quantity = int(quantity) product = Product.query.get(int(product_id)) # 关键：加行锁，避免并发超卖 stock = Stock.query.filter_by( product_id=product.id, store_id=store_id ).with_for_update().first() if not stock or stock.quantity < quantity: raise Exception(f'商品 {product.name} 库存不足') before_quantity = stock.quantity stock.quantity -= quantity movement = StockMovement( product_id=product.id, store_id=store_id, type='SALE', quantity=quantity, ref_type='SalesOrder', ref_no=order_no, balance_before=before_quantity, balance_after=stock.quantity, operator_id=get_current_user().id ) db.session.add(movement) item_total = product.sale_price * quantity total += item_total so_item = SalesOrderItem( product_id=product.id, quantity=quantity, price=product.sale_price, amount=item_total ) db.session.add(so_item) sale_order = SalesOrder( order_no=order_no, store_id=store_id, total_amount=total, status='paid', operator_id=get_current_user().id ) db.session.add(sale_order) # 注意：如果要保存明细，需要用 relationship 联动，这里简化只做示意 db.session.commit() flash(f'销售单 {order_no} 创建成功，金额 {total}', 'success') except Exception as e: db.session.rollback() flash(f'销售失败：{str(e)}', 'danger') return redirect(url_for('sales.index'))

这段逻辑里还有几个业务细节值得提一下：

一是订单号生成。我用的规则是SALE + 日期 + 序号，比如SALE2025061700001。订单号是财务和客服找人查单的重要凭据，不可重复，最好加唯一索引。

二是价格取数。销售价格从商品表带出，但实际业务里会有会员折扣、活动优惠，所以我在销售单明细里存了单价和实收价两个字段，这样对账的时候能区分“标价总额”和“实收金额”。

三是库存不足直接抛出异常，而不是装看不见继续扣。有些老业务员喜欢说“先开了单，库存后天补”，真按这种做法搞，月底对账绝对让你怀疑人生。我的建议是：系统强制不允许负库存，实在要支持负库存，也要加一个配置项让管理者显式打开，并且报表里把负库存商品标红。

3.4 权限系统与操作日志：进销存系统不能省的部分

进销存系统里，权限不是花架子，它直接关系到财务数据的安全。我按角色做了四级权限：

实现方式是在蓝图路由上套一个装饰器：

from functools import wraps from flask import session, redirect, url_for, flash def require_role(*roles): def decorator(f): @wraps(f) def wrapped(*args, **kwargs): user_role = session.get('role') if user_role not in roles: flash('权限不足', 'danger') return redirect(url_for('auth.login')) return f(*args, **kwargs) return wrapped return decorator # 用法示例 @app.route('/admin/users') @require_role('admin') def user_list(): ...

操作日志也不能省。我建了一张operation_logs表，记录谁在什么时间操作了哪张单据、动作是什么（创建、修改、作废、确认）。一开始我也觉得加日志麻烦，直到有一次客户来问“上个月这张采购单是谁改的价格”，我才意识到这功能有多刚需。

class OperationLog(db.Model): __tablename__ = 'operation_logs' id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, nullable=False) username = db.Column(db.String(64), nullable=False) action = db.Column(db.String(64), nullable=False, comment='create/update/cancel/confirm') target_type = db.Column(db.String(32), nullable=False, comment='PurchaseOrder/SalesOrder/Stocktake...') target_no = db.Column(db.String(64), nullable=False) description = db.Column(db.Text) created_at = db.Column(db.DateTime, default=datetime.now)

写日志的动作放在业务操作的事务里边，跟主操作同生共死。这样日志记录不会因为后续崩溃而丢失，审计才完整。

4. 常见问题排查与性能优化实录

进销存系统上线之后，运维才是真正的考验。我把自己这段时间实际遇到过的典型问题整理出来，希望能帮你避开同样的坑。

4.1 并发售卖导致库存出现负数

刚上线那会儿，门店反馈说有个爆款SKU库存变成-3了。我第一反应是不可能，因为代码里明明做了库存不足校验。排查后发现，负库存只会出现在并发请求下：两个收银员同时在两台POS上卖这个SKU，都读到库存还剩3件，各自扣掉2件，最后库存变成-1。

逻辑上我是在create_sale里先查库存再判断再扣减，但如果多个请求同时走到那一步，数据库默认的读提交隔离级别下就会读到同一份旧数据。解决办法就是我前面说的，查询库存时加with_for_update()行锁，让同一SKU在同一仓库的扣减操作串行执行。加了锁以后再拿测试脚本并发刷了几十次，库存始终正确，没有负数。

提示：凡是写库存操作，必须用with_for_update()锁库存行。如果你用的是Django ORM，对应的是select_for_update()，效果一样。

4.2 库存报表越跑越慢

系统跑了三个月之后，stock_movements表涨到了十万行，进销存报表接口开始变慢，一次查询要三四秒。我检查了SQL执行计划，发现查询条件product_id和store_id都有索引，但组合查询时索引没有生效，走了全表扫描。

解决办法是加了一个复合索引：

ALTER TABLE stock_movements ADD INDEX idx_stock_query (product_id, store_id, created_at);

加了索引之后，报表查询时间从3秒降到0.3秒以内。另外我还给“每日库存汇总”做了一个汇总表，每天凌晨跑脚本把当天的入库、出库、结存数算好，白天的查询只读汇总表，不再实时扫描流水。这个思路在数据量继续增长以后依旧能撑住。

4.3 上线后Flask Session时不时失效

有段时间门店导购反映，登录状态动不动就掉，一天要重新登录好几次。排查发现是因为Session存储用的是默认的filesystem，多进程部署时每个进程保存的Session文件不一致，导致请求落在不同进程上时就判断成未登录。

后来我把Session存储从文件系统改成了Redis：

from flask_session import Session app.config['SESSION_TYPE'] = 'redis' app.config['SESSION_REDIS'] = redis.from_url('redis://localhost:6379/0') Session(app)

改完这个再没出现掉登录的问题。如果你的系统规模不大、单进程跑，文件系统Session问题不大；但只要用gunicorn起了多个worker，Session存储就必须用Redis或数据库这种共享存储。

4.4 常见问题速查表

我把自己在实际项目中积累的问题整理成了一张速查表，方便你遇到问题时直接按图索骥：

4.5 部署环节的几个实用建议

项目交付时，部署这一关也是重点。我一般用gunicorn + Nginx组合，启多个worker进程，再配合Redis做Session存储，整体稳定性很好。启动命令大概是这样：

gunicorn -w 4 -b 127.0.0.1:8000 wsgi:app

Nginx那边反代到8000端口，顺便托管一下静态文件和上传的图片。生产环境千万不要用Flask自带的app.run()，那个开发服务器在多并发下非常脆弱。

数据库备份也要自动化。我写了条crontab，每天凌晨两点用mysqldump全量备份一次，保留最近7天：

0 2 * * * mysqldump -u backup_user -p****** gucci_erp > /backups/gucci_erp_$(date +%Y%m%d).sql

配置信息用.env统一管理，密钥、数据库密码这些敏感内容都不要写死在代码仓库里。代码里读取时用os.getenv取值，既方便切换测试/生产环境，也安全一些。

我在实际做进销存类系统的过程中最大的体会是：这种系统能不能落地，首当其冲的不是代码写得多漂亮，而是上线之前你有没有把业务规则问清楚——负库存允不允许、折扣怎么分摊、调拨在途库存怎么算、盘点差异怎么走审批。任何一个规则没定清楚，开发中途返工的代价都非常大。这个项目后续还可以做很多自然的扩展：对接小票打印机、增加门店间调拨审批流、做销售趋势图表、加一个采购预测模块，都是顺着现有数据模型往里添功能的事。只要库存流水这个根基稳了，往上盖楼就不慌。