liuwenxin和Claude Code
ca88ee7a0b
- 覆盖：401/引用校验/删除保护/子表全量替换/类别与组合唯一/归属变更留痕/审计双记录 - 缺陷修复：personnel/product/customer 三 service 的前置校验移入审计上下文， 校验失败（400）同样留痕（成功与失败都记的既定原则） - conftest 兜底清理扩展到 M2 五表（留痕无外键，按 operator_name + 前缀双轨清理） Co-Authored-By: Claude Code <noreply@anthropic.com>
117 行
3.8 KiB
Python
117 行
3.8 KiB
Python
"""产品主数据数据访问。
|
||
|
||
本层只 flush 不 commit —— 事务边界由 services 层控制。
|
||
"""
|
||
|
||
from sqlalchemy import Select, func, or_, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import Product
|
||
|
||
# ==================== 读取 ====================
|
||
|
||
|
||
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 create(session: Session, product: Product) -> Product:
|
||
"""新增产品。只 flush 以取得自增 id。"""
|
||
session.add(product)
|
||
session.flush()
|
||
return product
|
||
|
||
|
||
def delete(session: Session, product: Product) -> None:
|
||
"""物理删除产品。调用方须先完成引用保护校验（V5 就位后生效）。"""
|
||
session.delete(product)
|
||
session.flush()
|