liuwenxinandClaude Code
1625907380
- 判定=按 sort_order 升序首个 min_score≤总分，边界相邻不算重叠 - 新增同样执行重叠判定（判定表语义下插出重叠即脏配置） - 等级配置无下游引用（评级为纯计算器，对照表 #25），允许清空 Co-Authored-By: Claude Code <noreply@anthropic.com>
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
"""客户评级等级配置数据访问。
|
|
|
|
本层只 flush 不 commit —— 事务边界由 services 层控制。
|
|
"""
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import CustomerRatingGrade
|
|
|
|
# ==================== 读取 ====================
|
|
|
|
|
|
def list_all(session: Session) -> list[CustomerRatingGrade]:
|
|
"""全量等级，按 sort_order 升序、id 兜底（同序次序稳定）。
|
|
|
|
不分页：等级映射是判定表，数量个位数，分页无意义。
|
|
"""
|
|
stmt = select(CustomerRatingGrade).order_by(
|
|
CustomerRatingGrade.sort_order, CustomerRatingGrade.id
|
|
)
|
|
return list(session.execute(stmt).scalars())
|
|
|
|
|
|
def get_by_id(session: Session, grade_id: int) -> CustomerRatingGrade | None:
|
|
"""按主键查询。
|
|
|
|
等级表无下游引用（评级是纯计算器不落评分记录），无需 ``for_update``
|
|
锁行 —— product_repository 锁行防的是并发引用悬空，此处不存在该风险。
|
|
"""
|
|
return session.get(CustomerRatingGrade, grade_id)
|
|
|
|
|
|
def get_by_name(session: Session, grade_name: str) -> CustomerRatingGrade | None:
|
|
"""按等级名称查询。
|
|
|
|
grade_name 是 uk_customer_rating_grades_grade_name 唯一键的键值，先查
|
|
用于给出即时业务提示；并发窗口仍存在，最终由 IntegrityError 兜底
|
|
（见 rating_service）。
|
|
"""
|
|
stmt = select(CustomerRatingGrade).where(CustomerRatingGrade.grade_name == grade_name)
|
|
return session.execute(stmt).scalar_one_or_none()
|
|
|
|
|
|
# ==================== 写入 ====================
|
|
|
|
|
|
def create(session: Session, grade: CustomerRatingGrade) -> CustomerRatingGrade:
|
|
"""新增等级。只 flush 以取得自增 id。"""
|
|
session.add(grade)
|
|
session.flush()
|
|
return grade
|
|
|
|
|
|
def delete(session: Session, grade: CustomerRatingGrade) -> None:
|
|
"""物理删除等级。等级表无下游引用，调用方无需引用保护校验。"""
|
|
session.delete(grade)
|
|
session.flush()
|