91fc6f6293
- delete_product 补 count_refs 引用检查（业务明细+发货明细 RESTRICT 外键）与并发窗口 IntegrityError→400 兜底（customer 先例同构）， docstring 与两个 api 层过时注释同步更新 - 产品 schema 复用 schemas/business._check_amount：sale_price 走 (PRICE_MAX, scale=4)、stock_quantity 走 AMOUNT_MAX——此前私写 _parse_non_negative 不查小数位与容量，5 位小数被 Numeric(14,4) 静默截断（舍入即算错钱）；新增 6 个秒级 schema 单测锁定口径 - 全局注册 IntegrityError handler：漏补引用保护的删除类操作收敛为 400 通用文案（e2e 撞见一个补一个的模式无法保证全覆盖），完整 异常记日志；rating/customer 的「非 1062 按 500」注释同步更正 Co-Authored-By: Claude Code <noreply@anthropic.com>
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
"""产品主数据数据访问。
|
||
|
||
本层只 flush 不 commit —— 事务边界由 services 层控制。
|
||
"""
|
||
|
||
from sqlalchemy import Select, func, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import BusinessItem, Product, ShipmentItem
|
||
|
||
# ==================== 读取 ====================
|
||
|
||
|
||
def get_by_id(session: Session, product_id: int, *, for_update: bool = False) -> Product | None:
|
||
"""按主键查询。删除校验走 ``for_update`` 锁行（同 business_role_repository
|
||
的理由：校验与删除之间若发生并发引用，锁行使并发写入可见地失败，
|
||
而非引用被静默悬空）。"""
|
||
if for_update:
|
||
stmt = select(Product).where(Product.id == product_id).with_for_update()
|
||
return session.execute(stmt).scalar_one_or_none()
|
||
return session.get(Product, product_id)
|
||
|
||
|
||
def get_by_name_spec(session: Session, product_name: str, spec_model: str) -> Product | None:
|
||
"""按 (名称, 规格型号) 组合查询。
|
||
|
||
组合是 uk_products_name_spec 唯一键的键值，先查用于给出即时业务提示；
|
||
并发窗口仍存在，最终由 IntegrityError 兜底（见 product_service）。
|
||
"""
|
||
stmt = select(Product).where(
|
||
Product.product_name == product_name,
|
||
Product.spec_model == spec_model,
|
||
)
|
||
return session.execute(stmt).scalar_one_or_none()
|
||
|
||
|
||
def list_products(
|
||
session: Session,
|
||
*,
|
||
offset: int,
|
||
limit: int,
|
||
keyword: str | None = None,
|
||
product_category: str | None = None,
|
||
status: int | None = None,
|
||
) -> list[Product]:
|
||
"""分页查询产品（含停用）。按 id 升序，与 M1 岗位列表的次序口径一致。"""
|
||
stmt = _apply_filters(
|
||
select(Product),
|
||
keyword=keyword,
|
||
product_category=product_category,
|
||
status=status,
|
||
)
|
||
stmt = stmt.order_by(Product.id).offset(offset).limit(limit)
|
||
return list(session.execute(stmt).scalars())
|
||
|
||
|
||
def count_products(
|
||
session: Session,
|
||
*,
|
||
keyword: str | None = None,
|
||
product_category: str | None = None,
|
||
status: int | None = None,
|
||
) -> int:
|
||
"""统计符合条件的产品总数，供分页使用。"""
|
||
stmt = _apply_filters(
|
||
select(func.count(Product.id)),
|
||
keyword=keyword,
|
||
product_category=product_category,
|
||
status=status,
|
||
)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
def _apply_filters(
|
||
stmt: Select,
|
||
*,
|
||
keyword: str | None,
|
||
product_category: str | None,
|
||
status: int | None,
|
||
) -> Select:
|
||
"""应用通用筛选条件。列表与计数共用，避免两处口径不一致。
|
||
|
||
关键词对名称/规格两列取 OR 命中。转义与 query_utils.fuzzy 同一机制
|
||
（``contains(autoescape=True)`` 转义 LIKE 通配符并附加 ESCAPE 子句）；
|
||
因 fuzzy 是单列 where 语义（两次调用会叠加成 AND），双列 OR 在此直接
|
||
写 contains —— M1 business_role_repository 的双列关键词同款写法。
|
||
规格型号常含 ``_``（如 DN300_II），不转义会当单字符通配执行。
|
||
"""
|
||
if keyword:
|
||
stmt = stmt.where(
|
||
or_(
|
||
Product.product_name.contains(keyword, autoescape=True),
|
||
Product.spec_model.contains(keyword, autoescape=True),
|
||
)
|
||
)
|
||
if product_category:
|
||
stmt = stmt.where(Product.product_category == product_category)
|
||
if status is not None:
|
||
stmt = stmt.where(Product.status == status)
|
||
return stmt
|
||
|
||
|
||
def count_refs(session: Session, product_id: int) -> int:
|
||
"""统计引用该产品的业务明细与发货明细行数（删除保护，business_id 先例）。
|
||
|
||
两张明细表对 products 均 RESTRICT 外键：先查给出可读的 400；校验与
|
||
删除之间的并发引用由锁行 + FK 兜底（service 层再翻译 IntegrityError）。
|
||
"""
|
||
stmt = select(
|
||
select(func.count())
|
||
.select_from(BusinessItem)
|
||
.where(BusinessItem.product_id == product_id)
|
||
.scalar_subquery()
|
||
+ select(func.count())
|
||
.select_from(ShipmentItem)
|
||
.where(ShipmentItem.product_id == product_id)
|
||
.scalar_subquery()
|
||
)
|
||
return int(session.execute(stmt).scalar_one())
|
||
|
||
|
||
# ==================== 写入 ====================
|
||
|
||
|
||
def create(session: Session, product: Product) -> Product:
|
||
"""新增产品。只 flush 以取得自增 id。"""
|
||
session.add(product)
|
||
session.flush()
|
||
return product
|
||
|
||
|
||
def delete(session: Session, product: Product) -> None:
|
||
"""物理删除产品。调用方须先完成引用保护校验（count_refs + FK 兜底）。"""
|
||
session.delete(product)
|
||
session.flush()
|