查询需求分析
- 目标：对 products 表按 sku 批量扣减库存（stock = stock - delta），仅更新 active = true 且 warehouse_id 匹配的记录。
- 并发一致性：使用事务 + with_for_update 行级锁，避免并发扣减超卖。
- 业务规则：若库存不足（stock < delta）则跳过该 sku，并返回跳过明细；支持可选 operator_id 写入 last_operator。
- 输入：payload = [{sku, delta}, ...]，warehouse_id，operator_id（可选）。
- 输出：原始结果（Python 基本类型），包含受影响行数、成功清单、跳过清单，并演示一次性查询更新后的库存。
- 兼容性：使用 SQLAlchemy 2.0 ORM 风格，避免弃用 API，参数绑定防注入。
核心代码
from __future__ import annotations
from typing import Any, Dict, Iterable, List, Optional, Tuple
from dataclasses import dataclass
from sqlalchemy import create_engine, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, sessionmaker
from sqlalchemy import String, Integer, Boolean
# ---------- ORM 映射 ----------
class Base(DeclarativeBase):
pass
class Product(Base):
__tablename__ = "products"
id: Mapped[int] = mapped_column(primary_key=True)
sku: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
warehouse_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
stock: Mapped[int] = mapped_column(Integer, nullable=False)
active: Mapped[bool] = mapped_column(Boolean, index=True, nullable=False, default=True)
# 可选的操作人追踪字段
last_operator: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
# ---------- 业务函数 ----------
@dataclass
class DecrementResult:
affected_rows: int
successes: List[Dict[str, Any]]
skipped: List[Dict[str, Any]]
def _normalize_payload(payload: Iterable[Dict[str, Any]]) -> Dict[str, int]:
"""
归并 payload（同一 sku 多次出现则累加 delta），并做基础校验。
仅保留 delta > 0 的项。
"""
sku_to_delta: Dict[str, int] = {}
for i, item in enumerate(payload):
if not isinstance(item, dict):
raise ValueError(f"payload[{i}] 必须是字典")
sku = item.get("sku")
delta = item.get("delta")
if not sku or not isinstance(sku, str):
raise ValueError(f"payload[{i}].sku 必须是非空字符串")
if not isinstance(delta, int):
raise ValueError(f"payload[{i}].delta 必须是整数")
if delta <= 0:
# 非正扣减没有意义，跳过
continue
sku_to_delta[sku] = sku_to_delta.get(sku, 0) + delta
return sku_to_delta
def decrement_stocks(
session: Session,
*,
warehouse_id: int,
payload: Iterable[Dict[str, Any]],
operator_id: Optional[int] = None,
lock_nowait: bool = False,
) -> DecrementResult:
"""
批量扣减库存（行锁 + 事务环境中执行）。
规则：
- 仅 active = True 且 warehouse_id 匹配的记录参与。
- 库存不足（stock < delta）跳过。
- 支持可选 operator_id -> 写入 last_operator。
并在事务内返回原始结果（纯 Python 类型）。
要求：调用方应确保在 session.begin() 或外层事务内调用，以获得行锁效果。
"""
sku_to_delta = _normalize_payload(payload)
if not sku_to_delta:
return DecrementResult(affected_rows=0, successes=[], skipped=[])
skus = list(sku_to_delta.keys())
# 1) 选中目标行并加行级锁
stmt = (
select(Product)
.where(
Product.active.is_(True),
Product.warehouse_id == warehouse_id,
Product.sku.in_(skus),
)
.with_for_update(nowait=lock_nowait) # 行锁（SQLite 会忽略，见注意事项）
)
rows = session.execute(stmt).scalars().all()
found_skus = set()
successes: List[Dict[str, Any]] = []
skipped: List[Dict[str, Any]] = []
# 2) 逐行检查并更新（已加锁，避免并发超卖）
for p in rows:
found_skus.add(p.sku)
delta = sku_to_delta.get(p.sku, 0)
if delta <= 0:
skipped.append(
{"sku": p.sku, "delta": delta, "reason": "invalid_delta", "current_stock": p.stock}
)
continue
if p.stock >= delta:
p.stock = p.stock - delta
if operator_id is not None:
p.last_operator = operator_id
successes.append({"sku": p.sku, "delta": delta, "new_stock": p.stock})
else:
skipped.append(
{"sku": p.sku, "delta": delta, "reason": "insufficient_stock", "current_stock": p.stock}
)
# 3) 未找到或不满足 active/warehouse 条件的 SKU 视为跳过
missing_skus = set(skus) - found_skus
for sku in missing_skus:
skipped.append(
{"sku": sku, "delta": sku_to_delta[sku], "reason": "not_found_or_inactive_or_warehouse_mismatch"}
)
# 4) 刷新到数据库（仍在事务内，可继续查询最新库存）
session.flush()
return DecrementResult(
affected_rows=len(successes),
successes=successes,
skipped=skipped,
)
def get_stocks(
session: Session,
*,
warehouse_id: int,
skus: Iterable[str],
) -> List[Dict[str, Any]]:
"""
一次性查询给定 sku 的当前库存（原始结果）。
"""
skus = list(set(skus))
if not skus:
return []
stmt = select(Product.sku, Product.stock).where(
Product.warehouse_id == warehouse_id,
Product.sku.in_(skus),
)
rows = session.execute(stmt).all()
return [{"sku": sku, "stock": stock} for (sku, stock) in rows]
# ---------- Session 工厂（示例） ----------
# 请替换为你的数据库 URL（勿在代码中硬编码敏感信息；可用环境变量加载）
# engine = create_engine(os.getenv("DATABASE_URL"), pool_pre_ping=True, future=True)
# 这里仅作占位演示，不要在生产中直接使用 sqlite+aiosqlite 或内存数据库
# engine = create_engine("postgresql+psycopg://user:pass@host:5432/dbname", pool_pre_ping=True)
代码解释
- Product 映射：最小化定义所需字段 sku、warehouse_id、stock、active、last_operator，用于库存扣减与可选操作者记录。
- _normalize_payload：合并 payload 中相同 sku 的 delta，过滤 delta <= 0，保证输入有效性并减少重复处理。
- decrement_stocks：
- 使用 select(...).with_for_update() 对目标行加锁，保证并发下库存检查与更新的原子性。
- 对每条锁定的产品行，验证 stock >= delta 再扣减；否则加入 skipped，并记录当前库存。
- 对数据库中未匹配（可能因 inactive 或仓库不符）的 SKU，标记为跳过。
- session.flush() 确保更新已发往数据库，但仍处于同一事务中，后续可继续查询最新库存。
- 返回原始的 Python 数据结构（受影响行数、成功与跳过清单）。
- get_stocks：一次性查询指定 sku 在该仓库的最新库存，返回原始结果列表。
- with_for_update(nowait=...)：提供可选 nowait 行为；若为 True 且行被其它事务锁定，将抛出数据库异常，调用方可捕获以快速失败。
使用示例
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 1) 创建 Engine / Session
# 替换为你的数据库连接串（请从环境变量读取，避免泄漏敏感信息）
engine = create_engine("postgresql+psycopg://user:pass@host:5432/dbname", pool_pre_ping=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
# 首次建表（如果你还没有表结构，生产中请用迁移工具 Alembic 管理）
# Base.metadata.create_all(engine)
payload = [
{"sku": "SKU-001", "delta": 3},
{"sku": "SKU-002", "delta": 5},
{"sku": "SKU-001", "delta": 2}, # 同 SKU 会被合并为 delta=5
{"sku": "SKU-NotExist", "delta": 1},
]
warehouse_id = 10
operator_id = 9527
# 2) 正常执行并提交
with SessionLocal.begin() as session:
# 扣减库存（行锁生效）
result = decrement_stocks(
session,
warehouse_id=warehouse_id,
payload=payload,
operator_id=operator_id,
lock_nowait=False, # 可改 True 以在遇到锁冲突时立即失败
)
print("decrement result:", {
"affected_rows": result.affected_rows,
"successes": result.successes,
"skipped": result.skipped
})
# 一次性查询更新后的库存（仍在同一事务内，读取最新值）
updated_skus = list({item["sku"] for item in payload})
current_stocks = get_stocks(session, warehouse_id=warehouse_id, skus=updated_skus)
print("current stocks:", current_stocks)
# 3) 回滚示例（演示在出现业务错误时如何回滚）
try:
with SessionLocal.begin() as session:
# 故意传入非法 delta，触发 ValueError
bad_payload = [{"sku": "SKU-001", "delta": -100}]
_ = decrement_stocks(
session,
warehouse_id=warehouse_id,
payload=bad_payload,
operator_id=operator_id,
)
# 任何异常抛出都会导致该 with 块自动回滚
except Exception as e:
print("rolled back due to:", repr(e))
# 4) 事务外的读取（验证提交结果）
with SessionLocal() as session:
current_stocks = get_stocks(session, warehouse_id=warehouse_id, skus=["SKU-001", "SKU-002", "SKU-NotExist"])
print("stocks after commit:", current_stocks)
注意事项
- 事务与行锁：
- with_for_update 需要在事务中使用；示例通过 with SessionLocal.begin() 确保开启事务并在块结束时自动提交或回滚。
- 并发环境中，如果不希望等待锁，可以传 lock_nowait=True 并捕获数据库异常（如 Postgres 的 LockNotAvailable）。
- SQLite 差异：
- SQLite 会忽略 FOR UPDATE 子句，锁语义不同（更偏向整表/数据库级），不适合高并发扣减场景。生产建议使用支持行级锁的数据库（PostgreSQL、MySQL/InnoDB）。
- 性能建议：
- 大批量 SKU（如 > 1k）时 IN 子句可能较长，可分批处理或使用临时表/VALUES 表配合单条 UPDATE+RETURNING（若使用 PostgreSQL 可进一步优化）。当前实现重在跨库兼容和可读性。
- 数据校验：
- 代码已过滤 delta <= 0 项。根据需要可增加对 sku 字符串长度、字符集等校验。
- 返回结果：
- affected_rows 为成功扣减的记录条数，与 successes 长度一致。
- skipped 会包含库存不足、未找到/不满足条件和非法 delta 的条目，并尽可能附带 current_stock。
- 安全：
- 全程仅使用 SQLAlchemy 参数绑定与 ORM 更新，无原始字符串拼接，避免 SQL 注入风险。
- 审计字段：
- 如需记录更新时间，可在模型中添加 updated_at = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())。
- 隔离级别：
- 默认 READ COMMITTED 对大多数场景足够；若有更严格一致性需求，可在 Engine 层调整隔离级别，但要注意死锁与吞吐的平衡。
