83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.utils import generate_uuid
|
|
from app.models.qc_param import BizQcItemParam
|
|
|
|
|
|
def _row_to_dict(r: BizQcItemParam) -> Dict[str, Any]:
|
|
return {
|
|
"id": r.id,
|
|
"paramName": r.param_name,
|
|
"paramCode": r.param_code,
|
|
"paramUnit": r.param_unit or "",
|
|
"paramType": r.param_type or "text",
|
|
"requiredFlag": r.required_flag or "0",
|
|
"orderNum": r.order_num if r.order_num is not None else 0,
|
|
}
|
|
|
|
|
|
def list_params_by_item(
|
|
db: Session, item_source: str, item_id: str, tenant_no: str, *, user_info=None
|
|
) -> List[Dict[str, Any]]:
|
|
q = db.query(BizQcItemParam).filter(
|
|
BizQcItemParam.item_source == item_source,
|
|
BizQcItemParam.item_id == item_id,
|
|
BizQcItemParam.tenant_no == tenant_no,
|
|
BizQcItemParam.del_flag == 0,
|
|
)
|
|
if user_info:
|
|
from app.core.data_scope import apply_data_scope
|
|
q = apply_data_scope(q, BizQcItemParam, user_info, db, tenant_no)
|
|
rows = q.order_by(BizQcItemParam.order_num.asc()).all()
|
|
return [_row_to_dict(r) for r in rows]
|
|
|
|
|
|
def save_params_for_item(
|
|
db: Session,
|
|
item_source: str,
|
|
item_id: str,
|
|
params: List[Dict[str, Any]],
|
|
tenant_no: str,
|
|
user_id: Optional[str] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
now = datetime.now()
|
|
db.query(BizQcItemParam).filter(
|
|
BizQcItemParam.item_source == item_source,
|
|
BizQcItemParam.item_id == item_id,
|
|
BizQcItemParam.tenant_no == tenant_no,
|
|
BizQcItemParam.del_flag == 0,
|
|
).update(
|
|
{"del_flag": 1, "update_by": user_id, "update_time": now},
|
|
synchronize_session="fetch",
|
|
)
|
|
db.flush()
|
|
|
|
for idx, p in enumerate(params):
|
|
obj = BizQcItemParam(
|
|
id=generate_uuid(),
|
|
item_source=item_source,
|
|
item_id=item_id,
|
|
param_name=p.get("paramName", ""),
|
|
param_code=p.get("paramCode", ""),
|
|
param_unit=p.get("paramUnit") or None,
|
|
param_type=p.get("paramType", "text"),
|
|
required_flag=p.get("requiredFlag", "0"),
|
|
tenant_no=tenant_no,
|
|
order_num=p.get("orderNum", idx),
|
|
create_by=user_id,
|
|
create_dept_id="",
|
|
create_time=now,
|
|
update_by=user_id,
|
|
update_time=now,
|
|
del_flag=0,
|
|
)
|
|
db.add(obj)
|
|
|
|
db.commit()
|
|
return list_params_by_item(db, item_source, item_id, tenant_no)
|