前阵子帮一个开小五金店的朋友搭了一套进销存系统需求听起来简单落地时各种细节却相当磨人进货单要能追到每一批货的供应商销售单要能对应到具体客户库存一改就得留下痕迹月底对账最好半小时搞定而不是熬一晚上。最终敲定的方案就是标题里这套——Python SQLite。没有用 MySQL没有搞前后端分离没有引入 Redis。整套系统跑在一台普通 Windows 电脑上数据库就是一个.db文件。从需求分析到第一版跑通花了不到一个周末后续加报表、加库存预警、加操作日志也都在这套骨架上稳步迭代。这篇文章就把整个项目拆开来复盘一遍。适合正在考虑给自己店里、Small 仓库或者个人项目搭一套轻量管理系统的朋友参考也适合刚学 Python 想拿真实场景练手的人。我不会只贴代码更想把每个设计决策背后的为什么讲清楚因为这些取舍才是这套方案真正的价值所在。1. 为什么选择 Python SQLite 这套轻量组合1.1 小型商业进销存的真实痛点朋友的那家五金店规模不大旺季一天出货也就百来单。之前一直是纸质单据加 Excel 表格配着来问题很明显商品 SKU 一多Excel 里改价、改库存特别容易串行一个格子填错整行数据就乱了进货价、售价、供应商、客户信息分散在好几张表格里想查一件商品的毛利得手动拼接Excel 文件传来传去版本不一致时间一长根本不知道哪个是最新的库存没有预警螺丝钉这类小件货什么时候断货完全凭感觉这类场景有一个共同特征数据量不大但是业务关系和流程环节一点不少。一张商品表解决不了问题需要一套真正的数据结构来承载进货→库存→销售→统计这条链路。1.2 SQLite 能扛住什么样的并发量很多人在选型时会有个刻板印象SQLite 就是个嵌入式玩具不能用于商业项目。这个判断得拆开看。SQLite 的定位是嵌入式关系型数据库数据存到一个独立文件里不需要单独安装服务端进程。它支持标准 SQL、支持事务 ACID、支持外键和索引。对于每天几十到几百笔订单的场景它的性能完全够用——瓶颈通常不在数据库而在业务逻辑代码写得好不好。从并发角度讲SQLite 默认支持多进程读单进程写。也就是说如果只是收银台一台电脑操作或者两三台电脑同时读取配合 WAL 模式体验相当流畅。它的限制主要体现在高并发写入比如几十个终端同时开单——那确实不是它的战场但也不是这类小系统会遇到的场景。选择 SQLite 还有一个巨大的隐性红利部署和备份都极其简单。整个数据库就是一个.db文件拷贝走就是备份拷到新电脑就能继续跑。不需要专门安装数据库服务不需要配置账号权限。对非技术背景的店主来说这比帮你装个 MySQL友好得多。我后来的经验是把 SQLite 理解为文件型数据库而不是弱化版数据库。它在自己适合的领域内表现得非常专业你只需要尊重它的边界。1.3 Python 在项目里的角色Python 在这套系统里承担三个角色业务逻辑层、数据访问层和界面层。业务逻辑层负责进销存规则比如销售出库时库存必须足够入库时自动更新成本价退货时恢复库存。数据访问层负责把 Python 对象和 SQLite 表互相转换。界面层用 Python 自带的标准库就能做控制台版本也可以后续换成 PyQt 或 Web 界面。选择 Python 的深层原因在于它的开发效率和生态。整表操作、事务管理、Excel 导入导出、报表生成都有现成库。比起 Java 和 C# 的工程化成本Python 更适合这套系统的规模和迭代节奏。2. 数据库表结构设计进销存系统的数据骨架2.1 从业务环节反推表设计进销存的核心链路就三个字进、销、存。进从供应商采购商品入库销向客户销售商品出库存实时掌握每个 SKU 的库存数量围绕这条链路我设计了八张核心表覆盖了从单据到明细到库存流水的完整闭环表名作用关键字段products商品档案sku, name, category, spec, unit, purchase_price, sale_price, min_stock, stocksuppliers供应商档案name, contact, phone, address, remarkcustomers客户档案name, contact, phone, address, remarkpurchase_orders进货单主表order_no, supplier_id, total_amount, order_date, status, remarkpurchase_items进货单明细order_id, product_id, quantity, unit_price, amountsales_orders销售单主表order_no, customer_id, total_amount, order_date, status, remarksales_items销售单明细order_id, product_id, quantity, unit_price, amountstock_records库存变动流水product_id, change_type, change_quantity, before_stock, after_stock, related_order_no订货单和销售单拆成主表和明细表两张是这套设计的核心。主表记录这一单跟谁做的、总金额多少、日期是哪天明细表记录这一单包含了哪些商品、各自多少数量多少钱。这样既能按单号查全貌也能按商品反查它进出过哪些单子。2.2 库存改动必须留痕真正让我坚持加stock_records这张流水表的是一次对账事故。有一次朋友反映库存对不上账面上剪刀少了 5 把。但系统里没有任何地方记录了这次变动是怎么发生的。单查销售单没卖过单查进货单也没进过。最后发现是盘点时手动调整库存但操作时没留下痕迹。从那以后我的设计原则就变成了任何库存变动必须同时写一条流水。不管是通过进货单增加库存还是通过销售单扣减库存甚至盘点修正都必须记录变动前后的数值、变动类型和对应的单据号。有了流水表之后对账逻辑就变成了期末库存 期初库存 进货总量 - 销售总量 - 报损量。每一项都能在流水表中找到对应记录再也不用猜库存去哪了。2.3 建表脚本中的关键细节-- 开启外键约束SQLite 默认不启用 PRAGMA foreign_keys ON; CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT UNIQUE NOT NULL, name TEXT NOT NULL, category TEXT, spec TEXT, unit TEXT DEFAULT 个, purchase_price REAL NOT NULL DEFAULT 0, sale_price REAL NOT NULL DEFAULT 0, min_stock INTGEER NOT NULL DEFAULT 0, stock INTEGER NOT NULL DEFAULT 0, status INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL DEFAULT (datetime(now, localtime)), updated_at TEXT NOT NULL DEFAULT (datetime(now, localtime)) );有几点值得单独强调sku字段加 UNIQUE 约束同一款商品只能有一条档案重复建档会在数据库层被拦截而不是靠业务代码说我检查过了stock字段类型用 INTEGER库存是整数永远不要用浮点数存数量created_at用字符串存本地时间datetime(now, localtime)是 SQLite 的内置函数读写直观排序也没问题商品表里加status字段做逻辑删除下架商品不删行只把 status 改为 0历史单据引用它的时候不会断链2.4 SQLite 默认不开启外键约束这个坑建表脚本里第一行写了PRAGMA foreign_keys ON这一行非常关键。SQLite 为了兼容旧项目默认情况下不开启外键约束。也就是说就算你在建表语句里写了FOREIGN KEY引用默认环境下它也不会真去校验。往 sales_items 里插入一个 product_id 完全不存在的记录数据库会照单全收。这会导致什么后果日后做联表查询时会因为找不到对应商品而返回一堆空数据对账时根本找不出根因。正确做法是在每次建立数据库连接后立即执行PRAGMA foreign_keys ON。注意这条指令是连接级的不是数据库级的每次连接都要执行。如果用的是 Python 的sqlite3模块可以在连接后统一设置。conn sqlite3.connect(inventory.db) conn.execute(PRAGMA foreign_keys ON)3. 核心业务逻辑的 Python 实现3.1 统一的数据访问层刚开始写这套系统时我直接在业务代码里到处写 SQL 语句。结果项目量一上来就发现痛苦了业务规则和 SQL 混在一起改一个库存字段要翻遍所有涉及的地方而且冗余代码多到怀疑人生。后来我抽了一个db.py模块做统一的数据访问层把所有通用的 CRUD 和数据库连接管理集中在这里import sqlite3 DB_PATH inventory.db def get_connection(): conn sqlite3.connect(DB_PATH, timeout10) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) return conntimeout10的意思是当数据库文件被其他连接锁住时当前连接最多等 10 秒而不是直接报错。row_factory sqlite3.Row让查询结果可以通过字段名访问row[sku]比row[0]可读性好太多。3.2 入库与出库事务的原子性进销存系统最核心的规则是单据和库存变动必须同时成功或者同时失败。比如销售出库程序要做三件事在 sales_orders 插入主表记录在 sales_items 插入明细记录把 products 表的库存减掉这三件事是一个整体。如果第 3 步失败而前两步已经提交系统就会出现单据显示卖了库存却没扣的脏数据。解决这个问题靠的就是数据库事务。Python 的sqlite3模块在开启事务时有个行为细节。默认情况下如果在autocommit模式之外执行INSERT/UPDATE/DELETEPython 会自动开启事务。真正容易踩坑的是并发环境下默认的BEGIN延迟事务可能在读操作后才升级为写锁导致死锁或锁冲突。我的处理方式是显式声明BEGIN IMMEDIATE。这样一开始就拿到写锁避免中途升级锁带来的复杂度def create_sale_order(order_no, customer_id, items): 创建销售单并扣减库存 items 格式: [{product_id: 1, quantity: 2, unit_price: 15}, ...] conn get_connection() cursor conn.cursor() try: # IMMEDIATE 事务立刻获取写锁避免并发下的锁升级问题 conn.execute(BEGIN IMMEDIATE) total_amount 0 for item in items: product_id item[product_id] quantity item[quantity] unit_price item[unit_price] amount quantity * unit_price total_amount amount # 查出当前库存 product cursor.execute( SELECT stock FROM products WHERE id ?, (product_id,) ).fetchone() if product is None: raise ValueError(f商品 {product_id} 不存在) if product[stock] quantity: raise ValueError(f商品 {product[id]} 库存不足当前库存 {product[stock]}) # 扣减库存并记录变动前的库存值 cursor.execute( UPDATE products SET stock stock - ?, updated_at datetime(now, localtime) WHERE id ?, (quantity, product_id) ) cursor.execute( INSERT INTO stock_records (product_id, change_type, change_quantity, before_stock, after_stock, related_order_no) VALUES (?, sale, ?, ?, ?, ?), (product_id, quantity, product[stock], product[stock] - quantity, order_no) ) # 写销售单主表 cursor.execute( INSERT INTO sales_orders (order_no, customer_id, total_amount, order_date, status) VALUES (?, ?, ?, datetime(now, localtime), 1), (order_no, customer_id, total_amount) ) order_id cursor.lastrowid # 写销售单明细 for item in items: cursor.execute( INSERT INTO sales_items (order_id, product_id, quantity, unit_price, amount) VALUES (?, ?, ?, ?, ?), (order_id, item[product_id], item[quantity], item[unit_price], item[quantity] * item[unit_price]) ) conn.commit() return order_id except Exception: conn.rollback() raise finally: conn.close()这段代码里有几个值得称道的细节先查库存再扣减两次操作在同一个事务内别怕中间的并发修改——因为有写锁挡着记录before_stock和after_stock流水表里既有变动量也有变动前后绝对值日后回查非常有用用conn.rollback()包裹异常任何一步失败前面所有操作全部撤销订单号由外部传入保证了调用方可以统一生成有规则的单号含日期、序列号后续查询追踪就方便了3.3 入库逻辑不仅加库存还要维护价格信息入库相对简单但有一个容易忽略的商业细节进货价格是会波动的。同一种商品这次进价 10 元下次可能涨到 11 元。如果只简单累加库存不更新purchase_price那么过一段时间你会发现库存成本完全失真。对小型批发零售场景我采用最新进货价覆盖成本价的策略def create_purchase_order(order_no, supplier_id, items): conn get_connection() cursor conn.cursor() try: conn.execute(BEGIN IMMEDIATE) total_amount 0 for item in items: product cursor.execute( SELECT stock, purchase_price FROM products WHERE id ?, (item[product_id],) ).fetchone() if product is None: raise ValueError(商品不存在) quantity item[quantity] unit_price item[unit_price] cursor.execute( UPDATE products SET stock stock ?, purchase_price ?, updated_at datetime(now, localtime) WHERE id ?, (quantity, unit_price, item[product_id]) ) cursor.execute( INSERT INTO stock_records (product_id, change_type, change_quantity, before_stock, after_stock, related_order_no) VALUES (?, purchase, ?, ?, ?, ?), (item[product_id], quantity, product[stock], product[stock] quantity, order_no) ) total_amount quantity * unit_price cursor.execute( INSERT INTO purchase_orders (order_no, supplier_id, total_amount, order_date, status) VALUES (?, ?, ?, datetime(now, localtime), 1), (order_no, supplier_id, total_amount) ) conn.commit() except Exception: conn.rollback() raise finally: conn.close()这样入库单、库存、成本价都保持同步。经营分析时那一列毛利 销售额 - 销售数量 × 最新成本价才能算得准。3.4 查询与预警让数据派上用场数据录进去了还得能高效查出来。日常最频繁的三个查询场景是按分类统计库存数量conn get_connection() rows conn.execute( SELECT category, SUM(stock) AS total_stock, COUNT(*) AS sku_count FROM products WHERE status 1 GROUP BY category ORDER BY total_stock DESC ).fetchall()库存预警warnings conn.execute( SELECT sku, name, stock, min_stock FROM products WHERE status 1 AND stock min_stock ORDER BY stock ASC ).fetchall()这两个查询看起来平平无奇但它们是整套系统决策价值的体现。我后来直接把预警写入日志表每天开店时自动打印一份应补货清单朋友说光是这个功能就值回开发成本。4. 开发与调试过程中踩过的坑4.1 database is locked 到底怎么回事第一次遇到这个错误是在一次导入测试数据的时候程序跑批处理连续插入几千条记录中途跳了sqlite3.OperationalError: database is locked。排查过程让我确认了一个关键认知SQLite 的锁粒度是整个数据库文件不是某一行或某一张表。这意味着任何一个连接持有写锁其他连接的写操作都会等待或者直接报错。当多个进程或者一个进程内多个连接同时写数据库时锁冲突是很容易出现的。解决办法有三个层次第一个层次是加重试机制。用timeout10参数让连接等待锁释放第二个层次是缩短单事务的时间。不要在事务里做网络请求或长时间计算快进快出第三个层次是引入 WAL 模式。Write-Ahead Logging 允许读操作和写操作并发写事务之间仍然互斥但读写混跑的场景会顺滑很多conn.execute(PRAGMA journal_mode WAL;)注意WAL 模式会生成额外的.db-wal和.db-shm文件备份时要连它们一起拷贝或者先执行一次 checkpoint。这一点我在后面备份部分会再次提到。4.2 金额计算别用浮点数这是我早期代码里最早埋下的雷。当时unit_price字段用的REALPython 侧直接quantity * unit_price算金额。表面看起来没什么问题直到某天发现销售单汇总金额和 Excel 里手算的对不上差了几分钱。原因在于浮点数的二进制表示不精确。比如0.1在浮点数里存得其实是一个无限循环的近似值多次运算后误差就会被放大。解决方案也很直接金额统一用分为单位存整数INTEGER。数据库表里该字段可以命名为purchase_price_cents、sale_price_cents业务层再除以 100 转成元展示。# 正确做法金额以分为单位 UPDATE products SET sale_price_cents ? WHERE id ?, (1999, product_id) # 19.99 元 # 而不是 UPDATE products SET sale_price ? WHERE id ?, (19.99, product_id)这个习惯我到现在都保留着凡是涉及钱和数量的字段能用整数绝不用浮点数。宁可改表结构时多写几行也不去趟浮点误差的浑水。4.3 多线程下的 check_same_thread 陷阱后来系统加了 PyQt 界面SQLite 操作放到了子线程里程序运行几分钟后就开始报错sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread.这是sqlite3模块为了线程安全做的一个保护措施。默认情况下一个连接在哪个线程创建就只能被哪个线程使用。当时手足无措第一反应是搜到有人说加check_same_threadFalse就行了。确实能绕过这个限制但绕过不等于安全——如果多个线程同时使用同一个连接写数据底层锁冲突依然存在很可能出现不可预期的数据错乱。我最终采用的是**每线程独立连接 连接池**的折中方案业务层不共享连接每次操作通过函数获取独立连接用完即关。对于这套小系统的并发量完全够用还省去了 thread-local 管理的复杂度。import threading _local threading.local() def get_thread_connection(): conn getattr(_local, conn, None) if conn is None: conn sqlite3.connect(DB_PATH, timeout10) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) conn.execute(PRAGMA journal_mode WAL;) _local.conn conn return conn这个方案的好处是每个线程都有自己的专用连接不需要加锁而 WAL 模式保证了跨线程读写不互相阻塞。4.4 不要在主线程里跑批量导入顺带说一个相关经验。给朋友迁移 Excel 旧数据时几千行记录逐条 INSERT界面直接卡死。原因很明显主线程忙得没空处理界面刷新。正确做法是把耗时的数据导入放到后台线程里跑同时主线程隔一段时间刷新进度条。如果你是用控制台版至少也加一个进度打印别让用户以为程序死了。5. 日常运维备份、数据安全与可视化工具5.1 用 DB Browser for SQLite 做可视化管理系统跑起来之后我强烈建议装上DB Browser for SQLite这个小工具。它是免费开源的 SQLite 可视化客户端Windows / macOS / Linux 都有安装包。它的用途不只是看看数据。日常运维中你会频繁用到它做这些事情直接浏览八张表的原始数据比在代码里写查询调试快得多执行临时 SQL比如批量修正错误分类、手动调整库存导出 CSV / Excel在不写代码的前提下把数据交给店主用 Excel 二次分析查看数据库结构和索引帮你验证建表脚本执行后到底建成了什么样我见过不少人折腾半天sqlite3命令行去查数据效率极低。可视化工具虽然简单但它能让你把精力花在业务判断上而不是敲命令上。5.2 备份策略比拷贝文件再多做一步SQLite 备份最直觉的做法是直接复制.db文件。如果程序恰好没有打开这个数据库直接拷贝是有效的但如果程序正在运行中直接复制文件可能复制到不一致的状态尤其是开启了 WAL 模式后-wal文件里那些尚未合并进主库的数据可能丢失。我推荐用 Python 自带的备份 API 做在线备份import sqlite3 def backup_db(src_db_path, dst_db_path): src_conn sqlite3.connect(src_db_path) dst_conn sqlite3.connect(dst_db_path) try: src_conn.backup(dst_conn) print(f备份完成: {dst_db_path}) finally: dst_conn.close() src_conn.close()sqlite3.Connection.backup()是官方推荐方式支持在线备份多个连接同时读取时也能保证一致性。定时任务里每天凌晨跑一次把库备份到移动硬盘或者云盘同步目录就已经非常稳妥了。记得做一次备份恢复演练。我曾经顺顺利利备份了大半个月直到朋友有一天误删了一张表想恢复才发现备份文件本身就有问题。定期用备份文件恢复到临时目录跑一次查询几分钟的事能少掉一晚上的慌乱。5.3 关于 SQLite 文件加密的实话实说热搜词里有个sqlite数据库文件能否加密这里明确讲一下。SQLite 官方版本不提供原生加密。如果有人直接拿.db文件就能用文本编辑器看到里面的明文数据。想给 SQLite 加密码常规手段有几个SQLCipher它是 SQLite 的加密分支对数据库文件做 AES 加密。Python 接入的话要用pysqlcipher3或sqlcipher3库但安装依赖在部分平台上有点小折腾而且在加密模式下性能有折扣应用层加密只把敏感字段比如价格、电话在写入前加密读取时解密。用cryptography库就能做到不需要换数据库操作系统权限控制把.db文件放到只有程序账号可读写的目录对多用户共享的 Windows 电脑也算基本保障对小型商业系统我的建议是别一上来就想着加密数据库文件。先把备份做扎实、权限管好、操作日志留存好基本的安全要求就满足了。如果确实有合规层面的加密需求再评估 SQLCipher 的迁移成本——否则别让加密这件事拖慢你的开发进度。6. 这套系统还能往哪些方向扩展6.1 从单机到多终端三个需要留意的点当店面扩大、需要多个收银台或让管理电脑和前台电脑同时操作时SQLite 的写并发限制会被放大。这时候有几个平缓的过渡方案读写分离把报表统计、库存查询这类只读操作指向定期同步的副本库主库只服务日常开单写入换用文件型替代方案比如用 PostgreSQL 或 MySQL 替代 SQLite你的表结构和大部分 SQL 语句可以相对平滑地迁移核心工作在于替换数据访问层的实现上轻量服务化框架用 FastAPI 包一层 REST API前端用浏览器访问后端数据库仍然是 SQLite。这个方案对小型局域网内的多用户场景很友好而且 SQLite 依然能扛得住我的经验是在业务逻辑层和数据访问层之间留出接口比如一个get_connection函数日后换数据库时全局改动点会少得多。6.2 这个阶段值得加的实用功能回看整个项目有几个功能是性价比极高的增强Excel 导入/导出用openpyxl实现商品批量建档、销售单导出月报能省去大量手工录入时间条码/二维码扫码收银台配一个扫码枪按 SKU 直接开单效率提升非常明显操作日志表记录谁在什么时间改了什么库存对多人共用的场景不可或缺简易权限管理员和收银员不同权限收银员只能开销售单不能改商品价格这些功能加到现有骨架上都不需要对表结构伤筋动骨逐步迭代非常顺畅。6.3 什么时候应该考虑换掉 SQLite这个问题值得认真回答。我的判断依据是看三个信号并发写请求达到每秒几十次以上SQLite 的全局写锁会开始频繁报警数据量增长到 GB 级别查询性能开始明显下降业务走向多仓库、多门店、云部署需要实时同步、权限体系、故障恢复等能力只要还在这些问题之外SQLite 依然是性价比极高的选择。它不是学习用玩具它真的能稳定支撑一个小型商业体每天的真实运转。这套系统做到今天最让我有成就感的不是写了多少代码而是帮朋友把一个靠纸和 Excel 硬撑的门店变成了打开电脑就知道该进什么货、上个月赚多少钱、哪件商品周转最快的数字化小店。如果你也想为自己的场景搭一套类似的工具我的建议是别一上来就追求大而全先保证进货、销售、库存流水三条主链路跑通再一步步叠加需求。数据结构稳住了后面的功能都是水到渠成的事。最后分享一个从这次项目中沉淀下来的习惯每写一张表、每封装一个函数都顺手把设计理由记在注释里。三个月后再回来改代码的时候你会感谢当初那个认真写注释的自己。 Python+SQLite轻量进销存系统实战：从表设计到事务处理全解析
前阵子帮一个开小五金店的朋友搭了一套进销存系统需求听起来简单落地时各种细节却相当磨人进货单要能追到每一批货的供应商销售单要能对应到具体客户库存一改就得留下痕迹月底对账最好半小时搞定而不是熬一晚上。最终敲定的方案就是标题里这套——Python SQLite。没有用 MySQL没有搞前后端分离没有引入 Redis。整套系统跑在一台普通 Windows 电脑上数据库就是一个.db文件。从需求分析到第一版跑通花了不到一个周末后续加报表、加库存预警、加操作日志也都在这套骨架上稳步迭代。这篇文章就把整个项目拆开来复盘一遍。适合正在考虑给自己店里、Small 仓库或者个人项目搭一套轻量管理系统的朋友参考也适合刚学 Python 想拿真实场景练手的人。我不会只贴代码更想把每个设计决策背后的为什么讲清楚因为这些取舍才是这套方案真正的价值所在。1. 为什么选择 Python SQLite 这套轻量组合1.1 小型商业进销存的真实痛点朋友的那家五金店规模不大旺季一天出货也就百来单。之前一直是纸质单据加 Excel 表格配着来问题很明显商品 SKU 一多Excel 里改价、改库存特别容易串行一个格子填错整行数据就乱了进货价、售价、供应商、客户信息分散在好几张表格里想查一件商品的毛利得手动拼接Excel 文件传来传去版本不一致时间一长根本不知道哪个是最新的库存没有预警螺丝钉这类小件货什么时候断货完全凭感觉这类场景有一个共同特征数据量不大但是业务关系和流程环节一点不少。一张商品表解决不了问题需要一套真正的数据结构来承载进货→库存→销售→统计这条链路。1.2 SQLite 能扛住什么样的并发量很多人在选型时会有个刻板印象SQLite 就是个嵌入式玩具不能用于商业项目。这个判断得拆开看。SQLite 的定位是嵌入式关系型数据库数据存到一个独立文件里不需要单独安装服务端进程。它支持标准 SQL、支持事务 ACID、支持外键和索引。对于每天几十到几百笔订单的场景它的性能完全够用——瓶颈通常不在数据库而在业务逻辑代码写得好不好。从并发角度讲SQLite 默认支持多进程读单进程写。也就是说如果只是收银台一台电脑操作或者两三台电脑同时读取配合 WAL 模式体验相当流畅。它的限制主要体现在高并发写入比如几十个终端同时开单——那确实不是它的战场但也不是这类小系统会遇到的场景。选择 SQLite 还有一个巨大的隐性红利部署和备份都极其简单。整个数据库就是一个.db文件拷贝走就是备份拷到新电脑就能继续跑。不需要专门安装数据库服务不需要配置账号权限。对非技术背景的店主来说这比帮你装个 MySQL友好得多。我后来的经验是把 SQLite 理解为文件型数据库而不是弱化版数据库。它在自己适合的领域内表现得非常专业你只需要尊重它的边界。1.3 Python 在项目里的角色Python 在这套系统里承担三个角色业务逻辑层、数据访问层和界面层。业务逻辑层负责进销存规则比如销售出库时库存必须足够入库时自动更新成本价退货时恢复库存。数据访问层负责把 Python 对象和 SQLite 表互相转换。界面层用 Python 自带的标准库就能做控制台版本也可以后续换成 PyQt 或 Web 界面。选择 Python 的深层原因在于它的开发效率和生态。整表操作、事务管理、Excel 导入导出、报表生成都有现成库。比起 Java 和 C# 的工程化成本Python 更适合这套系统的规模和迭代节奏。2. 数据库表结构设计进销存系统的数据骨架2.1 从业务环节反推表设计进销存的核心链路就三个字进、销、存。进从供应商采购商品入库销向客户销售商品出库存实时掌握每个 SKU 的库存数量围绕这条链路我设计了八张核心表覆盖了从单据到明细到库存流水的完整闭环表名作用关键字段products商品档案sku, name, category, spec, unit, purchase_price, sale_price, min_stock, stocksuppliers供应商档案name, contact, phone, address, remarkcustomers客户档案name, contact, phone, address, remarkpurchase_orders进货单主表order_no, supplier_id, total_amount, order_date, status, remarkpurchase_items进货单明细order_id, product_id, quantity, unit_price, amountsales_orders销售单主表order_no, customer_id, total_amount, order_date, status, remarksales_items销售单明细order_id, product_id, quantity, unit_price, amountstock_records库存变动流水product_id, change_type, change_quantity, before_stock, after_stock, related_order_no订货单和销售单拆成主表和明细表两张是这套设计的核心。主表记录这一单跟谁做的、总金额多少、日期是哪天明细表记录这一单包含了哪些商品、各自多少数量多少钱。这样既能按单号查全貌也能按商品反查它进出过哪些单子。2.2 库存改动必须留痕真正让我坚持加stock_records这张流水表的是一次对账事故。有一次朋友反映库存对不上账面上剪刀少了 5 把。但系统里没有任何地方记录了这次变动是怎么发生的。单查销售单没卖过单查进货单也没进过。最后发现是盘点时手动调整库存但操作时没留下痕迹。从那以后我的设计原则就变成了任何库存变动必须同时写一条流水。不管是通过进货单增加库存还是通过销售单扣减库存甚至盘点修正都必须记录变动前后的数值、变动类型和对应的单据号。有了流水表之后对账逻辑就变成了期末库存 期初库存 进货总量 - 销售总量 - 报损量。每一项都能在流水表中找到对应记录再也不用猜库存去哪了。2.3 建表脚本中的关键细节-- 开启外键约束SQLite 默认不启用 PRAGMA foreign_keys ON; CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT UNIQUE NOT NULL, name TEXT NOT NULL, category TEXT, spec TEXT, unit TEXT DEFAULT 个, purchase_price REAL NOT NULL DEFAULT 0, sale_price REAL NOT NULL DEFAULT 0, min_stock INTGEER NOT NULL DEFAULT 0, stock INTEGER NOT NULL DEFAULT 0, status INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL DEFAULT (datetime(now, localtime)), updated_at TEXT NOT NULL DEFAULT (datetime(now, localtime)) );有几点值得单独强调sku字段加 UNIQUE 约束同一款商品只能有一条档案重复建档会在数据库层被拦截而不是靠业务代码说我检查过了stock字段类型用 INTEGER库存是整数永远不要用浮点数存数量created_at用字符串存本地时间datetime(now, localtime)是 SQLite 的内置函数读写直观排序也没问题商品表里加status字段做逻辑删除下架商品不删行只把 status 改为 0历史单据引用它的时候不会断链2.4 SQLite 默认不开启外键约束这个坑建表脚本里第一行写了PRAGMA foreign_keys ON这一行非常关键。SQLite 为了兼容旧项目默认情况下不开启外键约束。也就是说就算你在建表语句里写了FOREIGN KEY引用默认环境下它也不会真去校验。往 sales_items 里插入一个 product_id 完全不存在的记录数据库会照单全收。这会导致什么后果日后做联表查询时会因为找不到对应商品而返回一堆空数据对账时根本找不出根因。正确做法是在每次建立数据库连接后立即执行PRAGMA foreign_keys ON。注意这条指令是连接级的不是数据库级的每次连接都要执行。如果用的是 Python 的sqlite3模块可以在连接后统一设置。conn sqlite3.connect(inventory.db) conn.execute(PRAGMA foreign_keys ON)3. 核心业务逻辑的 Python 实现3.1 统一的数据访问层刚开始写这套系统时我直接在业务代码里到处写 SQL 语句。结果项目量一上来就发现痛苦了业务规则和 SQL 混在一起改一个库存字段要翻遍所有涉及的地方而且冗余代码多到怀疑人生。后来我抽了一个db.py模块做统一的数据访问层把所有通用的 CRUD 和数据库连接管理集中在这里import sqlite3 DB_PATH inventory.db def get_connection(): conn sqlite3.connect(DB_PATH, timeout10) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) return conntimeout10的意思是当数据库文件被其他连接锁住时当前连接最多等 10 秒而不是直接报错。row_factory sqlite3.Row让查询结果可以通过字段名访问row[sku]比row[0]可读性好太多。3.2 入库与出库事务的原子性进销存系统最核心的规则是单据和库存变动必须同时成功或者同时失败。比如销售出库程序要做三件事在 sales_orders 插入主表记录在 sales_items 插入明细记录把 products 表的库存减掉这三件事是一个整体。如果第 3 步失败而前两步已经提交系统就会出现单据显示卖了库存却没扣的脏数据。解决这个问题靠的就是数据库事务。Python 的sqlite3模块在开启事务时有个行为细节。默认情况下如果在autocommit模式之外执行INSERT/UPDATE/DELETEPython 会自动开启事务。真正容易踩坑的是并发环境下默认的BEGIN延迟事务可能在读操作后才升级为写锁导致死锁或锁冲突。我的处理方式是显式声明BEGIN IMMEDIATE。这样一开始就拿到写锁避免中途升级锁带来的复杂度def create_sale_order(order_no, customer_id, items): 创建销售单并扣减库存 items 格式: [{product_id: 1, quantity: 2, unit_price: 15}, ...] conn get_connection() cursor conn.cursor() try: # IMMEDIATE 事务立刻获取写锁避免并发下的锁升级问题 conn.execute(BEGIN IMMEDIATE) total_amount 0 for item in items: product_id item[product_id] quantity item[quantity] unit_price item[unit_price] amount quantity * unit_price total_amount amount # 查出当前库存 product cursor.execute( SELECT stock FROM products WHERE id ?, (product_id,) ).fetchone() if product is None: raise ValueError(f商品 {product_id} 不存在) if product[stock] quantity: raise ValueError(f商品 {product[id]} 库存不足当前库存 {product[stock]}) # 扣减库存并记录变动前的库存值 cursor.execute( UPDATE products SET stock stock - ?, updated_at datetime(now, localtime) WHERE id ?, (quantity, product_id) ) cursor.execute( INSERT INTO stock_records (product_id, change_type, change_quantity, before_stock, after_stock, related_order_no) VALUES (?, sale, ?, ?, ?, ?), (product_id, quantity, product[stock], product[stock] - quantity, order_no) ) # 写销售单主表 cursor.execute( INSERT INTO sales_orders (order_no, customer_id, total_amount, order_date, status) VALUES (?, ?, ?, datetime(now, localtime), 1), (order_no, customer_id, total_amount) ) order_id cursor.lastrowid # 写销售单明细 for item in items: cursor.execute( INSERT INTO sales_items (order_id, product_id, quantity, unit_price, amount) VALUES (?, ?, ?, ?, ?), (order_id, item[product_id], item[quantity], item[unit_price], item[quantity] * item[unit_price]) ) conn.commit() return order_id except Exception: conn.rollback() raise finally: conn.close()这段代码里有几个值得称道的细节先查库存再扣减两次操作在同一个事务内别怕中间的并发修改——因为有写锁挡着记录before_stock和after_stock流水表里既有变动量也有变动前后绝对值日后回查非常有用用conn.rollback()包裹异常任何一步失败前面所有操作全部撤销订单号由外部传入保证了调用方可以统一生成有规则的单号含日期、序列号后续查询追踪就方便了3.3 入库逻辑不仅加库存还要维护价格信息入库相对简单但有一个容易忽略的商业细节进货价格是会波动的。同一种商品这次进价 10 元下次可能涨到 11 元。如果只简单累加库存不更新purchase_price那么过一段时间你会发现库存成本完全失真。对小型批发零售场景我采用最新进货价覆盖成本价的策略def create_purchase_order(order_no, supplier_id, items): conn get_connection() cursor conn.cursor() try: conn.execute(BEGIN IMMEDIATE) total_amount 0 for item in items: product cursor.execute( SELECT stock, purchase_price FROM products WHERE id ?, (item[product_id],) ).fetchone() if product is None: raise ValueError(商品不存在) quantity item[quantity] unit_price item[unit_price] cursor.execute( UPDATE products SET stock stock ?, purchase_price ?, updated_at datetime(now, localtime) WHERE id ?, (quantity, unit_price, item[product_id]) ) cursor.execute( INSERT INTO stock_records (product_id, change_type, change_quantity, before_stock, after_stock, related_order_no) VALUES (?, purchase, ?, ?, ?, ?), (item[product_id], quantity, product[stock], product[stock] quantity, order_no) ) total_amount quantity * unit_price cursor.execute( INSERT INTO purchase_orders (order_no, supplier_id, total_amount, order_date, status) VALUES (?, ?, ?, datetime(now, localtime), 1), (order_no, supplier_id, total_amount) ) conn.commit() except Exception: conn.rollback() raise finally: conn.close()这样入库单、库存、成本价都保持同步。经营分析时那一列毛利 销售额 - 销售数量 × 最新成本价才能算得准。3.4 查询与预警让数据派上用场数据录进去了还得能高效查出来。日常最频繁的三个查询场景是按分类统计库存数量conn get_connection() rows conn.execute( SELECT category, SUM(stock) AS total_stock, COUNT(*) AS sku_count FROM products WHERE status 1 GROUP BY category ORDER BY total_stock DESC ).fetchall()库存预警warnings conn.execute( SELECT sku, name, stock, min_stock FROM products WHERE status 1 AND stock min_stock ORDER BY stock ASC ).fetchall()这两个查询看起来平平无奇但它们是整套系统决策价值的体现。我后来直接把预警写入日志表每天开店时自动打印一份应补货清单朋友说光是这个功能就值回开发成本。4. 开发与调试过程中踩过的坑4.1 database is locked 到底怎么回事第一次遇到这个错误是在一次导入测试数据的时候程序跑批处理连续插入几千条记录中途跳了sqlite3.OperationalError: database is locked。排查过程让我确认了一个关键认知SQLite 的锁粒度是整个数据库文件不是某一行或某一张表。这意味着任何一个连接持有写锁其他连接的写操作都会等待或者直接报错。当多个进程或者一个进程内多个连接同时写数据库时锁冲突是很容易出现的。解决办法有三个层次第一个层次是加重试机制。用timeout10参数让连接等待锁释放第二个层次是缩短单事务的时间。不要在事务里做网络请求或长时间计算快进快出第三个层次是引入 WAL 模式。Write-Ahead Logging 允许读操作和写操作并发写事务之间仍然互斥但读写混跑的场景会顺滑很多conn.execute(PRAGMA journal_mode WAL;)注意WAL 模式会生成额外的.db-wal和.db-shm文件备份时要连它们一起拷贝或者先执行一次 checkpoint。这一点我在后面备份部分会再次提到。4.2 金额计算别用浮点数这是我早期代码里最早埋下的雷。当时unit_price字段用的REALPython 侧直接quantity * unit_price算金额。表面看起来没什么问题直到某天发现销售单汇总金额和 Excel 里手算的对不上差了几分钱。原因在于浮点数的二进制表示不精确。比如0.1在浮点数里存得其实是一个无限循环的近似值多次运算后误差就会被放大。解决方案也很直接金额统一用分为单位存整数INTEGER。数据库表里该字段可以命名为purchase_price_cents、sale_price_cents业务层再除以 100 转成元展示。# 正确做法金额以分为单位 UPDATE products SET sale_price_cents ? WHERE id ?, (1999, product_id) # 19.99 元 # 而不是 UPDATE products SET sale_price ? WHERE id ?, (19.99, product_id)这个习惯我到现在都保留着凡是涉及钱和数量的字段能用整数绝不用浮点数。宁可改表结构时多写几行也不去趟浮点误差的浑水。4.3 多线程下的 check_same_thread 陷阱后来系统加了 PyQt 界面SQLite 操作放到了子线程里程序运行几分钟后就开始报错sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread.这是sqlite3模块为了线程安全做的一个保护措施。默认情况下一个连接在哪个线程创建就只能被哪个线程使用。当时手足无措第一反应是搜到有人说加check_same_threadFalse就行了。确实能绕过这个限制但绕过不等于安全——如果多个线程同时使用同一个连接写数据底层锁冲突依然存在很可能出现不可预期的数据错乱。我最终采用的是**每线程独立连接 连接池**的折中方案业务层不共享连接每次操作通过函数获取独立连接用完即关。对于这套小系统的并发量完全够用还省去了 thread-local 管理的复杂度。import threading _local threading.local() def get_thread_connection(): conn getattr(_local, conn, None) if conn is None: conn sqlite3.connect(DB_PATH, timeout10) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) conn.execute(PRAGMA journal_mode WAL;) _local.conn conn return conn这个方案的好处是每个线程都有自己的专用连接不需要加锁而 WAL 模式保证了跨线程读写不互相阻塞。4.4 不要在主线程里跑批量导入顺带说一个相关经验。给朋友迁移 Excel 旧数据时几千行记录逐条 INSERT界面直接卡死。原因很明显主线程忙得没空处理界面刷新。正确做法是把耗时的数据导入放到后台线程里跑同时主线程隔一段时间刷新进度条。如果你是用控制台版至少也加一个进度打印别让用户以为程序死了。5. 日常运维备份、数据安全与可视化工具5.1 用 DB Browser for SQLite 做可视化管理系统跑起来之后我强烈建议装上DB Browser for SQLite这个小工具。它是免费开源的 SQLite 可视化客户端Windows / macOS / Linux 都有安装包。它的用途不只是看看数据。日常运维中你会频繁用到它做这些事情直接浏览八张表的原始数据比在代码里写查询调试快得多执行临时 SQL比如批量修正错误分类、手动调整库存导出 CSV / Excel在不写代码的前提下把数据交给店主用 Excel 二次分析查看数据库结构和索引帮你验证建表脚本执行后到底建成了什么样我见过不少人折腾半天sqlite3命令行去查数据效率极低。可视化工具虽然简单但它能让你把精力花在业务判断上而不是敲命令上。5.2 备份策略比拷贝文件再多做一步SQLite 备份最直觉的做法是直接复制.db文件。如果程序恰好没有打开这个数据库直接拷贝是有效的但如果程序正在运行中直接复制文件可能复制到不一致的状态尤其是开启了 WAL 模式后-wal文件里那些尚未合并进主库的数据可能丢失。我推荐用 Python 自带的备份 API 做在线备份import sqlite3 def backup_db(src_db_path, dst_db_path): src_conn sqlite3.connect(src_db_path) dst_conn sqlite3.connect(dst_db_path) try: src_conn.backup(dst_conn) print(f备份完成: {dst_db_path}) finally: dst_conn.close() src_conn.close()sqlite3.Connection.backup()是官方推荐方式支持在线备份多个连接同时读取时也能保证一致性。定时任务里每天凌晨跑一次把库备份到移动硬盘或者云盘同步目录就已经非常稳妥了。记得做一次备份恢复演练。我曾经顺顺利利备份了大半个月直到朋友有一天误删了一张表想恢复才发现备份文件本身就有问题。定期用备份文件恢复到临时目录跑一次查询几分钟的事能少掉一晚上的慌乱。5.3 关于 SQLite 文件加密的实话实说热搜词里有个sqlite数据库文件能否加密这里明确讲一下。SQLite 官方版本不提供原生加密。如果有人直接拿.db文件就能用文本编辑器看到里面的明文数据。想给 SQLite 加密码常规手段有几个SQLCipher它是 SQLite 的加密分支对数据库文件做 AES 加密。Python 接入的话要用pysqlcipher3或sqlcipher3库但安装依赖在部分平台上有点小折腾而且在加密模式下性能有折扣应用层加密只把敏感字段比如价格、电话在写入前加密读取时解密。用cryptography库就能做到不需要换数据库操作系统权限控制把.db文件放到只有程序账号可读写的目录对多用户共享的 Windows 电脑也算基本保障对小型商业系统我的建议是别一上来就想着加密数据库文件。先把备份做扎实、权限管好、操作日志留存好基本的安全要求就满足了。如果确实有合规层面的加密需求再评估 SQLCipher 的迁移成本——否则别让加密这件事拖慢你的开发进度。6. 这套系统还能往哪些方向扩展6.1 从单机到多终端三个需要留意的点当店面扩大、需要多个收银台或让管理电脑和前台电脑同时操作时SQLite 的写并发限制会被放大。这时候有几个平缓的过渡方案读写分离把报表统计、库存查询这类只读操作指向定期同步的副本库主库只服务日常开单写入换用文件型替代方案比如用 PostgreSQL 或 MySQL 替代 SQLite你的表结构和大部分 SQL 语句可以相对平滑地迁移核心工作在于替换数据访问层的实现上轻量服务化框架用 FastAPI 包一层 REST API前端用浏览器访问后端数据库仍然是 SQLite。这个方案对小型局域网内的多用户场景很友好而且 SQLite 依然能扛得住我的经验是在业务逻辑层和数据访问层之间留出接口比如一个get_connection函数日后换数据库时全局改动点会少得多。6.2 这个阶段值得加的实用功能回看整个项目有几个功能是性价比极高的增强Excel 导入/导出用openpyxl实现商品批量建档、销售单导出月报能省去大量手工录入时间条码/二维码扫码收银台配一个扫码枪按 SKU 直接开单效率提升非常明显操作日志表记录谁在什么时间改了什么库存对多人共用的场景不可或缺简易权限管理员和收银员不同权限收银员只能开销售单不能改商品价格这些功能加到现有骨架上都不需要对表结构伤筋动骨逐步迭代非常顺畅。6.3 什么时候应该考虑换掉 SQLite这个问题值得认真回答。我的判断依据是看三个信号并发写请求达到每秒几十次以上SQLite 的全局写锁会开始频繁报警数据量增长到 GB 级别查询性能开始明显下降业务走向多仓库、多门店、云部署需要实时同步、权限体系、故障恢复等能力只要还在这些问题之外SQLite 依然是性价比极高的选择。它不是学习用玩具它真的能稳定支撑一个小型商业体每天的真实运转。这套系统做到今天最让我有成就感的不是写了多少代码而是帮朋友把一个靠纸和 Excel 硬撑的门店变成了打开电脑就知道该进什么货、上个月赚多少钱、哪件商品周转最快的数字化小店。如果你也想为自己的场景搭一套类似的工具我的建议是别一上来就追求大而全先保证进货、销售、库存流水三条主链路跑通再一步步叠加需求。数据结构稳住了后面的功能都是水到渠成的事。最后分享一个从这次项目中沉淀下来的习惯每写一张表、每封装一个函数都顺手把设计理由记在注释里。三个月后再回来改代码的时候你会感谢当初那个认真写注释的自己。