"""计提参数数据访问。 本层只 flush 不 commit —— 事务边界由 services 层控制。 时点化解析查询(窗口函数版)留存 db/queries/计提参数.sql。 """ from datetime import date from sqlalchemy import Select, delete, func, select from sqlalchemy.orm import Session from app.models import ParamChangeBatch, ParamDefinition, ParamValue # ==================== 读取 ==================== def resolve_values(session: Session, biz_date: date) -> list[tuple[str, str, str, date]]: """时点化解析:每个 (param_id, dimension_value) 取生效日期 <= 业务日期的最新条目。 排序 (effective_date DESC, id DESC) 保证同日多次变更取最新一条 (不设日期唯一键,同日修正是合法留痕)。仅返回启用状态的参数定义。 返回 (param_code, dimension_value, value_text, effective_date)。 """ row_number = ( func.row_number() .over( partition_by=(ParamValue.param_id, ParamValue.dimension_value), order_by=(ParamValue.effective_date.desc(), ParamValue.id.desc()), ) .label("rn") ) subq = ( select( ParamDefinition.param_code.label("param_code"), ParamValue.dimension_value.label("dimension_value"), ParamValue.value_text.label("value_text"), ParamValue.effective_date.label("effective_date"), row_number, ) .join(ParamDefinition, ParamDefinition.id == ParamValue.param_id) .where(ParamValue.effective_date <= biz_date, ParamDefinition.status == 1) .subquery() ) stmt = select( subq.c.param_code, subq.c.dimension_value, subq.c.value_text, subq.c.effective_date ).where(subq.c.rn == 1) return [ (row.param_code, row.dimension_value, row.value_text, row.effective_date) for row in session.execute(stmt) ] def latest_value_before( session: Session, param_id: int, dimension_value: str, biz_date: date ) -> ParamValue | None: """取某参数某维度在业务日期(含)之前的最新条目 —— 提交变更时记录「旧值」。""" stmt = ( select(ParamValue) .where( ParamValue.param_id == param_id, ParamValue.dimension_value == dimension_value, ParamValue.effective_date <= biz_date, ) .order_by(ParamValue.effective_date.desc(), ParamValue.id.desc()) .limit(1) ) return session.execute(stmt).scalar_one_or_none() def list_definitions(session: Session, *, status: int | None = None) -> list[ParamDefinition]: """参数定义列表(目录)。""" stmt = select(ParamDefinition).order_by(ParamDefinition.sort_order, ParamDefinition.id) if status is not None: stmt = stmt.where(ParamDefinition.status == status) return list(session.execute(stmt).scalars()) def get_definition_by_code(session: Session, param_code: str) -> ParamDefinition | None: """按编码取参数定义。""" stmt = select(ParamDefinition).where(ParamDefinition.param_code == param_code) return session.execute(stmt).scalar_one_or_none() def list_definitions_by_codes( session: Session, param_codes: list[str] ) -> dict[str, ParamDefinition]: """按编码批量取参数定义(提交批次的校验段用,消除逐条 N+1)。 返回 {编码: 定义};未定义的编码不在结果中,由调用方判缺失。 """ if not param_codes: return {} rows = session.execute( select(ParamDefinition).where(ParamDefinition.param_code.in_(param_codes)) ).scalars() return {definition.param_code: definition for definition in rows} def list_history( session: Session, param_id: int, *, offset: int, limit: int, dimension_value: str | None = None, ) -> list[tuple[ParamValue, ParamChangeBatch]]: """单参数的版本历史(含批次信息),最新在前。""" stmt = _history_stmt(session, param_id, dimension_value=dimension_value) return [(value, batch) for value, batch in session.execute(stmt.offset(offset).limit(limit))] def count_history(session: Session, param_id: int, *, dimension_value: str | None = None) -> int: """统计单参数的历史条目数。""" stmt = select(func.count()).select_from( _history_stmt(session, param_id, dimension_value=dimension_value).subquery() ) return int(session.execute(stmt).scalar_one()) def _history_stmt(session: Session, param_id: int, *, dimension_value: str | None) -> Select: """历史查询基语句:条目 join 批次(版本号/操作人),最新在前。""" stmt = ( select(ParamValue, ParamChangeBatch) .join(ParamChangeBatch, ParamChangeBatch.id == ParamValue.batch_id) .where(ParamValue.param_id == param_id) .order_by(ParamValue.effective_date.desc(), ParamValue.id.desc()) ) if dimension_value is not None: stmt = stmt.where(ParamValue.dimension_value == dimension_value) return stmt # ==================== 批次 ==================== def create_batch(session: Session, batch: ParamChangeBatch) -> ParamChangeBatch: """新增变更批次。只 flush 以取得自增 id。""" session.add(batch) session.flush() return batch def get_batch_by_id( session: Session, batch_id: int, *, for_update: bool = False ) -> ParamChangeBatch | None: """按主键取批次。撤回校验走 ``for_update`` 锁行(防并发双撤)。""" if for_update: stmt = select(ParamChangeBatch).where(ParamChangeBatch.id == batch_id).with_for_update() return session.execute(stmt).scalar_one_or_none() return session.get(ParamChangeBatch, batch_id) def list_batches( session: Session, *, offset: int, limit: int, start_date: date | None = None, end_date: date | None = None, ) -> list[ParamChangeBatch]: """分页查询变更批次(按创建时间倒序)。""" stmt = _batch_filters(select(ParamChangeBatch), start_date=start_date, end_date=end_date) return list( session.execute( stmt.order_by(ParamChangeBatch.create_time.desc(), ParamChangeBatch.id.desc()) .offset(offset) .limit(limit) ).scalars() ) def count_batches( session: Session, *, start_date: date | None = None, end_date: date | None = None ) -> int: """统计符合条件的批次总数。""" stmt = _batch_filters( select(func.count(ParamChangeBatch.id)), start_date=start_date, end_date=end_date ) return int(session.execute(stmt).scalar_one()) def _batch_filters(stmt: Select, *, start_date: date | None, end_date: date | None) -> Select: """批次筛选(生效日期区间,含边界)。列表与计数共用。""" if start_date is not None: stmt = stmt.where(ParamChangeBatch.effective_date >= start_date) if end_date is not None: stmt = stmt.where(ParamChangeBatch.effective_date <= end_date) return stmt def list_values_by_batch( session: Session, batch_id: int ) -> list[tuple[ParamValue, ParamDefinition]]: """批次内全部条目(join 参数定义取编码/名称),按 id 排序。""" stmt = ( select(ParamValue, ParamDefinition) .join(ParamDefinition, ParamDefinition.id == ParamValue.param_id) .where(ParamValue.batch_id == batch_id) .order_by(ParamValue.id) ) return [(value, definition) for value, definition in session.execute(stmt)] def delete_values_by_batch(session: Session, batch_id: int) -> None: """删除批次下全部条目(撤回未生效批次时使用,2.0 风格 delete 语句)。""" session.execute(delete(ParamValue).where(ParamValue.batch_id == batch_id)) def delete_batch(session: Session, batch: ParamChangeBatch) -> None: """删除批次本体(调用方须先删条目并完成保护校验)。""" session.delete(batch) session.flush() # ==================== 写入 ==================== def create_value(session: Session, value: ParamValue) -> ParamValue: """新增版本条目(append-only,禁改禁删已生效条目)。只 flush。""" session.add(value) session.flush() return value # ==================== 导出 ==================== def list_export_rows( session: Session, *, start_date: date | None = None, end_date: date | None = None ) -> list[tuple[ParamChangeBatch, ParamValue, ParamDefinition]]: """导出行:批次 × 条目 × 定义(变更人/时间/参数/维度/旧值/新值/生效日期)。""" stmt = ( select(ParamChangeBatch, ParamValue, ParamDefinition) .join(ParamValue, ParamValue.batch_id == ParamChangeBatch.id) .join(ParamDefinition, ParamDefinition.id == ParamValue.param_id) .order_by(ParamChangeBatch.create_time.desc(), ParamChangeBatch.id.desc(), ParamValue.id) ) if start_date is not None: stmt = stmt.where(ParamChangeBatch.effective_date >= start_date) if end_date is not None: stmt = stmt.where(ParamChangeBatch.effective_date <= end_date) return [(batch, value, definition) for batch, value, definition in session.execute(stmt)]