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.xORMFlask-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 核心表结构设计我把核心表列出来并附上关键字段这样你对整体模型会有直观认识。表名用途关键字段users系统用户id, username, password_hash, role, store_idstores门店/仓库id, name, code, type(store/warehouse)suppliers供应商id, name, contact, phone, statusproducts商品id, sku, barcode, name, category_id, cost_price, sale_price, min_stockpurchase_orders采购单id, order_no, supplier_id, status, total_amount, operator_idpurchase_order_items采购单明细id, order_id, product_id, quantity, priceinbound_records入库记录id, order_id, product_id, store_id, quantity, batch_no, created_atsales_orders销售单id, order_no, store_id, customer_info, total_amount, status, operator_idsales_order_items销售单明细id, order_id, product_id, quantity, priceoutbound_records出库记录id, order_id, product_id, store_id, quantity, created_atstock实时库存id, product_id, store_id, quantity, locked_quantity, updated_atstock_movements库存流水id, product_id, store_id, type, quantity, ref_no, ref_type, balance_before, balance_after, operator_id, created_atstocktake_records盘点记录id, store_id, product_id, actual_qty, book_qty, diff_qty, status, operator_idtransfer_orders调拨单id, order_no, from_store_id, to_store_id, status, operator_id这里有几个设计细节值得细说。商品表我要求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, mysqlpymysql://root:passwordlocalhost/gucci_erp?charsetutf8mb4) 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_keyTrue) username db.Column(db.String(64), uniqueTrue, nullableFalse, indexTrue) password_hash db.Column(db.String(255), nullableFalse) role db.Column(db.String(32), nullableFalse, defaultclerk) # admin, purchaser, clerk, finance, store_manager store_id db.Column(db.Integer, db.ForeignKey(stores.id)) status db.Column(db.SmallInteger, default1, comment1-正常 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_idorder.id)) # 开启事务 try: for item in order.items: product Product.query.filter_by(iditem.product_id).with_for_update().first() if not product: raise Exception(f商品ID {item.product_id} 不存在) # 判断入库目标仓库 store_id request.form.get(store_id, typeint) if not store_id: raise Exception(请选择入库仓库) stock Stock.query.filter_by( product_idproduct.id, store_idstore_id ).with_for_update().first() if not stock: stock Stock(product_idproduct.id, store_idstore_id, quantity0, locked_quantity0) db.session.add(stock) before_quantity stock.quantity stock.quantity item.quantity movement StockMovement( product_idproduct.id, store_idstore_id, typeINBOUND, quantityitem.quantity, ref_typePurchaseOrder, ref_noorder.order_no, balance_beforebefore_quantity, balance_afterstock.quantity, operator_idget_current_user().id ) db.session.add(movement) inbound InboundRecord( order_idorder.id, product_idproduct.id, store_idstore_id, quantityitem.quantity, batch_nof{order.order_no}-{product.id}, operator_idget_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_idorder.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, typeint) 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_idproduct.id, store_idstore_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_idproduct.id, store_idstore_id, typeSALE, quantityquantity, ref_typeSalesOrder, ref_noorder_no, balance_beforebefore_quantity, balance_afterstock.quantity, operator_idget_current_user().id ) db.session.add(movement) item_total product.sale_price * quantity total item_total so_item SalesOrderItem( product_idproduct.id, quantityquantity, priceproduct.sale_price, amountitem_total ) db.session.add(so_item) sale_order SalesOrder( order_noorder_no, store_idstore_id, total_amounttotal, statuspaid, operator_idget_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 权限系统与操作日志进销存系统不能省的部分进销存系统里权限不是花架子它直接关系到财务数据的安全。我按角色做了四级权限角色权限范围管理员 admin全部功能含用户管理、系统配置采购员 purchaser采购单、入库单、供应商管理店长 store_manager销售、调拨、盘点、查看本店库存导购/收银员 clerk仅销售开单、查看销售记录财务 finance查看报表、对账、导出实现方式是在蓝图路由上套一个装饰器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_keyTrue) user_id db.Column(db.Integer, nullableFalse) username db.Column(db.String(64), nullableFalse) action db.Column(db.String(64), nullableFalse, commentcreate/update/cancel/confirm) target_type db.Column(db.String(32), nullableFalse, commentPurchaseOrder/SalesOrder/Stocktake...) target_no db.Column(db.String(64), nullableFalse) description db.Column(db.Text) created_at db.Column(db.DateTime, defaultdatetime.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存储从文件系统改成了Redisfrom flask_session import Session app.config[SESSION_TYPE] redis app.config[SESSION_REDIS] redis.from_url(redis://localhost:6379/0) Session(app)改完这个再没出现掉登录的问题。如果你的系统规模不大、单进程跑文件系统Session问题不大但只要用gunicorn起了多个workerSession存储就必须用Redis或数据库这种共享存储。4.4 常见问题速查表我把自己在实际项目中积累的问题整理成了一张速查表方便你遇到问题时直接按图索骥现象可能原因解决建议库存为负数并发扣减未加锁库存操作打包进事务使用with_for_update()行锁单据金额和历史不一致明细表未冗余价格商品调价导致历史联表查出新价明细表存储交易时的单价与金额登录状态频繁丢失Session存在本地文件多进程不共享改用Redis存储Session报表查询越来越慢缺少复合索引或每次都全表扫描流水加(product_id, store_id, created_at)复合索引使用每日汇总表商品被误删导致历史单据串味商品表直接DELETE改为软删除status字段禁止物理删除库存变动对不上账只改库存表没写流水表建立强制规则任何库存变动必须同时写stock_movements流水断电/崩溃后部分单据丢失未使用事务多条SQL部分成功部分失败所有多步写操作包进事务失败整体回滚页面接口返回乱码MySQL连接未指定utf8mb4连接串加?charsetutf8mb4并确认表和库的字符集一致4.5 部署环节的几个实用建议项目交付时部署这一关也是重点。我一般用gunicorn Nginx组合启多个worker进程再配合Redis做Session存储整体稳定性很好。启动命令大概是这样gunicorn -w 4 -b 127.0.0.1:8000 wsgi:appNginx那边反代到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取值既方便切换测试/生产环境也安全一些。我在实际做进销存类系统的过程中最大的体会是这种系统能不能落地首当其冲的不是代码写得多漂亮而是上线之前你有没有把业务规则问清楚——负库存允不允许、折扣怎么分摊、调拨在途库存怎么算、盘点差异怎么走审批。任何一个规则没定清楚开发中途返工的代价都非常大。这个项目后续还可以做很多自然的扩展对接小票打印机、增加门店间调拨审批流、做销售趋势图表、加一个采购预测模块都是顺着现有数据模型往里添功能的事。只要库存流水这个根基稳了往上盖楼就不慌。