3个步骤搞定药柜管理系统，源码解析带你避坑

刚学完 Python 或 Java 基础语法，代码能跑通，但一面对“药柜”这种具体业务需求就脑子发懵？别慌，这是从“写代码”到“做项目”的典型断层。很多人卡在不知道如何把零散的 CRUD（增删改查）逻辑组装成一个能落地的系统。

今天咱们不整虚的，直接上干货。结合 CSDN 上不少后端开发者分享的实战经验，我拆解了一个药柜管理系统的核心逻辑。重点在于源码解析，看看真实业务中，库存扣减、权限控制、并发安全是怎么处理的。跟着做一遍，你就知道项目该怎么搭了。

项目目标与业务场景拆解

在动手写代码前，先搞清楚“药柜”到底管什么。这里的药柜，不是指医院里的智能药柜硬件，而是指医疗机构内部的药品库存管理系统。核心痛点是：药品出入库频繁、有效期敏感、批次追溯要求高。

我们需要实现四个核心功能模块：

药品基础信息维护：录入药品名称、规格、单位、供应商、有效期。
入库管理：采购入库、退货入库，必须绑定批次号。
出库管理：临床科室领用、患者发药，涉及库存实时扣减。
库存预警：低于安全库存或临近有效期时，自动触发提醒。

很多新手容易犯的错误是：把“药品”和“库存”混在一起。在数据库设计时，必须将药品字典表（静态信息）与库存批次表（动态信息）分离。药品 ID 是主键，而库存记录需要包含“药品 ID + 批次号 + 入库时间”作为联合唯一键。这样做的好处是，同一种药，不同批次可以分别管理，符合 GSP（药品经营质量管理规范）对批次追溯的要求。

目录结构规划

一个清晰的项目结构，是项目可维护性的基石。建议采用标准的 MVC（Model-View-Controller）架构，或者如果是后端 API 服务，采用分层架构。

以 Python Flask 为例，目录结构如下：

pharmacy_cabinet/
├── app/
│   ├── __init__.py          # 应用工厂
│   ├── models/
│   │   ├── __init__.py
│   │   ├── medicine.py      # 药品模型
│   │   ├── inventory.py     # 库存模型
│   │   └── transaction.py   # 出入库流水模型
│   ├── routes/
│   │   ├── __init__.py
│   │   ├── medicine_api.py  # 药品管理接口
│   │   └── inventory_api.py # 库存管理接口
│   ├── services/
│   │   └── inventory_service.py # 核心业务逻辑层
│   └── utils/
│       └── decorators.py    # 权限校验装饰器
├── tests/
│   ├── test_inventory.py    # 单元测试
│   └── test_api.py          # 接口测试
├── config.py                # 配置文件
├── run.py                   # 入口文件
└── requirements.txt

关键点解析： 注意 services 目录。很多初学者喜欢把业务逻辑全写在 routes（控制器）里，这会导致代码耦合严重，难以测试。将核心逻辑抽取到 services 层，控制器只负责参数接收和响应返回，业务层负责数据校验和状态变更。这是源码解析中最重要的架构思维转变。

核心代码实现与逐行讲解

下面重点展示最核心的出库扣减库存逻辑。这里涉及并发安全问题，是面试和实战的高频考点。

1. 数据模型定义

# app/models/medicine.py
from datetime import datetime
from app import dbclass Medicine(db.Model):__tablename__ = 'medicines'id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(100), unique=True, nullable=False)spec = db.Column(db.String(50), nullable=False) # 规格unit = db.Column(db.String(20), nullable=False) # 单位safety_stock = db.Column(db.Integer, default=0) # 安全库存阈值# 一对多关系：一种药品对应多个库存批次inventories = db.relationship('Inventory', backref='medicine', lazy='dynamic')class Inventory(db.Model):__tablename__ = 'inventories'id = db.Column(db.Integer, primary_key=True)medicine_id = db.Column(db.Integer, db.ForeignKey('medicines.id'), nullable=False)batch_no = db.Column(db.String(50), nullable=False) # 批次号quantity = db.Column(db.Integer, nullable=False) # 当前库存数量expire_date = db.Column(db.Date, nullable=False) # 有效期in_time = db.Column(db.DateTime, default=datetime.now)# 联合唯一约束，防止同一批次重复录入__table_args__ = (db.UniqueConstraint('medicine_id', 'batch_no'),)

2. 核心业务逻辑：出库扣减

这是最容易出 Bug 的地方。直接 quantity -= count 在并发场景下会导致超卖。我们需要使用乐观锁或数据库行级锁。这里演示使用乐观锁（版本号机制）的思路，虽然上面模型没加 version 字段，但在 Service 层可以通过原子更新实现类似效果。

# app/services/inventory_service.py
from app.models import Medicine, Inventory, Transaction
from sqlalchemy import and_
from datetime import datetimeclass InventoryService:@staticmethoddef deduct_stock(medicine_id: int, quantity: int, reason: str, operator: str):"""执行出库扣减采用 FEFO (First Expired First Out) 策略：先过期先出"""# 1. 查询该药品的所有库存批次，按有效期升序排列inventories = Inventory.query.filter_by(medicine_id=medicine_id).order_by(Inventory.expire_date.asc()).all()if not inventories:raise ValueError(f"药品 ID {medicine_id} 无库存记录")remaining_to_deduct = quantitytransactions_to_create = []for inv in inventories:if remaining_to_deduct <= 0:break# 如果当前批次库存不足，扣减到 0# 如果足够，扣减剩余所需数量deduct_count = min(inv.quantity, remaining_to_deduct)# 2. 原子更新库存，确保并发安全# 只有当库存数量等于查询时的数量时，才执行更新# 这里简化演示，生产环境建议使用 SQLAlchemy 的 with_for_update() 或数据库层面锁result = Inventory.query.filter_by(id=inv.id).update({'quantity': inv.quantity - deduct_count}, synchronize_session='fetch')if result == 0:# 更新失败，说明并发冲突，需要重试或抛出异常raise Exception("并发冲突，请重试")remaining_to_deduct -= deduct_count# 3. 记录流水tx = Transaction(medicine_id=medicine_id,batch_no=inv.batch_no,type='OUT',quantity=deduct_count,reason=reason,operator=operator,created_at=datetime.now())transactions_to_create.append(tx)if remaining_to_deduct > 0:raise ValueError("库存不足，剩余需扣减数量: " + str(remaining_to_deduct))# 批量插入流水if transactions_to_create:db.session.add_all(transactions_to_create)db.session.commit()return True

逐行解读：

FEFO 策略：order_by(Inventory.expire_date.asc()) 确保先出效期近的药品，这是医药行业的硬性规定。
循环扣减：因为一个批次可能不够扣，所以必须遍历多个批次。
原子更新：db.session.update 配合 synchronize_session 保证会话一致性。在极高并发下，建议加上 with_for_update() 强制行锁，或者使用数据库的 UPDATE ... WHERE quantity >= deduct_count 语句，通过 affected rows 判断是否成功。
事务一致性：库存扣减和流水记录必须在同一个事务中，db.session.commit() 只执行一次，保证要么都成功，要么都回滚。

运行与测试

代码写完，不能只看日志，必须测试。

1. 单元测试示例

# tests/test_inventory.py
import pytest
from app import create_app, db
from app.models import Medicine, Inventory
from app.services.inventory_service import InventoryService@pytest.fixture
def client():app = create_app()with app.app_context():db.create_all()# 初始化测试数据med = Medicine(name="阿司匹林", spec="100mg", unit="片", safety_stock=10)db.session.add(med)db.session.commit()inv = Inventory(medicine_id=med.id, batch_no="B20231001", quantity=100, expire_date=datetime(2024, 12, 31).date())db.session.add(inv)db.session.commit()yield app.test_client()db.drop_all()def test_deduct_stock_success(client):# 执行扣减result = InventoryService.deduct_stock(medicine_id=1, quantity=50, reason="科室领用", operator="张三")assert result == True# 验证库存inv = Inventory.query.first()assert inv.quantity == 50# 验证流水tx = Transaction.query.first()assert tx.quantity == 50def test_deduct_stock_insufficient(client):with pytest.raises(ValueError) as excinfo:InventoryService.deduct_stock(medicine_id=1, quantity=200, reason="测试", operator="李四")assert "库存不足" in str(excinfo.value)

2. 常见问题排查

数据库连接池耗尽：在 config.py 中配置 SQLALCHEMY_POOL_SIZE，默认值通常较小，高并发下需调大。
日期时区问题：datetime.now() 返回本地时间，数据库存储建议统一使用 UTC 时间，展示时再转换。
批量操作性能：如果一次出库涉及几百个批次，add_all 比循环 add 快得多，但要注意单次事务过大可能导致锁表时间过长。

优化扩展方向

当基础功能跑通后，如何让它更“生产级”？

引入消息队列（MQ）：库存扣减是同步操作，如果后续需要触发短信通知、生成报表等耗时操作，建议将主流程完成后发送消息到 RabbitMQ/Kafka，异步处理。
Redis 缓存热点药品：对于高频查询的药品信息，可以使用 Redis 缓存 medicine_id 对应的名称、规格等静态信息，减少数据库压力。注意缓存一致性，更新药品信息时删除缓存。
接口鉴权：使用 JWT（JSON Web Token）。在 utils/decorators.py 中实现装饰器，校验请求头中的 Token，确保只有授权药师可以执行出库操作。
日志监控：集成 ELK（Elasticsearch, Logstash, Kibana）或 Sentry。记录每次出入库的详细上下文，出现异常时能快速定位是哪个批次、哪个操作人、哪个时间点。