| name | wms-business-flow |
| description | Guides understanding and implementation of WMS business processes including inbound, outbound, inventory management, and system operations. Invoke when developing business logic, understanding workflows, or implementing new features. |
WMS 业务流程开发指南
本技能提供完整的 WMS（仓库管理系统）业务流程指导，涵盖入库、出库、库存管理等核心业务流程。
📋 目录
系统架构概览
SaaS 多租户架构
ModernWMS (SaaS平台)
├── 租户管理 (Tenant)
│ ├── 租户基本信息
│ ├── 租户数据库配置
│ └── 租户AI配置
├── 用户权限 (User & Role)
│ ├── 用户管理
│ ├── 角色管理
│ └── 菜单权限
└── 业务数据 (租户隔离)
├── 基础数据
├── 入库管理
├── 出库管理
└── 库存管理
核心业务模块
WMS业务模块
├── 基础数据管理
│ ├── 仓库管理 (Warehouse)
│ ├── 库区管理 (WarehouseArea)
│ ├── 库位管理 (WarehouseLocation)
│ ├── 商品管理 (SPU/SKU)
│ ├── 分类管理 (Category)
│ ├── 供应商管理 (Supplier)
│ ├── 客户管理 (Customer)
│ └── 货主管理 (GoodsOwner)
├── 入库管理
│ ├── 入库订单 (InboundOrder)
│ ├── 入库拣货上架单 (InboundPickPutaway)
│ └── 入库单 (InboundReceipt)
├── 出库管理
│ ├── 出库订单 (OutboundOrder)
│ ├── 出库拣货单 (OutboundPickPutaway)
│ └── 出库单 (OutboundReceipt)
├── 库存管理
│ ├── 库存查询 (Stock)
│ ├── 库存盘点 (Stocktaking)
│ ├── 库存移动 (Stockmove)
│ ├── 库存调整 (Stockadjust)
│ └── 库存冻结 (Stockfreeze)
└── 系统管理
├── 操作日志 (ActionLog)
├── 打印方案 (PrintSolution)
└── 运费管理 (Freightfee)
核心业务流程
1. 入库流程
🔄 流程图
入库订单 (InboundOrder)
↓ [生成拣货上架单]
拣货上架单 (InboundPickPutaway)
↓ [上架完成]
入库单 (InboundReceipt)
↓ [确认入库]
库存增加 (Stock)
📊 状态流转
入库订单 (InboundOrder)
0-待处理
↓ 生成拣货上架单
1-已生成上架单
↓ 取消
2-已取消
拣货上架单 (InboundPickPutaway)
0-待上架
↓ 开始上架
1-上架中
↓ 上架完成
2-上架完成
↓ 生成入库单
3-已生成入库单
↓ 取消
4-已取消
入库单 (InboundReceipt)
0-待入库
↓ 确认入库
1-已入库
↓ 取消
2-已取消
📝 详细步骤
阶段一：创建入库订单
from app.models.entities.inbound_order import InboundOrder
inbound_order = InboundOrder(
order_no='IB202501010001',
order_status=0,
supplier_id=1,
supplier_name='供应商A',
warehouse_id=1,
warehouse_name='北京仓',
goods_owner_id=1,
goods_owner_name='货主A',
total_qty=100,
total_weight=500.50,
total_volume=200.30,
estimated_arrival_time=1735689600,
creator='admin',
tenant_id='tenant-uuid'
)
关键信息：
- 订单号：唯一标识入库订单
- 供应商：货物的来源方
- 仓库：入库的目标仓库
- 货主：货物的所有者
- 明细：InboundOrderItem（包含SKU、数量、批次等）
阶段二：生成拣货上架单
from app.models.entities.inbound_pick_putaway import InboundPickPutaway
pick_putaway = InboundPickPutaway(
pick_putaway_no='PP202501010001',
pick_putaway_status=0,
order_id=1,
order_no='IB202501010001',
supplier_id=1,
supplier_name='供应商A',
warehouse_id=1,
goods_owner_id=1,
goods_owner_name='货主A',
total_qty=100,
putaway_qty=0,
creator='admin',
tenant_id='tenant-uuid'
)
关键信息：
- 关联入库订单
- 预分配库位：为每个SKU分配目标库位
- 上架人员：负责上架的仓库人员
- 明细：InboundPickPutawayItem（包含SKU、目标库位、数量）
阶段三：上架操作
async def process_putaway(pick_putaway_id: int):
"""处理上架操作"""
pick_putaway = await inbound_pick_putaway_repository.get_by_id(pick_putaway_id)
pick_putaway.pick_putaway_status = 1
pick_putaway.putaway_start_time = int(datetime.now().timestamp())
for item in pick_putaway.items:
pick_putaway.putaway_qty += item.qty
pick_putaway.pick_putaway_status = 2
pick_putaway.putaway_end_time = int(datetime.now().timestamp())
阶段四：生成入库单
from app.models.entities.inbound_receipt import InboundReceipt
receipt = InboundReceipt(
receipt_no='IR202501010001',
receipt_status=0,
pick_putaway_id=1,
order_id=1,
order_no='IB202501010001',
supplier_id=1,
supplier_name='供应商A',
warehouse_id=1,
goods_owner_id=1,
goods_owner_name='货主A',
total_qty=100,
actual_qty=100,
total_weight=500.50,
actual_weight=500.50,
total_volume=200.30,
actual_volume=200.30,
arrival_time=1735689600,
unload_time=1735689900,
unload_person='张三',
creator='admin',
tenant_id='tenant-uuid'
)
阶段五：确认入库（增加库存）
async def confirm_inbound(receipt_id: int):
"""确认入库并增加库存"""
receipt = await inbound_receipt_repository.get_by_id(receipt_id)
receipt.receipt_status = 1
receipt.inbound_time = int(datetime.now().timestamp())
receipt.inbound_person = 'admin'
for item in receipt.items:
stock = Stock(
sku_id=item.sku_id,
goods_location_id=item.goods_location_id,
qty=item.qty,
goods_owner_id=receipt.goods_owner_id,
is_freeze=False,
series_number=item.series_number,
expiry_date=item.expiry_date,
price=item.price,
putaway_date=int(datetime.now().timestamp()),
warehouse_id=receipt.warehouse_id,
warehouse_name=receipt.warehouse_name,
warehouse_area_id=item.warehouse_area_id,
warehouse_area_name=item.warehouse_area_name,
warehouse_location_name=item.warehouse_location_name,
spu_name=item.spu_name,
sku_code=item.sku_code,
sku_name=item.sku_name,
batch_no=item.batch_no,
production_date=item.production_date,
tenant_id=receipt.tenant_id
)
await stock_repository.create(stock)
2. 出库流程
🔄 流程图
出库订单 (OutboundOrder)
↓ [生成拣货单]
拣货单 (OutboundPickPutaway)
↓ [拣货完成]
出库单 (OutboundReceipt)
↓ [确认出库]
库存扣减 (Stock)
📊 状态流转
出库订单 (OutboundOrder)
0-待处理
↓ 生成拣货单
1-已生成拣货单
↓ 取消
2-已取消
拣货单 (OutboundPickPutaway)
0-待拣货
↓ 开始拣货
1-拣货中
↓ 拣货完成
2-拣货完成
↓ 生成出库单
3-已生成出库单
↓ 取消
4-已取消
出库单 (OutboundReceipt)
0-待出库
↓ 确认出库
1-已出库
↓ 取消
2-已取消
📝 详细步骤
阶段一：创建出库订单
from app.models.entities.outbound_order import OutboundOrder
outbound_order = OutboundOrder(
order_no='OB202501010001',
order_status=0,
customer_id=1,
customer_name='客户A',
warehouse_id=1,
warehouse_name='北京仓',
goods_owner_id=1,
goods_owner_name='货主A',
total_qty=50,
total_weight=250.25,
total_volume=100.15,
creator='admin',
tenant_id='tenant-uuid'
)
关键信息：
- 订单号：唯一标识出库订单
- 客户：货物的接收方
- 仓库：出库的来源仓库
- 货主：货物的所有者
- 明细：OutboundOrderItem（包含SKU、数量等）
阶段二：生成拣货单
from app.models.entities.outbound_pick_putaway import OutboundPickPutaway
pick_putaway = OutboundPickPutaway(
pick_putaway_no='OP202501010001',
pick_putaway_status=0,
order_id=1,
order_no='OB202501010001',
customer_id=1,
customer_name='客户A',
warehouse_id=1,
goods_owner_id=1,
goods_owner_name='货主A',
total_qty=50,
picked_qty=0,
creator='admin',
tenant_id='tenant-uuid'
)
关键信息：
- 关联出库订单
- 锁定库存：为拣货的SKU锁定库存
- 拣货人员：负责拣货的仓库人员
- 明细：OutboundPickPutawayItem（包含SKU、源库位、数量）
阶段三：拣货操作
async def process_pick(pick_putaway_id: int):
"""处理拣货操作"""
pick_putaway = await outbound_pick_putaway_repository.get_by_id(pick_putaway_id)
pick_putaway.pick_putaway_status = 1
pick_putaway.pick_start_time = int(datetime.now().timestamp())
for item in pick_putaway.items:
pick_putaway.picked_qty += item.qty
pick_putaway.pick_putaway_status = 2
pick_putaway.pick_end_time = int(datetime.now().timestamp())
阶段四：生成出库单
from app.models.entities.outbound_receipt import OutboundReceipt
receipt = OutboundReceipt(
receipt_no='OR202501010001',
receipt_status=0,
pick_putaway_id=1,
order_id=1,
order_no='OB202501010001',
customer_id=1,
customer_name='客户A',
warehouse_id=1,
goods_owner_id=1,
goods_owner_name='货主A',
total_qty=50,
actual_qty=50,
total_weight=250.25,
actual_weight=250.25,
total_volume=100.15,
actual_volume=100.15,
package_no='PKG202501010001',
package_person='李四',
package_time=1735690000,
weighing_no='WG202501010001',
weighing_person='王五',
weighing_weight=250.25,
waybill_no='SF1234567890',
carrier='顺丰速运',
freightfee=15.50,
creator='admin',
tenant_id='tenant-uuid'
)
阶段五：确认出库（扣减库存）
async def confirm_outbound(receipt_id: int):
receipt = await outbound_receipt_repository.get_by_id(receipt_id)
receipt.receipt_status = 1
receipt.outbound_time = int(datetime.now().timestamp())
receipt.outbound_person = 'admin'
for item in receipt.items:
stocks = await stock_repository.get_by_location_and_sku(
item.goods_location_id,
item.sku_id
)
for stock in stocks:
if stock.qty >= item.qty:
stock.qty -= item.qty
await stock_repository.update(stock.id, {'qty': stock.qty})
break
else:
item.qty -= stock.qty
stock.qty = 0
await stock_repository.update(stock.id, {'qty': stock.qty})
3. 库存管理流程
3.1 库存查询
from app.models.entities.stock import Stock
async def query_stock(
sku_id: int,
goods_owner_id: int,
warehouse_id: int
) -> List[Stock]:
"""查询指定SKU的库存"""
stocks = await stock_repository.get_by_conditions({
'sku_id': sku_id,
'goods_owner_id': goods_owner_id,
'warehouse_id': warehouse_id,
'tenant_id': tenant_id
})
return stocks
async def summarize_stock(sku_id: int) -> dict:
"""汇总库存信息"""
stocks = await stock_repository.get_by_sku_id(sku_id)
total_qty = sum(stock.qty for stock in stocks)
frozen_qty = sum(stock.qty for stock in stocks if stock.is_freeze)
available_qty = total_qty - frozen_qty
return {
'total_qty': total_qty,
'frozen_qty': frozen_qty,
'available_qty': available_qty
}
3.2 库存盘点
from app.models.entities.stocktaking import Stocktaking
stocktaking = Stocktaking(
job_code='ST202501010001',
job_status=False,
sku_id=1,
goods_owner_id=1,
goods_location_id=1,
series_number='SN123456',
expiry_date=1735689600,
price=100.00,
putaway_date=1735689600,
book_qty=100,
counted_qty=95,
difference_qty=-5,
creator='admin',
tenant_id='tenant-uuid'
)
async def process_stocktaking(stocktaking_id: int):
stocktaking = await stocktaking_repository.get_by_id(stocktaking_id)
stocktaking.difference_qty = stocktaking.counted_qty - stocktaking.book_qty
if stocktaking.difference_qty != 0:
stockadjust = Stockadjust(
job_code=f'SA{datetime.now().strftime("%Y%m%d%H%M%S")}',
sku_id=stocktaking.sku_id,
goods_owner_id=stocktaking.goods_owner_id,
goods_location_id=stocktaking.goods_location_id,
qty=stocktaking.difference_qty,
job_type=1,
source_table_id=stocktaking.id,
series_number=stocktaking.series_number,
expiry_date=stocktaking.expiry_date,
price=stocktaking.price,
putaway_date=stocktaking.putaway_date,
creator=stocktaking.creator,
tenant_id=stocktaking.tenant_id
)
await stockadjust_repository.create(stockadjust)
stocktaking.job_status = True
stocktaking.handler = 'admin'
stocktaking.handle_time = int(datetime.now().timestamp())
3.3 库存移动
from app.models.entities.stockmove import Stockmove
stockmove = Stockmove(
job_code='SM202501010001',
move_status=0,
sku_id=1,
orig_goods_location_id=1,
dest_googs_location_id=2,
qty=50,
goods_owner_id=1,
handler='张三',
handle_time=0,
creator='admin',
tenant_id='tenant-uuid',
series_number='SN123456',
expiry_date=1735689600,
price=100.00,
putaway_date=1735689600
)
async def process_stockmove(stockmove_id: int):
stockmove = await stockmove_repository.get_by_id(stockmove_id)
orig_stock = await stock_repository.get_by_location_and_sku(
stockmove.orig_goods_location_id,
stockmove.sku_id
)
if not orig_stock or orig_stock.qty < stockmove.qty:
raise Exception('原货位库存不足')
dest_stock = await stock_repository.get_by_location_and_sku(
stockmove.dest_googs_location_id,
stockmove.sku_id
)
orig_stock.qty -= stockmove.qty
await stock_repository.update(orig_stock.id, {'qty': orig_stock.qty})
if dest_stock:
dest_stock.qty += stockmove.qty
await stock_repository.update(dest_stock.id, {'qty': dest_stock.qty})
else:
dest_stock = Stock(
sku_id=stockmove.sku_id,
goods_location_id=stockmove.dest_googs_location_id,
qty=stockmove.qty,
goods_owner_id=stockmove.goods_owner_id,
is_freeze=False,
series_number=stockmove.series_number,
expiry_date=stockmove.expiry_date,
price=stockmove.price,
putaway_date=stockmove.putaway_date,
tenant_id=stockmove.tenant_id
)
await stock_repository.create(dest_stock)
stockmove.move_status = 1
stockmove.handle_time = int(datetime.now().timestamp())
3.4 库存调整
from app.models.entities.stockadjust import Stockadjust
stockadjust = Stockadjust(
job_code='SA202501010001',
sku_id=1,
goods_owner_id=1,
goods_location_id=1,
qty=10,
job_type=0,
source_table_id=0,
series_number='SN123456',
expiry_date=1735689600,
price=100.00,
putaway_date=1735689600,
creator='admin',
tenant_id='tenant-uuid',
is_update_stock=False
)
async def process_stockadjust(stockadjust_id: int):
stockadjust = await stockadjust_repository.get_by_id(stockadjust_id)
stock = await stock_repository.get_by_location_and_sku(
stockadjust.goods_location_id,
stockadjust.sku_id
)
if not stock:
stock = Stock(
sku_id=stockadjust.sku_id,
goods_location_id=stockadjust.goods_location_id,
qty=stockadjust.qty,
goods_owner_id=stockadjust.goods_owner_id,
is_freeze=False,
series_number=stockadjust.series_number,
expiry_date=stockadjust.expiry_date,
price=stockadjust.price,
putaway_date=stockadjust.putaway_date,
tenant_id=stockadjust.tenant_id
)
await stock_repository.create(stock)
else:
stock.qty += stockadjust.qty
if stock.qty < 0:
raise Exception('库存不能为负数')
await stock_repository.update(stock.id, {'qty': stock.qty})
stockadjust.is_update_stock = True
3.5 库存冻结
from app.models.entities.stockfreeze import Stockfreeze
stockfreeze = Stockfreeze(
job_code='SF202501010001',
job_type=True,
sku_id=1,
goods_owner_id=1,
goods_location_id=1,
handler='admin',
handle_time=0,
tenant_id='tenant-uuid',
series_number='SN123456'
)
async def process_stockfreeze(stockfreeze_id: int):
stockfreeze = await stockfreeze_repository.get_by_id(stockfreeze_id)
stocks = await stock_repository.get_by_conditions({
'sku_id': stockfreeze.sku_id,
'goods_owner_id': stockfreeze.goods_owner_id,
'goods_location_id': stockfreeze.goods_location_id,
'tenant_id': stockfreeze.tenant_id
})
for stock in stocks:
if stockfreeze.job_type:
stock.is_freeze = True
else:
stock.is_freeze = False
await stock_repository.update(stock.id, {'is_freeze': stock.is_freeze})
stockfreeze.handle_time = int(datetime.now().timestamp())
基础数据管理
4.1 仓库结构管理
仓库 (Warehouse)
↓ 包含多个
库区 (WarehouseArea)
↓ 包含多个
库位 (WarehouseLocation)
仓库层级关系：
- 仓库：最高层级，代表一个物理仓库
- 库区：仓库内的区域划分（如收货区、存储区、发货区）
- 库位：最小的存储单元，具体的货架位置
库位类型：
4.2 商品管理
商品分类 (Category)
↓ 包含多个
SPU (Standard Product Unit - 标准产品单位)
↓ 包含多个
SKU (Stock Keeping Unit - 库存量单位)
SPU vs SKU：
- SPU：标准产品单位，代表一种商品（如：iPhone 15）
- SKU：库存量单位，代表具体的规格（如：iPhone 15 128GB 黑色）
4.3 业务伙伴管理
- 供应商：提供货物的企业或个人
- 客户：接收货物的企业或个人
- 货主：货物的所有者（在多货主场景下）
多租户隔离
租户隔离原则
class BaseModel:
tenant_id = Column(String(36), nullable=False, comment='租户ID')
async def get_by_tenant(tenant_id: str, conditions: dict):
conditions['tenant_id'] = tenant_id
return await repository.get_by_conditions(conditions)
租户数据库隔离
class TenantDatabaseConfig:
tenant_id: str
db_host: str
db_port: int
db_database: str
db_username: str
db_password: str
AI集成
AI配置管理
from app.models.entities.tenant_ai_config import TenantAIConfig
ai_config = TenantAIConfig(
provider='openai',
model='gpt-4',
temperature=0.7,
top_p=0.9,
max_tokens=2000,
config='{}',
tenant_id='tenant-uuid',
creator='admin',
is_valid=True
)
AI应用场景
- 智能推荐：推荐合适的库位
- 路径优化：优化拣货路径
- 需求预测：预测库存需求
- 异常检测：检测库存异常
业务规则
1. 入库规则
- 入库订单必须指定供应商、仓库、货主
- 拣货上架单必须关联入库订单
- 上架完成后才能生成入库单
- 确认入库后才能增加库存
- 库存按SKU+库位+货主+批次号唯一
2. 出库规则
- 出库订单必须指定客户、仓库、货主
- 拣货单必须关联出库订单
- 拣货时锁定库存
- 拣货完成后才能生成出库单
- 确认出库后才能扣减库存
- 先进先出（FIFO）原则
3. 库存规则
- 库存不能为负数
- 冻结的库存不能出库
- 库存调整需要审批
- 库存盘点需要处理差异
- 库存移动需要记录轨迹
4. 数据一致性规则
- 所有操作必须记录操作日志
- 订单状态流转必须符合业务流程
- 关联数据删除需要级联处理
- 租户数据必须严格隔离
开发规范
1. 服务层开发
from app.services.base_service import TenantAwareService
from app.core.current_user import CurrentUser
class BusinessService(TenantAwareService):
"""业务服务基类"""
async def create_order(self, data: dict, current_user: CurrentUser):
"""创建订单"""
self._validate_data(data)
entity = self._create_entity(data, current_user)
await self._repository.create(entity)
await self._log_action('create', entity.id, current_user)
await self._clear_cache(current_user.tenant_id)
return entity
2. 状态流转管理
class OrderStatusManager:
"""订单状态管理器"""
STATUS_FLOW = {
'inbound_order': {
0: [1, 2],
1: [2],
2: []
},
'outbound_order': {
0: [1, 2],
1: [2],
2: []
}
}
@classmethod
def can_transition(cls, order_type: str, from_status: int, to_status: int) -> bool:
"""检查状态是否可以流转"""
allowed = cls.STATUS_FLOW.get(order_type, {}).get(from_status, [])
return to_status in allowed
3. 库存操作
class StockOperator:
"""库存操作器"""
@staticmethod
async def increase_stock(stock_data: dict):
"""增加库存"""
stock = await stock_repository.find_stock(
stock_data['sku_id'],
stock_data['goods_location_id'],
stock_data['goods_owner_id']
)
if stock:
stock.qty += stock_data['qty']
await stock_repository.update(stock.id, {'qty': stock.qty})
else:
await stock_repository.create(Stock(**stock_data))
@staticmethod
async def decrease_stock(stock_data: dict):
"""减少库存"""
stock = await stock_repository.find_stock(
stock_data['sku_id'],
stock_data['goods_location_id'],
stock_data['goods_owner_id']
)
if not stock or stock.qty < stock_data['qty']:
raise Exception('库存不足')
stock.qty -= stock_data['qty']
await stock_repository.update(stock.id, {'qty': stock.qty})
性能优化
1. 内存优化
1.1 避免不必要的对象创建
async def process_orders(order_ids: List[int]):
for order_id in order_ids:
repository = InboundOrderRepository()
order = await repository.get_by_id(order_id)
async def process_orders(order_ids: List[int]):
repository = InboundOrderRepository()
for order_id in order_ids:
order = await repository.get_by_id(order_id)
1.2 及时释放不再需要的资源
async def query_large_data():
results = await db.execute("SELECT * FROM large_table")
return results
async def query_large_data():
async with db.connect() as conn:
results = await conn.execute("SELECT * FROM large_table")
return results
1.3 注意内存泄漏问题
cache = {}
async def get_data(key: str):
if key not in cache:
cache[key] = await fetch_data(key)
return cache[key]
from functools import lru_cache
@lru_cache(maxsize=1000)
async def get_data(key: str):
return await fetch_data(key)
2. 计算优化
2.1 避免重复计算
async def calculate_inventory(order_items: List[OrderItem]):
total_qty = 0
total_weight = 0
total_volume = 0
for item in order_items:
total_qty += item.qty
total_weight += item.qty * item.weight
total_volume += item.qty * item.volume
for item in order_items:
total_qty += item.qty
total_weight += item.qty * item.weight
total_volume += item.qty * item.volume
return total_qty, total_weight, total_volume
async def calculate_inventory(order_items: List[OrderItem]):
total_qty = sum(item.qty for item in order_items)
total_weight = sum(item.qty * item.weight for item in order_items)
total_volume = sum(item.qty * item.volume for item in order_items)
return total_qty, total_weight, total_volume
2.2 使用适当的数据结构和算法
async def find_stock_by_sku(sku_id: int, stocks: List[Stock]):
for stock in stocks:
if stock.sku_id == sku_id:
return stock
return None
async def find_stock_by_sku(sku_id: int, stocks: List[Stock]):
stock_dict = {stock.sku_id: stock for stock in stocks}
return stock_dict.get(sku_id)
2.3 延迟计算直到必要时
async def get_order_details(order_id: int):
order = await order_repository.get_by_id(order_id)
items = await item_repository.get_by_order_id(order_id)
stocks = await stock_repository.get_by_items(items)
inventory_info = calculate_inventory(stocks)
return order, items, inventory_info
async def get_order_details(order_id: int):
order = await order_repository.get_by_id(order_id)
items = await item_repository.get_by_order_id(order_id)
async def get_inventory_info():
stocks = await stock_repository.get_by_items(items)
return calculate_inventory(stocks)
return order, items, get_inventory_info
3. 并行优化
3.1 识别可并行化的任务
async def process_multiple_orders(order_ids: List[int]):
results = []
for order_id in order_ids:
result = await process_order(order_id)
results.append(result)
return results
import asyncio
async def process_multiple_orders(order_ids: List[int]):
tasks = [process_order(order_id) for order_id in order_ids]
results = await asyncio.gather(*tasks)
return results
3.2 避免不必要的同步
async def update_stock_and_log(stock_id: int, qty: int):
await stock_repository.update(stock_id, {'qty': qty})
await log_repository.create({'action': 'update', 'qty': qty})
return True
async def update_stock_and_log(stock_id: int, qty: int):
stock_task = asyncio.create_task(stock_repository.update(stock_id, {'qty': qty}))
log_task = asyncio.create_task(log_repository.create({'action': 'update', 'qty': qty}))
await asyncio.gather(stock_task, log_task)
return True
3.3 注意线程安全问题
class StockManager:
def __init__(self):
self.stock_cache = {}
async def get_stock(self, sku_id: int):
if sku_id not in self.stock_cache:
self.stock_cache[sku_id] = await fetch_stock(sku_id)
return self.stock_cache[sku_id]
import asyncio
class StockManager:
def __init__(self):
self.stock_cache = {}
self.lock = asyncio.Lock()
async def get_stock(self, sku_id: int):
if sku_id not in self.stock_cache:
async with self.lock:
if sku_id not in self.stock_cache:
self.stock_cache[sku_id] = await fetch_stock(sku_id)
return self.stock_cache[sku_id]
4. 数据库查询优化
4.1 批量查询代替循环查询
async def get_orders_with_items(order_ids: List[int]):
orders = []
for order_id in order_ids:
order = await order_repository.get_by_id(order_id)
items = await item_repository.get_by_order_id(order_id)
order.items = items
orders.append(order)
return orders
async def get_orders_with_items(order_ids: List[int]):
orders = await order_repository.get_by_ids(order_ids)
all_items = await item_repository.get_by_order_ids(order_ids)
items_by_order = {}
for item in all_items:
if item.order_id not in items_by_order:
items_by_order[item.order_id] = []
items_by_order[item.order_id].append(item)
for order in orders:
order.items = items_by_order.get(order.id, [])
return orders
4.2 使用索引优化查询
async def get_stock_by_location(location_name: str):
stocks = await stock_repository.get_by_conditions({
'warehouse_location_name': location_name,
'tenant_id': tenant_id
})
return stocks
async def get_stock_by_location(location_id: int):
stocks = await stock_repository.get_by_conditions({
'goods_location_id': location_id,
'tenant_id': tenant_id
})
return stocks
4.3 分页查询大数据集
async def export_all_stocks():
stocks = await stock_repository.get_all()
return stocks
async def export_all_stocks():
page_size = 1000
page = 0
while True:
stocks = await stock_repository.get_by_page(
page=page,
page_size=page_size
)
if not stocks:
break
yield stocks
page += 1
5. 代码抽象和聚合
5.1 提取公共逻辑到基类
class InboundOrderService:
async def create(self, data: dict, current_user: CurrentUser):
entity = InboundOrder(**data)
entity.tenant_id = current_user.tenant_id
entity.creator = current_user.username
entity.create_time = int(datetime.now().timestamp())
await self.repository.create(entity)
class OutboundOrderService:
async def create(self, data: dict, current_user: CurrentUser):
entity = OutboundOrder(**data)
entity.tenant_id = current_user.tenant_id
entity.creator = current_user.username
entity.create_time = int(datetime.now().timestamp())
await self.repository.create(entity)
class BaseOrderService:
"""订单服务基类"""
def _create_entity(self, entity_class, data: dict, current_user: CurrentUser):
"""创建实体并设置公共字段"""
entity = entity_class(**data)
entity.tenant_id = current_user.tenant_id
entity.creator = current_user.username
entity.create_time = int(datetime.now().timestamp())
return entity
class InboundOrderService(BaseOrderService):
async def create(self, data: dict, current_user: CurrentUser):
entity = self._create_entity(InboundOrder, data, current_user)
await self.repository.create(entity)
class OutboundOrderService(BaseOrderService):
async def create(self, data: dict, current_user: CurrentUser):
entity = self._create_entity(OutboundOrder, data, current_user)
await self.repository.create(entity)
5.2 使用装饰器处理横切关注点
async def get_order(order_id: int):
cache_key = f"order:{order_id}"
cached = await cache.get(cache_key)
if cached:
return cached
logger.info(f"查询订单: {order_id}")
order = await order_repository.get_by_id(order_id)
await cache.set(cache_key, order, expire=3600)
return order
def cache_with_log(expire: int = 3600):
"""缓存和日志装饰器"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
order_id = args[1] if len(args) > 1 else kwargs.get('order_id')
cache_key = f"order:{order_id}"
cached = await cache.get(cache_key)
if cached:
return cached
logger.info(f"查询订单: {order_id}")
result = await func(*args, **kwargs)
await cache.set(cache_key, result, expire=expire)
return result
return wrapper
return decorator
@cache_with_log(expire=3600)
async def get_order(order_id: int):
return await order_repository.get_by_id(order_id)
5.3 使用策略模式处理不同业务逻辑
async def process_stock_operation(operation_type: str, data: dict):
if operation_type == 'increase':
pass
elif operation_type == 'decrease':
pass
elif operation_type == 'move':
pass
elif operation_type == 'freeze':
pass
else:
raise Exception('不支持的操作类型')
class StockOperationStrategy:
"""库存操作策略基类"""
async def execute(self, data: dict):
raise NotImplementedError
class IncreaseStockStrategy(StockOperationStrategy):
"""增加库存策略"""
async def execute(self, data: dict):
pass
class DecreaseStockStrategy(StockOperationStrategy):
"""减少库存策略"""
async def execute(self, data: dict):
pass
class StockOperationProcessor:
"""库存操作处理器"""
def __init__(self):
self.strategies = {
'increase': IncreaseStockStrategy(),
'decrease': DecreaseStockStrategy(),
}
async def process(self, operation_type: str, data: dict):
strategy = self.strategies.get(operation_type)
if not strategy:
raise Exception('不支持的操作类型')
return await strategy.execute(data)
快速参考
状态码速查表
入库订单状态
| 状态码 | 状态名称 | 说明 |
|---|
| 0 | 待处理 | 订单已创建，等待处理 |
| 1 | 已生成上架单 | 已生成拣货上架单 |
| 2 | 已取消 | 订单已取消 |
拣货上架单状态
| 状态码 | 状态名称 | 说明 |
|---|
| 0 | 待上架 | 等待上架操作 |
| 1 | 上架中 | 正在进行上架 |
| 2 | 上架完成 | 上架操作完成 |
| 3 | 已生成入库单 | 已生成入库单 |
| 4 | 已取消 | 已取消 |
入库单状态
| 状态码 | 状态名称 | 说明 |
|---|
| 0 | 待入库 | 等待确认入库 |
| 1 | 已入库 | 已确认入库，库存已增加 |
| 2 | 已取消 | 已取消 |
出库订单状态
| 状态码 | 状态名称 | 说明 |
|---|
| 0 | 待处理 | 订单已创建，等待处理 |
| 1 | 已生成拣货单 | 已生成拣货单 |
| 2 | 已取消 | 订单已取消 |
拣货单状态
| 状态码 | 状态名称 | 说明 |
|---|
| 0 | 待拣货 | 等待拣货操作 |
| 1 | 拣货中 | 正在进行拣货 |
| 2 | 拣货完成 | 拣货操作完成 |
| 3 | 已生成出库单 | 已生成出库单 |
| 4 | 已取消 | 已取消 |
出库单状态
| 状态码 | 状态名称 | 说明 |
|---|
| 0 | 待出库 | 等待确认出库 |
| 1 | 已出库 | 已确认出库，库存已扣减 |
| 2 | 已取消 | 已取消 |
常用查询速查
库存查询
stocks = await stock_repository.get_by_conditions({
'sku_id': sku_id,
'tenant_id': tenant_id
})
stocks = await stock_repository.get_by_conditions({
'goods_location_id': location_id,
'tenant_id': tenant_id
})
stocks = await stock_repository.get_by_conditions({
'goods_owner_id': owner_id,
'tenant_id': tenant_id
})
stocks = await stock_repository.get_by_conditions({
'is_freeze': False,
'tenant_id': tenant_id
})
订单查询
orders = await inbound_order_repository.get_by_conditions({
'order_status': 0,
'tenant_id': tenant_id
})
orders = await inbound_order_repository.get_by_conditions({
'supplier_id': supplier_id,
'tenant_id': tenant_id
})
pick_putaways = await outbound_pick_putaway_repository.get_by_conditions({
'pick_putaway_status': 0,
'tenant_id': tenant_id
})
常用操作速查
库存操作
stock.qty += qty
await stock_repository.update(stock.id, {'qty': stock.qty})
if stock.qty < qty:
raise Exception('库存不足')
stock.qty -= qty
await stock_repository.update(stock.id, {'qty': stock.qty})
stock.is_freeze = True
await stock_repository.update(stock.id, {'is_freeze': True})
stock.is_freeze = False
await stock_repository.update(stock.id, {'is_freeze': False})
订单状态流转
order.order_status = new_status
await order_repository.update(order.id, {'order_status': new_status})
if not OrderStatusManager.can_transition('inbound_order', 0, 1):
raise Exception('状态流转不允许')
实体关系速查
InboundOrder (入库订单)
├── InboundOrderItem (入库订单明细)
├── InboundPickPutaway (拣货上架单)
│ ├── InboundPickPutawayItem (拣货上架单明细)
│ └── InboundReceipt (入库单)
│ └── InboundReceiptItem (入库单明细)
OutboundOrder (出库订单)
├── OutboundOrderItem (出库订单明细)
├── OutboundPickPutaway (拣货单)
│ ├── OutboundPickPutawayItem (拣货单明细)
│ └── OutboundReceipt (出库单)
│ └── OutboundReceiptItem (出库单明细)
Stock (库存)
├── 关联到 WarehouseLocation (库位)
├── 关联到 Sku (SKU)
└── 关联到 GoodsOwner (货主)
开发流程 Checklist
入库流程开发
出库流程开发
库存管理开发
常见问题
1. 状态流转错误
order.order_status = 2
if not OrderStatusManager.can_transition('inbound_order', 0, 2):
raise Exception('状态流转不允许')
order.order_status = 2
2. 库存不足
stock.qty -= qty
if stock.qty < qty:
raise Exception('库存不足')
stock.qty -= qty
3. 租户隔离
entities = await repository.get_all()
entities = await repository.get_by_tenant(tenant_id, {})
4. 缓存失效
await stock_repository.update(stock_id, data)
await stock_repository.update(stock_id, data)
await cache.remove(f"stock:{stock_id}")
5. 事务处理
await repository.create(entity1)
await repository.create(entity2)
async with db_session.begin():
await repository.create(entity1)
await repository.create(entity2)
6. 异常处理
stock.qty -= qty
await repository.update(stock.id, {'qty': stock.qty})
try:
if stock.qty < qty:
raise Exception('库存不足')
stock.qty -= qty
await repository.update(stock.id, {'qty': stock.qty})
except Exception as e:
logger.error(f'库存扣减失败: {e}')
raise
