Python+Flask实现五谷杂粮仓库进销存与保质期管理

简介库存管理系统是仓储业务数字化的核心而进销存逻辑的稳健性直接决定系统能否长期可靠运行。在食品类仓储场景中批次管理与保质期预警更是不可忽视的刚性需求。本文以五谷杂粮养生仓库为实例基于Python与Flask框架搭建了一套完整的仓库管理体系覆盖商品资料、批次库存、出入库作业、临期预警及养生标签推荐等关键模块。通过SQLite轻量级存储与WAL模式优化解决了小规模场景下的并发写入与数据备份问题。同时围绕单位换算、小数精度、先进先出策略等工程细节给出了可落地的实现方案。这套设计不仅适用于杂粮店铺的库存管理也可为其他食品类进销存系统的开发提供参考帮助开发者快速构建具备批次追溯与效期管控能力的仓储应用。 作为一个常年和库存管理系统打交道的人我见过太多拍脑袋写出来的进销存代码。很多项目上线第一天就崩不是功能不够炫而是压根没想清楚这个仓库到底要管什么。这次分享一个我最近重构过的项目——基于Python的五谷杂粮养生仓库设计源码。它不是那种教科书里改个商品名的玩具项目而是把杂粮仓储的特殊性、养生场景的标签体系、以及批次保质期管理全部揉进去的完整方案。如果你正准备做课程设计、接一个小店进销存私活或者单纯想练练Python项目架构这篇内容会帮你省掉大量试错成本。我会把核心表结构、出入库策略、保质期预警的实现方案以及在实测里踩到的坑全部摊开讲。1. 为什么是五谷杂粮养生仓库需求先行的设计起点1.1 杂粮仓储与普通进销存的本质差异大多数人一听到仓库管理系统第一反应就是商品表加库存数字加减法。这个思路放到五谷杂粮仓库里一定会出问题。我最早接这个需求时也差点走偏后来跟开养生杂粮铺的朋友聊完才发现这个领域有几个普通系统根本cover不住的硬约束。首先是保质期管理。普通超市进销存常常把保质期当一个备注字段存字符串但杂粮仓库里薏米、红豆、黑芝麻这些食材的保质期随季节和存储条件变化非常大。同一个批次进货的糙米夏天可能45天就开始出油变质冬天却能放3个月。如果不用date类型精确到天并且把批次维度落到每个SKU上预警逻辑根本无从谈起。其次是单位换算问题。杂粮行业散装和袋装并存进货时按麻袋50公斤零售时按250克一包养生套餐搭配时又可能按120克一盒。这种多单位换算如果靠前端表单硬填早晚会把库存搞出负数。我在设计里统一用克作为基准单位入库时自动换算出库时输入目标单位代码里直接换算后扣减。第三个差异更隐蔽五谷杂粮仓库存的是食品不是五金零件。客户不仅关心还有没有货更关心这批货新不新鲜适不适合我现在的体质。这意味着系统的信息架构要能承载养生标签、食材功效、搭配建议这些维度。这不是在商品表里加一个备注字段那么简单而是要配套独立的标签体系和推荐逻辑。1.2 系统边界与功能清单明确差异之后我把这个项目的功能边界划定为五个核心模块基础资料杂粮品项管理、供应商管理、仓库/库位管理批次库存按批次记录进货日期、保质期、来源供应商自动汇总可用库存出入库作业入库单、出库单、退库单统一走批次策略预警机制低库存预警、临期商品预警、过期商品锁定养生维度食材功效标签、体质标签、搭配推荐排除在外的是订单支付、会员积分、复杂财务报表。原因很简单仓库管理的核心是账实相符把出入库和批次账做扎实比堆砌一百个报表页有用得多。这也是我给所有做类似系统的人的第一个建议——先划边界再写代码。2. 技术选型与项目骨架搭建2.1 为什么是Flask SQLite的组合技术选型这块我见过不少人一上来就Spring Boot加MySQL结果一个单机小店项目搞出十几个微服务。这个项目我选了Python 3.10 Flask 2.3 SQLite理由很实际。SQLite在这个场景下完全够用。五谷杂粮仓库哪怕经营得不错一天的出入库单量也就是几十到几百条SQLite的读写性能毫无压力。它最大的好处是零配置数据库就是单个文件备份时直接复制文件走人对非专业运维的小店来说是刚需。只有当你想做多门店数据汇总或实时数据分析时才需要考虑换PostgreSQL。Flask的选型更简单——它足够轻又能清晰地把蓝图Blueprint拆开。对比过Django的同学都懂Django的ORM和Admin确实强大但在这个项目里我们需要高频定制库存逻辑、批次流水和养生标签推荐Flask的灵活度更顺手。当然纯粹的Python内置库也是可以的但Flask的请求上下文和模板渲染能省下不少重复代码。2.2 项目结构与配置一个能长期维护的项目目录结构在动手写第一行代码之前就应该定下来。我采用的是分模块结构grain_warehouse/ ├── app.py # 应用入口 ├── config.py # 配置 ├── requirements.txt ├── modules/ │ ├── __init__.py │ ├── goods.py # 品项管理蓝图 │ ├── stock.py # 库存蓝图 │ ├── batch.py # 批次蓝图 │ ├── inbound.py # 入库蓝图 │ ├── outbound.py # 出库蓝图 │ ├── health.py # 养生标签蓝图 │ └── report.py # 报表蓝图 ├── models/ │ ├── __init__.py │ ├── goods.py │ ├── batch.py │ ├── inventory.py │ ├── transaction.py │ ├── supplier.py │ └── tag.py ├── utils/ │ ├── unit.py # 单位换算 │ ├── date_helper.py # 日期计算 │ └── validators.py ├── templates/ │ ├── base.html │ ├── goods/ │ ├── stock/ │ └── health/ ├── static/ └── data/ └── warehouse.dbmodels和modules分开业务逻辑和数据模型解耦以后数据库迁移或增加接口层都会轻松很多。config.py里需要关注的配置项我列一下import os BASE_DIR os.path.dirname(os.path.abspath(__file__)) class Config: SECRET_KEY os.environ.get(SECRET_KEY, dev-secret-key-change-me) SQLALCHEMY_DATABASE_URI sqlite:/// os.path.join(BASE_DIR, data, warehouse.db) SQLALCHEMY_TRACK_MODIFICATIONS False # 低库存预警百分比 LOW_STOCK_THRESHOLD 0.2 # 临期预警天数 EXPIRY_WARNING_DAYS 30这里要特别提醒一个坑SQLAlchemy直接连SQLite时数据库目录必须提前存在否则会报找不到文件。我一般会在启动时用os.makedirs自动创建data目录这个细节能省去部署时很多莫名奇怪的报错。3. 数据库模型设计库存、批次与保质期3.1 核心表结构与字段设计的理由数据库是整个系统最不能将就的部分字段多一个冗余少一个就等着后面打补丁。我最终落地了六张核心表每张表的字段都有明确的设计理由。第一张是商品表goods。这里和普通商品表最大的区别是我加了unit_base和unit_conversion两个字段。unit_base固定是克unit_conversion存JSON格式的换算关系例如{公斤: 1000, 袋: 500, 斤: 500}。这样一个糙米SKU既能按公斤进货又能按袋出库。class Goods(db.Model): __tablename__ goods id db.Column(db.Integer, primary_keyTrue) code db.Column(db.String(32), uniqueTrue, nullableFalse) name db.Column(db.String(128), nullableFalse) category db.Column(db.String(32), indexTrue) # 米类/豆类/谷类/干货类 unit_base db.Column(db.String(16), default克) unit_conversion db.Column(db.JSON, defaultdict) shelf_life_days db.Column(db.Integer, nullableFalse) # 默认保质期天 health_tags db.Column(db.JSON, defaultlist) # 养生标签冗余存储第二张也是最核心的一张是批次表batch。每个批次代表一次独立进货拥有独立的有效期。这样设计才能回答库存里这50斤薏米哪些还有40天到期这类问题。字段里的quantity表示这个批次初始数量remaining是剩余数量两者分离的意义在于可以追溯该批次的消耗速率。class Batch(db.Model): __tablename__ batch id db.Column(db.Integer, primary_keyTrue) goods_id db.Column(db.Integer, db.ForeignKey(goods.id), nullableFalse, indexTrue) supplier_id db.Column(db.Integer, db.ForeignKey(supplier.id)) batch_no db.Column(db.String(64), uniqueTrue, nullableFalse) quantity db.Column(db.Numeric(12, 3), nullableFalse) # 初始入库数量克 remaining db.Column(db.Numeric(12, 3), nullableFalse) # 剩余数量克 unit db.Column(db.String(16), default克) inbound_date db.Column(db.Date, nullableFalse) expire_date db.Column(db.Date, nullableFalse) location db.Column(db.String(64)) # 库位信息第三张表是库存聚合表inventory。你可能觉得有批次表就能算出库存了为什么还要单独一张表原因很简单批次数据是明细每次查询汇总都要做聚合计算数据量大了之后会拖慢展示速度。inventory表以goods_id为维度维护当前总可用库存、已被订单锁定的数量、预警状态是一种典型的空间换时间策略。第四张表是流水表transaction_log。所有入库、出库、盘盈盘亏、退库都落这里字段统一设计为类型、关联批次、变动数量、变动前库存、变动后库存、操作人、操作时间。这张表是不能改不能删的它是将来对账和审计的凭证。第五张和第六张分别是supplier供应商表和health_tag养生标签表。供应商表很简单保存名称、联系人、资质编号、合作状态。养生标签表我单独拎出来是因为它和商品是多对多关系而且标签本身需要承载功效说明和适用体质。3.2 保质期与小数精度处理的特殊考虑关于保质期我在设计阶段就排除了某个字段存字符串的偷懒做法。所有日期都必须用datetime.date类存储因为后面要做的临期预警本质上是日期减法from datetime import date, timedelta warn_date date.today() timedelta(daysConfig.EXPIRY_WARNING_DAYS) expiring_batches Batch.query.filter( Batch.expire_date warn_date, Batch.expire_date date.today(), Batch.remaining 0 ).all()这个查询能直接捞出未来30天内临期且还有库存的批次。如果当初把保质期存成2025年6月这种字符串这段逻辑根本跑不起来。另一个非常容易翻车的点是小数精度。库存数量涉及公斤、克、袋三个维度的换算如果用float存储入库100袋糙米每袋500克总量50000克float的二进制误差会让库存对账差出0.00000001克。虽然单笔看不出来但流水跑一个月后盘点表一定会出幺蛾子。所以所有重量字段我都用了Numeric(12, 3)即最大10位整数加3位小数后缀克为单位完全够用。单位换算工具函数统一收口在utils/unit.py里禁止在业务代码里手工乘除。4. 核心业务模块的实现细节4.1 入库流程批次拆分与库存联动入库是仓库系统的起点也是最容易产生脏数据的地方。我设计的入库逻辑分三步创建入库单、生成批次记录、更新聚合库存。这三步必须在一个数据库事务里完成任何一步失败都要全部回滚。每批次唯一的batch_no我用时间戳加随机数生成格式是IN 年月日 随机四位。例如IN202506120482。这个编号同时用于后续的保质期追溯无论哪次出库都能通过流水关联回具体批次和供应商。入库视图函数的关键代码如下注意事务和异常回滚的写法inbound_bp.route(/create, methods[POST]) def create_inbound(): form_data request.get_json() try: db.session.begin() goods Goods.query.get(form_data[goods_id]) supplier Supplier.query.get(form_data[supplier_id]) # 单位换算统一转为基准单位克 qty_grams convert_to_base(goods, form_data[quantity], form_data[unit]) # 计算到期日期 expire_date date.fromisoformat(form_data[inbound_date]) timedelta( daysgoods.shelf_life_days ) batch Batch( goods_idgoods.id, supplier_idsupplier.id, batch_nogenerate_batch_no(), quantityqty_grams, remainingqty_grams, unit克, inbound_datedate.fromisoformat(form_data[inbound_date]), expire_dateexpire_date, locationform_data.get(location, ) ) db.session.add(batch) # 同步聚合库存 inventory Inventory.query.filter_by(goods_idgoods.id).with_for_update().first() if not inventory: inventory Inventory(goods_idgoods.id, available_qty0, locked_qty0) db.session.add(inventory) inventory.available_qty Decimal(str(inventory.available_qty)) Decimal(str(qty_grams)) # 写流水 log TransactionLog( transaction_typeINBOUND, batch_idbatch.id, change_qtyqty_grams, before_qty0, after_qtyqty_grams, operatorform_data.get(operator, admin) ) db.session.add(log) db.session.commit() return jsonify({code: 0, message: 入库成功, batch_no: batch.batch_no}) except Exception as e: db.session.rollback() return jsonify({code: 1, message: f入库失败: {str(e)}}), 500这里有两个细节值得展开。第一Inventory查询时用了with_for_update()目的是行级锁防止两个客户端同时入库导致库存被覆盖。SQLite在默认情况下其实对写操作有数据库级锁但使用with_for_update让逻辑更加明确将来切换MySQL/PostgreSQL时也能无缝迁移。第二所有计算过程中的Decimal加法必须把数据库返回的Decimal或float先转成字符串再构造Decimal对象直接Decimal(float_value)会重新把二进制误差带进来这个坑我踩过一次。4.2 出库策略低成本先进先出的实现思路出库是库存管理最考验业务逻辑的部分。五谷杂粮作为食品必须坚持先进先出不能让早期入库的薏米因为放在货架靠里就被遗忘到过期。具体实现时我没有直接去更新Batch.remaining而是先通过一个计算函数找出需要扣减的批次列表再逐批更新。这个函数的核心思路是按到期日期升序排批次从剩余量最多的旧批次开始扣减直到扣完全部需求数量def allocate_batches(goods_id, quantity_grams): batches Batch.query.filter( Batch.goods_id goods_id, Batch.remaining 0, Batch.expire_date date.today() ).order_by(Batch.expire_date.asc()).all() allocations [] need Decimal(str(quantity_grams)) for batch in batches: if need 0: break take min(batch.remaining, need) allocations.append((batch, take)) need - take if need 0: raise ValueError(库存不足当前可用库存无法满足出库数量) return allocations然后出库接口拿到allocations后逐条更新batch.remaining和聚合库存。注意每一步都要写流水一条出库单可能会拆成多条批次流水这样每批次的动销速率才能真实反映出来。outbound_bp.route(/create, methods[POST]) def create_outbound(): form_data request.get_json() goods Goods.query.get(form_data[goods_id]) qty_grams convert_to_base(goods, form_data[quantity], form_data[unit]) try: db.session.begin() allocations allocate_batches(goods.id, qty_grams) inventory Inventory.query.filter_by(goods_idgoods.id).with_for_update().first() if not inventory or Decimal(str(inventory.available_qty)) qty_grams: db.session.rollback() return jsonify({code: 1, message: 库存不足}), 400 for batch, take in allocations: before batch.remaining batch.remaining Decimal(str(batch.remaining)) - take log TransactionLog( transaction_typeOUTBOUND, batch_idbatch.id, change_qty-take, before_qtybefore, after_qtybatch.remaining, operatorform_data.get(operator, admin) ) db.session.add(log) inventory.available_qty Decimal(str(inventory.available_qty)) - qty_grams db.session.commit() return jsonify({code: 0, message: 出库成功}) except Exception as e: db.session.rollback() return jsonify({code: 1, message: str(e)}), 400这样设计的好处是每一个批次剩余多少、出了多少、还剩多少在批次表里一目了然。对杂粮店来说哪个供应商的货卖得慢、哪个批次的损耗高都能直接统计出来。4.3 低库存与临期预警定时任务与查询结合预警系统不能只在用户打开页面时才触发因为库存变化最频繁的时候往往是出入库操作之后。我的做法是双管齐下在一次完整的请求里做同步预警检测同时提供后台定时任务扫描。同步检测集中在出库和入库视图函数里每次库存变动后重新计算该商品的可用库存和临期批次数量如果触发条件就写入预警消息表。这样店主打开网页时就能第一时间看到糙米库存已低于20%3批次红豆将于15天后过期。定时任务我用的是APScheduler每天早上8点扫描所有批次对临期批次自动生成预警from apscheduler.schedulers.background import BackgroundScheduler def daily_expiry_scan(): today date.today() warn_date today timedelta(daysConfig.EXPIRY_WARNING_DAYS) expiring_batches Batch.query.filter( Batch.expire_date warn_date, Batch.expire_date today, Batch.remaining 0 ).all() for batch in expiring_batches: create_warning(f批次{batch.batch_no}的{batch.goods.name}将于{batch.expire_date}到期剩余{format_kg(batch.remaining)}) scheduler BackgroundScheduler() scheduler.add_job(daily_expiry_scan, cron, hour8, minute0) scheduler.start()预警表warnings我设置了status字段NEW表示未处理CONFIRMED表示已确认RESOLVED表示已处理比如做了促销或低价处理。这样店主看到预警后能跟进下一步动作而不是看一眼就被动消失预警才真正有管理闭环的价值。5. 养生维度的价值延伸功效标签与搭配推荐5.1 标签体系设计让食材和体质对上话这个项目和其他仓库系统最显著的差异点就是养生维度的标签体系。五谷杂粮的消费者有强烈的目的性有人想健脾有人想祛湿有人想补气血。如果只是把食材当成普通商品管理就浪费了这个垂直领域的独特性。我在health_tag表里按两个维度设计标签功效标签如健脾养胃、祛湿消肿、补气养血、润肺止咳和体质标签如气虚质、湿热质、阳虚质。一个食材可以挂多个功效标签一个功效标签也可以反向推荐多个食材因此用单独的表关联goods和tag。class HealthTag(db.Model): __tablename__ health_tag id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(32), uniqueTrue, nullableFalse) description db.Column(db.String(255)) tag_type db.Column(db.String(16), nullableFalse) # SCENE / CONSTITUTION商品与标签的关联我就没单独建表了而是在Goods表里用JSON冗余存储health_tags字段方便列表页快速展示。当然如果你对关联查询的性能特别敏感可以建一张goods_tag关联表但这个项目的数据量完全没必要。5.2 基于标签的搭配推荐逻辑有了标签体系我增加了一个实用的推荐接口和页面输入一个体质或场景关键词系统自动从当前有库存的食材中找出匹配项并根据标签重叠度推荐搭配组合。举个例子用户输入祛湿系统会筛选出带有祛湿消肿标签的食材赤小豆、薏米、芡实、茯苓、陈皮。再进一步按功效相似度做黄金组合推荐——赤小豆薏米是经典搭配可以再辅以少量陈皮理气。推荐逻辑的核心是一个简单的打分函数def recommend_by_tag(tag_name, limit6): tag HealthTag.query.filter_by(nametag_name).first() if not tag: return [] goods_list Goods.query.filter( Goods.health_tags.contains(tag.name) ).all() scored [] for goods in goods_list: inv Inventory.query.filter_by(goods_idgoods.id).first() qty Decimal(str(inv.available_qty)) if inv else Decimal(0) if qty 0: continue score len(set(goods.health_tags) recommended_synergy_tags.get(tag_name, set())) scored.append({goods: goods.name, available: format_kg(qty), score: score}) scored.sort(keylambda x: x[score], reverseTrue) return scored[:limit]打分的时候如果一个食材同时满足祛湿和健脾两个推荐维度权重就更高。这个逻辑虽然简单但效果非常符合五谷杂粮店的场景具备复合功效的食材在页面里的排序更靠前店主推荐给顾客时也更有底气。5.3 场景页面的落地从标签到养生方案为了让标签体系不止停留在代码里我给系统加了一个养生方案展示页。这个页面按典型场景分组例如春季祛湿熬夜护肝调理脾胃孕妇营养每个场景下动态展示当前有库存的食材和推荐搭配理由。这个页面在模板里直接用Jinja2渲染数据来自场景配置加标签推荐逻辑的组合。实际部署后朋友店里的店员反馈说这个页面对销售帮助很大因为很多顾客其实不知道自己需要什么看到祛湿组合推荐之后更容易下单。仓库系统从纯后台工具变成了能辅助销售决策的活字典这是最初设计时完全没想到的意外收获。6. 实测运行、踩坑记录与优化建议6.1 实测中遇到的典型问题项目写了大概1200多行Python代码后我把它跑起来做了几个月的模拟数据测试。以下是测试中遇到的最有代表性的几个问题每个都花了不少时间排查。第一个是SQLite的并发写锁问题。用Flask默认的SQLite配置两个请求同时写库时偶尔会出现database is locked错误。原因在于SQLite只允许一个进程同时写而Flask开发服务器多线程模式下并发写很容易撞车。解决方案有两个层面第一把应用改用WAL模式让读写并发能力提升几个量级第二在数据库连接层面设置合理的timeout。我最终两个都用了# config.py import sqlite3 from sqlalchemy import event from sqlalchemy.engine import Engine event.listens_for(Engine, connect) def set_sqlite_pragma(dbapi_connection, connection_record): cursor dbapi_connection.cursor() cursor.execute(PRAGMA journal_modeWAL) cursor.execute(PRAGMA busy_timeout5000) cursor.close()改完以后开发环境并发写基本稳定。如果将来数据量真的大到要换MySQL代码结构也能平滑迁移。第二个坑是单位换算的精度。我前期用float做乘法测试时发现10次出库、每次按袋出1袋500克后剩余库存变成了499.9999999999克盘点时永远差一口气。后来把所有涉及重量的加减乘除全部改为Decimal并在边界做四舍五入到3位小数问题才彻底解决。强烈建议在项目一开始就把Decimal当成默认选择不要觉得麻烦。第三个问题出在后端日期传递。前端表单把日期字符串传给后端时我最初用了datetime.strptime(%Y-%m-%d)解析结果每次输入格式不一致就报ValueError。后来统一封装了date.parse_date函数做容错并强制前端用date控件输入总算一劳永逸。6.2 性能、并发与数据备份经验这个项目在实际场景里最大并发量不会太高但我还是做了一些优化以防万一。查询侧我在batch表的goods_id、expire_date、remaining上分别建了索引低库存预警和临期批次扫描的SQL都不慢。对inventory表的查询本来就以goods_id为唯一条件加主键索引就足够了。报表页面的累计销量统计我用的是流水表按batch关联goods的分组聚合数据量在几万条时响应时间仍然在毫秒级。数据备份是仓库系统的生命线。SQLite的单文件特性让备份变得异常简单——直接用文件复制。我写了一个简单的自动备份脚本每天凌晨将warehouse.db复制到backup目录文件名带上日期保留最近30天。另外因为启用了WAL模式备份前最好先执行一次checkpoint把WAL文件中的数据合并回主库不然直接复制主库文件可能丢数据。这个细节很多人容易忽略。# backup.py import shutil from datetime import date def backup_db(): today date.today().isoformat() src data/warehouse.db dst fbackup/warehouse_{today}.db # 执行 checkpoint 将 WAL 内容合并回主文件 from sqlalchemy import text db.session.execute(text(PRAGMA wal_checkpoint(FULL))) db.session.commit() shutil.copy2(src, dst)6.3 可扩展的方向如果这个项目要继续往前走我个人最想做的扩展有两个方向。第一个是增加完整的批次效期看板用可视化方式展示每个批次的生命周期从入库、临期到售罄全流程跟踪。第二个是把它做成简单的B/S多端适配店主在手机上就能查库存、录出库而不用每次跑回电脑前面操作。前端用Bootstrap已经做了一点适配但要真正在小屏幕上舒服地用还得调整导航和信息密度。另外养生推荐这个点其实有非常大的延展空间。如果后续数据量积累到足够多可以基于食材的性味归经、季节时令做更细粒度的推荐规则甚至可以接入外部食材数据库做营养分析。这些都是把这个垂直仓库系统做深做透的好方向。从需求梳理到数据库建模再到业务代码落地这个基于Python的五谷杂粮养生仓库项目最核心的收获是在面对一个垂直领域时不要急着套通用模板先想清楚这个领域的特殊规则。食品有批次和保质期的底线养生场景有标签和推荐的增值把这两层做扎实了系统才真正好用。最后再提醒一句所有出库、入库、预警的代码在上线前一定要用真实历史数据做一轮联调模拟账实不符时的对账流程。这套系统我已经在朋友的店里跑了大半年目前最常被夸的反而不是界面多好看而是三个月前的哪一批薏米还剩多少一查就知道了这类看似基础但足以让经营者安心的能力。本文还有配套的精品资源点击获取