ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

基于Python的ERP系统源码指南:库存并发与事务处理实战

基于Python的ERP系统源码指南:库存并发与事务处理实战 简介基于Python语言的企业资源规划系统设计源码面向企业信息化建设者、Python开发人员及企业资源规划学习者展示了从业务流程建模到系统落地的完整设计思路。压缩包共一百七十个文件大小约七点六三兆字节以一百六十个Python源文件为核心覆盖权限控制、数据处理与业务模块协作同时包含四个可扩展标记语言配置文件用于数据库连接、第三方服务接入和界面布局另有轻量级SQLite数据库、说明文档、Git忽略文件、日志与开发环境项目文件等辅助内容整体目录结构清晰便于按功能模块查阅。目前已有六百一十六人学习浏览适合初入企业级开发的学习者对照研究也可作为小型企业资源规划项目二次开发的基础。借助这套源码读者能够深入理解订单、产品序列化、支付与回执等典型模块的编码思路并借助数据库文件和配置文件掌握前后端数据流转方式对系统设计和Python工程化能力提升很有价值。1. 基于Python语言的ERP系统设计源码一份能改得动、跑得起来的业务骨架在网上下载过“基于Python语言的ERP系统设计源码”这个包的人第一反应多半是两种要么觉得项目文件夹多得吓人要么对着 README 里那句“功能仅作学习参考”发呆。这份源码的本质是一个用 Python 搭起来的企业资源计划最小骨架——它覆盖采购、销售、库存、财务这些核心业务线但通常不做复杂权限、不做多级审批、也不承诺高并发。它的价值不在“开箱即用”而在“你改得动”。对 Python 入门者这是一份综合练习题对小团队它是一套可以快速私有化改造的房子而不是装修好的样板间。这篇笔记会从源码结构讲到跑起来、改业务、避坑最后用测试把改动兜住。2. 看懂ERP源码的业务骨架三层架构与六条关键业务线2.1 为什么Python能承担ERP别只盯着性能短板传统ERP市场长期被 Java 和 .NET 把持原因很简单大型企业的并发量、事务要求、权限复杂度Python 默认实现确实吃力。但一套给几十人使用的小型ERP瓶颈从来不在语言本身。Python 做 ERP 的真实优势是开发速度和生态ORM 省掉大量 SQL 样板代码报表可以直接用 pandas 聚合对接供应商 Excel 有 openpyxl甚至接入爬虫抓价格表也顺手。选 Python 搭 ERP 要接受三个前提一是 CPU 密集场景尽量用数据库层解决别把循环放到 Python 里二是类型安全靠 pydantic 这类库补别裸写 dict 到处传三是真正高并发时换异步框架或干脆拆服务。看清这三点Python 在中小规模 ERP 里的性价比是明显高于 Java 的。我接手过一套用 Flask 写的进销存三个人改了两个月就把采购审批、库存预警、对账报表全部跑通放 Java 里光配 Spring 那一套就要吃掉不少工期。2.2 源码包里的常见布局从入口文件到业务模块网上下载的 Python ERP 源码不管是 Django 还是 Flask 系目录布局多半逃不出下面这个模式。先花十分钟认路比直接跑 python main.py 有用得多。erp_source/ ├── manage.py # Django 项目入口Flask 则换成 app.py ├── requirements.txt # 依赖清单 ├── config/ │ ├── settings.py # 全局配置 │ └── database.py # 数据库连接 ├── apps/ │ ├── products/ # 商品档案 │ ├── purchase/ # 采购管理 │ ├── sales/ # 销售管理 │ ├── inventory/ # 库存管理 │ └── finance/ # 应收应付 ├── common/ │ ├── models.py # 公共模型 │ ├── transaction.py # 事务封装 │ └── response.py # 统一返回格式 └── scripts/ ├── init_data.py # 初始化数据 └── daily_report.py # 日结脚本认路要点就三条config 目录决定了环境apps 目录决定了业务边界scripts 目录决定了运维方式。最关键的判断是业务模块怎么分层——成熟一点的源码会在每个 apps 子目录里再拆 models.py、services.py、views.py把数据定义、业务规则、请求处理分开。如果一份源码把所有逻辑都堆在 views.py 里那它更像是课程设计而不是工程样板改造时要做好重构准备。调用链一般是 views 收参数 - services 做业务校验和计算 - models 通过 ORM 操作数据库顺序别搞反。2.3 ERP系统业务流程从采购到应收的六条主线看懂 ERP 系通先看单据流转。下面这六条业务线几乎覆盖了所有小型ERP的核心源码能不能改好就看你对这些线的业务语义理解得透不透。业务线上游输入下游结果典型数据表商品档案手工录入 / Excel导入被采购、销售、库存引用Product, Category采购入库采购订单 供应商送货库存增加、应付增加PurchaseOrder, StockLog销售出库销售订单 客户付款库存扣减、应收增加SalesOrder, Receivable库存调拨仓库间转移申请两仓库存此消彼长TransferOrder应收应付销售单/采购单生成对账与收款核销Receivable, Payable经营报表以上所有单据聚合毛利、库存周转、应收账龄视图或汇总表三句经验之谈第一库存是结果不是源头任何直接改库存表的操作都是定时炸弹第二单据编号必须唯一且有业务含义来源单号要能串起来——从销售单能一路查到采购单第三金额字段的精度问题一开始就要用 Decimal后面处理巨麻烦。很多惊艳的“免费 Python 源码大全”里下的 ERP业务线都不完整最常见的缺失是应收应付只做了个表头没有核销逻辑改造前先对着这张表盘一遍缺口。3. 把源码跑起来Python环境、数据库与首次启动配置3.1 Python版本、虚拟环境与VSCode解释器选择跑这套源码前先定版本不要直接用系统里最贵的那个 Python。多数 ERP 源码基于 Django 或 FlaskDjango 4.2 LTS 支持 Python 3.8 到 3.12但第三方依赖和 MySQL 驱动往往在 3.12 上慢半拍。我的习惯是装 Python 3.10兼容性足够数据库驱动基本都跟上也不像 3.12 那样偶发一些二进制包编译问题。安装教程里常提到的“勾选 Add to PATH”只是第一步真正要命的是解释器选错。cd erp_source python3.10 -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate python -V # 确认输出是 Python 3.10.x pip install --upgrade pip pip install -r requirements.txt在 VSCode 里配置 Python 环境时按 CtrlShiftP 选 Python: Select Interpreter必须选到刚才创建的 .venv 路径。这个坑我踩过系统全局装着一套 PythonVSCode 默认选了它pip list 能看到 Djangoimport 却报错——因为 VSCode 用的是另一个解释器。核对办法很简单在 VSCode 终端里执行 which python路径里必须包含 .venv。requirements.txt 如果源码里没给或者给得不完整手工补一个最小集合Django/Flask 按框架选ORM 用 SQLAlchemy 或 Django ORM再加 openpyxl、pandas、python-dotenv。注意版本号建议用范围锁别用最新大版本冲刺一个 unpinned 的依赖升级就能让整个项目哑火。3.2 数据库配置分离SQLite先跑通生产切MySQL拿到源码先看数据库配置写在哪。常见做法是把连接信息抽到环境变量文件里config/database.py 读环境变量没有就用 SQLite 兜底。这个设计值得学开发时零成本跑通部署时改几个环境变量就能切 MySQL。# config/database.py import os from dotenv import load_dotenv load_dotenv() # 读取项目根目录的 .env 文件 DB_ENGINE os.getenv(DB_ENGINE, sqlite) # 默认 sqlite DB_NAME os.getenv(DB_NAME, erp.db) DB_USER os.getenv(DB_USER, erp) DB_PASSWORD os.getenv(DB_PASSWORD, ) DB_HOST os.getenv(DB_HOST, 127.0.0.1) DB_PORT os.getenv(DB_PORT, 3306) if DB_ENGINE mysql: DATABASE_URL fmysql://{DB_USER}:{DB_PASSWORD}{DB_HOST}:{DB_PORT}/{DB_NAME} else: DATABASE_URL fsqlite:///{DB_NAME}DB_ENGINE 这个参数是最关键的切换开关源码里如果写死了 SQLite 路径生产环境迟早要改回来。SQLite 适合单人开发和演示一旦多人同时开单就会频繁遇到锁问题这个在第 5 章会展开讲。迁移到 MySQL 时注意 Django 要装 mysqlclientFlaskSQLAlchemy 要装 PyMySQL两者安装都有系统依赖Ubuntu 下先 sudo apt install default-libmysqlclient-dev 再 pip install 才是正确的顺序。3.3 初始化数据与首次登录把空表变成可操作单据跑起来只是第一步把空数据库填上能用的初始数据才算真正打通。源码里一般会有 init_data.py 或 migrations 里的 data migration没有就自己写一条命令。python manage.py makemigrations python manage.py migrate python manage.py shell -c from scripts.init_data import run; run() python manage.py runserver 0.0.0.0:8000makemigrations 把模型变更生成迁移文件migrate 把迁移写到数据库里这两步顺序错了就会提示表不存在。init_data 里通常做三件事创建超级用户、写入基础商品分类、生成一张测试采购单。登录后首先要改默认密码很多课程设计源码默认 admin/admin123 不带验证码局域网里被扫到就是分分钟的事。初启动后验证三件事能用默认账号登录、商品列表有数据、新建一张销售单后库存数发生变化。这三条通了说明源码的模型、视图、事务封装基本是自洽的。跑的时候留意启动日志里有没有 warning特别是关于 naive datetime 的提示那个是时区坑的前兆。4. 把骨架改成可用系统库存扣减、单据事务与经营报表4.1 ERP库存场景高并发的解决方案悲观锁与乐观锁怎么选小型 ERP 最容易翻车的业务就是库存扣减。“用户A和用户B同时下单各自查到库存还有10件都扣了一单结果库存变8而不是9”——这不是数据库坏了是并发读写的经典竞态。解决思路就是两条路悲观锁和乐观锁。先看悲观锁核心是用 select_for_update 把目标行锁住锁释放前其他事务的同类查询会等待。from django.db import transaction transaction.atomic def deduct_stock(sku_id, quantity): # select_for_update 必须在事务内使用锁到的是数据库行级锁 product Product.objects.select_for_update().get(skusku_id) if product.stock quantity: raise InsufficientStock(f{sku_id} 库存不足) product.stock - quantity product.save(update_fields[stock, updated_at]) StockLog.objects.create(skusku_id, change-quantity, reasonSALE)select_for_update 的语义是当前事务不提交其他任何事务想改这行都被阻塞。参数上注意两点nowaitTrue 时锁冲突直接报错而非等待适合需要快速失败的场景配合事务隔离级别使用效果最佳MySQL 默认的 REPEATABLE READ 下锁机制是可靠的。这个方案代码好写、语义清晰适合库存记录竞争不极端的情况。乐观锁不锁行而是靠 update 语句带条件判断来保证原子性from django.db.models import F # 一次 update 同时完成“库存够才扣”和“扣减”两个动作天然原子 updated Product.objects.filter( skusku_id, stock__gtequantity ).update(stockF(stock) - quantity) if updated 0: # 说明库存不足或商品不存在此时再查库给出友好提示 raise InsufficientStock(f{sku_id} 库存不足或商品已下架)乐观锁的优点是并发性能高、不占连接缺点是库存充足但频繁冲突时上层要做重试。ERP 场景我的建议是日单量在几千以下直接上悲观锁简单不容易错到了几十万再考虑乐观锁加 Redis 预扣。相比分布式锁这两种数据库方案都更务实也是“ERP库存场景高并发的解决方案”里最常见的落地姿势。4.2 创建销售单据事务边界画在库存变动那一行库存扣减是单点操作创建单据是多表操作两者必须在一个事务里。常见的错误是把“生成销售单”和“扣库存”分在两个接口里调用结果客户付了款、库存没扣或者库存扣了单据没生成。正确做法是用一个事务把多步写操作包起来。transaction.atomic def create_sales_order(order_no, customer_id, items): # 1. 创建订单头 order SalesOrder.objects.create( order_noorder_no, customer_idcustomer_id, statusPENDING ) # 2. 创建订单明细同时累加金额 total Decimal(0.00) for item in items: line SalesItem.objects.create( orderorder, skuitem[sku], qtyitem[qty], priceitem[price] ) total line.price * line.qty deduct_stock(item[sku], item[qty]) # 复用上一步的扣减逻辑 # 3. 创建应收记录 Receivable.objects.create( orderorder, amounttotal, statusUNPAID ) return order事务边界的核心原则一笔业务涉及的写操作全部放进去外部调用和耗时操作不放进来。这里 dedect_stock 内部已经用了 select_for_update嵌套在 create_sales_order 的事务里时锁的释放随外层事务走不会提前释放。注意逐行扣库存的性能问题如果一张单有几百条明细每条都 select_for_update 一次会拖慢事务。优化办法是先一次性把涉及的库存记录查出来用 dict 缓存住再逐条扣减。事务函数外层一定要捕获异常回滚。Django 的 transaction.atomic 在函数抛出异常时自动回滚但要注意捕捉异常的地方不要在函数内部否则回滚逻辑可能被吞掉。我的习惯是 services 层抛业务异常views 层统一捕获并返回错误信息。4.3 报表模块用查询和Pandas把数据变成经营看板业务单据能正常流转之后下一步自然就是看数据。报表模块我一般分两层做底层用 ORM 聚合查询把原始数据变成结构化结果上层用 pandas 做二次加工前端展示用简单的 ECharts 或图表库不必一开始就上 Heavy 的 BI 工具。from datetime import date from django.db.models import Sum, F import pandas as pd # 当日销售明细聚合按商品汇总数量和金额 today date.today() daily_sales ( SalesItem.objects .filter(order__created_at__datetoday) .values(sku, sku__name) # sku__name 表示关联 Product 表的 name 字段 .annotate( total_qtySum(qty), total_amountSum(F(qty) * F(price)) ) .order_by(-total_amount) ) # 转成 DataFrame 便于后续做趋势对比和导出 df pd.DataFrame(list(daily_sales)) if not df.empty: df[占总金额比] df[total_amount] / df[total_amount].sum() df.to_excel(fdaily_sales_{today}.xlsx, indexFalse)这个写法里 values 和 annotate 的组合是固定的套路values 指定分组维度annotate 指定聚合指标。F(qty) * F(price) 表示在数据库层做乘法避免把数据拉回 Python 再算几万行明细时性能差距明显。报表模块最容易翻车的是时区如果你按“当天”过滤而后台是 UTC 时间晚上 8 点以后下的单会被记到第二天。解决办法是数据库里统一存 UTC查询过滤前先转成本地时区的日期边界再做范围过滤而不是直接按日期等值匹配。报表这种读操作性能不够时先加索引再不行做汇总表。ERP 里最常用的索引就是外键字段和日期字段Django 里在模型字段上 db_indexTrue 声明即可。不要一上来就上缓存缓存的失效逻辑在 ERP 里比查询本身难写十倍。5. 基于Python的ERP源码避坑记录五条花钱买来的教训5.1 依赖装完仍报ImportError现象pip install -r requirements.txt 全程没报错启动后 Python 却说 ModuleNotFoundError: No module named django。原因终端里激活了虚拟环境但启动项目时用了 IDE 或系统级解释器。最常见于 VSCode 用户右下角解释器还是全局的Terminal 却已经 source 了 venv两边各跑各的。另一类原因是 requirements 里没锁版本Django 5.x 装上了代码里还用的是 4.x 的写法某些 API 在启动阶段才报错。解决启动前先敲 which python 和 python -V 确认解释器路径requirements.txt 里给关键依赖写版本范围例如 Django4.2,5.0。排查时用 pip list 和 pip show 看依赖实际装到哪个 site-packages然后和 sys.path 对比——如果项目根目录被加进了 sys.path而依赖装在 venv 里两者都能导入时就会遇到同名模块覆盖这种更隐蔽的问题。5.2 Excel导入单据乱码现象用 pandas 或 openpyxl 读供应商发来的商品清单中文名变成“锟斤拷”或者直接抛 UnicodeDecodeError。原因九成是 read_csv 时保留了系统默认编码Windows 下默认 GBK而文件是 UTF-8 带 BOM 格式。另一类是 Excel 文件本身混着不同编码的历史版本特别是从老财务系统导出的文件列名和内容编码不一致。解决读 CSV 一律指定 encodingutf-8-sig它能正确处理 BOM 头不确定文件编码时先做探测再读取。import chardet with open(supplier_products.csv, rb) as f: raw f.read(10000) # 只读前 10KB 足够判断大多数编码 result chardet.detect(raw) df pd.read_csv( supplier_products.csv, encodingresult[encoding] if result[encoding] else utf-8-sig, )重点不是这段代码本身而是写导入功能时必须留一个“预览”步骤用户先传文件系统读取前几行把列名和数据样例展示给用户确认然后再真正入库。这一步拦住了大量脏数据问题比任何捕获异常都好使。5.3 金额对账差了几分钱现象销售明细表汇总的金额和订单头金额对不上差 0.01 元有时候利润率算出来是 0.199999999 这种奇怪的数字。原因把金额字段定义成了 Float/Double数据库里 19.9 存成 19.899999。Python 的 float 在运算时同样有这个精度问题乘除法越做偏差越大累加到几千张单据之后对账必然差几分钱。解决金额一律用 Decimal数据库字段用 DECIMAL(12, 2)ORM 里对应 DecimalField。创建单据时金额计算全程用 Decimal不要混用 float 和 Decimal。from decimal import Decimal price Decimal(19.90) # 不要写 Decimal(19.9)那是先转 float 再转 Decimal qty Decimal(3) total price * qty # 结果是 Decimal(59.70) order_total Receivable.objects.aggregate( sSum(amount) # 数据库 DECIMAL 字段聚合的结果也是 Decimal )[s]这个坑在涉及金额的写操作和统计里都适用也常出现在 Python 量化交易策略代码里。统一的防御姿势模型里用 DecimalField业务计算用 Decimal对外序列化时再转字符串不要用 float 序列化。已经用 float 存了老数据的系统改字段类型后用 SQL 做一次四舍五入重建再补一个对账任务验证前后一致。5.4 SQLite频繁报database is locked现象开发环境一切正常部署到服务器上给三个人用每天出现几十次 OperationalError: database is locked。原因SQLite 的写入锁是数据库级的一个写事务没提交其他写事务要么等待要么直接超时。三个人的浏览器同时点“保存单据”先到的人锁住了库后到的人立刻报错。解决开发期用 SQLite 没问题真正投产前迁到 MySQL 或 PostgreSQL。如果只能用 SQLite 顶一阵把连接超时调大但这是拖延不是治疗。# settings.py 对 SQLite 的优化配置 DATABASES { default: { ENGINE: django.db.backends.sqlite3, NAME: erp.db, OPTIONS: { timeout: 30, # 等待锁的时间单位秒默认 5 秒 transaction_mode: IMMEDIATE, # 避免先拿读锁再升级写锁的死锁窗口 }, } }timeout 改成 30 只能缓解低并发下的偶发锁问题WAL 模式也建议开SQLite 开启 journal_modeWAL 后读和写不互相阻塞多读少写场景能撑久一些。但 ERP 是持续多写场景SQLite 终究是玩具。迁库不是跑一遍 migrate 就完事还要检查模型里有没有 JSON 字段这类在 MySQL 上表现不同的类型以及 CSV 导出的排序规则差异。5.5 定时任务把库存扣了两遍现象每天早上 8 点跑一次“自动处理过期未支付订单”的任务某天运维手动重跑了一次结果所有相关订单的库存都被双倍扣减。原因定时任务没有幂等设计。同一个批次被重复执行时程序没有判断批次是否已处理过直接又跑了一遍扣减逻辑。故障恢复、手动补跑、多个 worker 节点同时调度都会触发这类问题。解决给任务批次加唯一标识处理前先登记批次处理成功后再标记完成。处理过程中用事务保证批次状态和业务数据一起落库。from django.db import transaction def expire_pending_orders(batch_no): # get_or_create 保证同一批次号只会被处理一次 batch, created ProcessBatch.objects.get_or_create( batch_nobatch_no, defaults{status: RUNNING} ) if not created: return # 批次已存在说明之前已经开始处理过直接返回 orders SalesOrder.objects.filter(statusPENDING, created_at__ltcutoff) try: with transaction.atomic(): for order in orders: cancel_order(order) # 回收库存写入库存流水 order.status CANCELLED order.save() batch.status DONE batch.save() except Exception: batch.status FAILED batch.save() raise关键点在 get_or_create 和 created 的判断逻辑batch 记录最早在业务操作之前就创建那么同一批次号的第二次调用必然看到已存在记录。注意不要把“检查批次状态”和“执行业务”放成一个事务里还放错顺序——先查再干查和干之间被另一个节点插入照样会重复用唯一约束兜底才是硬保证。生产环境里我给批次表加过数据库级唯一索引效果比业务层判断更可靠。6. 用测试和幂等设计给这套ERP源码上保险最后的进阶建议6.1 三条主链路的接口测试先保住业务底线改这套源码之前先把三条最核心的接口测试补上创建商品、创建销售订单并扣库、采购入库并加库存。测试不追求覆盖率追求的是改动后第一道防线。def test_sales_order_flow(client, db): # 准备商品 resp client.post(/api/products/, { sku: A001, name: 测试商品, price: 19.90, stock: 100 }) assert resp.status_code 201 # 创建销售单数量 3应收 59.70 resp client.post(/api/sales/, { items: [{sku: A001, qty: 3, price: 19.90}] }) assert resp.status_code 201 # 断言库存扣减成功 product Product.objects.get(skuA001) assert product.stock 97 # 断言应收记录金额 receivable Receivable.objects.get(orderresp.json()[id]) assert receivable.amount Decimal(59.70)测试用例里断言的是业务结果而不是中间过程——库存余量、应收金额这两点错了说明业务逻辑被破坏。事务包裹的接口测试天然回滚数据跑完不会污染数据库这也是 Django 的 TestCase 默认行为。我给这套源码上的第一个保险就是补了这三个测试后续每次改动先跑一遍比人工点十遍页面省力得多。6.2 数据级幂等让重复提交订单不会二次扣库存接口测试防的是改坏代码幂等设计防的是运行环境的重试和误操作。前端按钮不小心点了两次、后端超时被网关重试、运维手动补单都会造成同一个业务被提交多次。import uuid def create_sales_order_api(request): request_id request.headers.get(X-Request-Id) or str(uuid.uuid4()) # 同一个 request_id 已经处理过直接返回原结果 existed SalesOrder.objects.filter(request_idrequest_id).first() if existed: return JsonResponse({order_no: existed.order_no, duplicate: True}) order create_sales_order( order_nogenerate_order_no(), customer_idrequest.data[customer_id], itemsrequest.data[items], request_idrequest_id, ) return JsonResponse({order_no: order.order_no, duplicate: False})实现上给 SalesOrder 表加一个 request_id 唯一索引前端每次提交生成一个 UUID 放在请求头里后端先查后建唯一索引兜底两个条件同时生效才能挡住并发重复。这项设计加上之后我遇到过一次手工补数据库记录导致的事务错乱排查成本从半天降到了十分钟——看 request_id 就知道是哪次请求产生的数据。6.3 后续演进接口化、语义检索与本体化跑通和改完业务线之后这套源码的下一步不是继续堆功能而是把边界理清楚。优先做两件事一是把 views 里的逻辑彻底抽到 services 层为后面前后端分离做准备二是把商品主数据做规范化让“产品编号”在采购、销售、库存、财务里完全一致——这正是企业 ERP 或 CRM 产品里 Ontology 建模要解决的事名称不统一的对账噩梦比代码崩溃难查得多。再往后走可以在商品检索上做升级传统 ERP 靠 SQL 模糊搜索想语义化就得引入向量库。本地 ERP RAG LLM 产品检索已经能落地理想状态是用户输入“耐用的红色无线鼠标”检索结果按语义排序而不是死磕关键词匹配再让大模型生成简单的库存问答。但这属于锦上添花前提是第一步的数据规范化和接口化做扎实。我记得第一个自己维护的 ERP 项目因为没做 API 层前端想换个图表库都动不了数据。现在拿到任何一套 ERP 源码第一件事永远是先看 models 和事务边界再谈别的。这个习惯帮我避开了好多次返工希望帮到你。本文还有配套的精品资源点击获取
返回列表