158 lines
4.5 KiB
Python
158 lines
4.5 KiB
Python
"""字典数据 CRUD（sys_dict_data）"""
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import and_, func, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import generate_uuid
|
||
from app.models.models import SysDictData
|
||
|
||
|
||
def _row_to_dict(d: SysDictData) -> Dict[str, Any]:
|
||
return {
|
||
"id": d.id,
|
||
"dictLabel": d.dict_label,
|
||
"dictValue": d.dict_value,
|
||
"dictType": d.dict_type,
|
||
"tenantNo": d.tenant_no,
|
||
"orderNum": d.order_num if d.order_num is not None else 0,
|
||
"remark": d.remark or "",
|
||
"createTime": d.create_time.strftime("%Y-%m-%d %H:%M:%S") if d.create_time else "",
|
||
"updateTime": d.update_time.strftime("%Y-%m-%d %H:%M:%S") if d.update_time else "",
|
||
}
|
||
|
||
|
||
def list_dict_data(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
dict_type_code: Optional[str] = None,
|
||
keyword: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
def _filtered():
|
||
q0 = db.query(SysDictData).filter(SysDictData.del_flag == 0)
|
||
if tenant_no:
|
||
q0 = q0.filter(SysDictData.tenant_no == tenant_no)
|
||
if dict_type_code:
|
||
q0 = q0.filter(SysDictData.dict_type == dict_type_code)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q0 = q0.filter(SysDictData.dict_label.like(like))
|
||
return q0
|
||
|
||
total = _filtered().order_by(None).with_entities(func.count(SysDictData.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
_filtered()
|
||
.order_by(SysDictData.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
return [_row_to_dict(x) for x in rows], int(total)
|
||
|
||
|
||
def get_dict_data(db: Session, data_id: str) -> Optional[SysDictData]:
|
||
return (
|
||
db.query(SysDictData)
|
||
.filter(and_(SysDictData.id == data_id, SysDictData.del_flag == 0))
|
||
.first()
|
||
)
|
||
|
||
|
||
def create_dict_data(
|
||
db: Session,
|
||
*,
|
||
dict_label: str,
|
||
dict_value: str,
|
||
dict_type: str,
|
||
tenant_no: str,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> SysDictData:
|
||
obj = SysDictData(
|
||
id=generate_uuid(),
|
||
dict_label=(dict_label or "").strip(),
|
||
dict_value=(dict_value or "").strip(),
|
||
dict_type=(dict_type or "").strip(),
|
||
tenant_no=tenant_no,
|
||
order_num=order_num if order_num is not None else 0,
|
||
remark=(remark or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def update_dict_data(
|
||
db: Session,
|
||
data_id: str,
|
||
*,
|
||
dict_label: Optional[str] = None,
|
||
dict_value: Optional[str] = None,
|
||
order_num: Optional[int] = None,
|
||
remark: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
) -> SysDictData:
|
||
obj = get_dict_data(db, data_id)
|
||
if not obj:
|
||
raise ValueError("字典数据不存在")
|
||
if dict_label is not None:
|
||
obj.dict_label = (dict_label or "").strip()
|
||
if dict_value is not None:
|
||
obj.dict_value = (dict_value or "").strip()
|
||
if order_num is not None:
|
||
obj.order_num = order_num
|
||
if remark is not None:
|
||
obj.remark = (remark or "").strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_dict_data(db: Session, data_id: str) -> None:
|
||
obj = get_dict_data(db, data_id)
|
||
if not obj:
|
||
raise ValueError("字典数据不存在")
|
||
obj.del_flag = 1
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
|
||
|
||
def list_dict_data_options(
|
||
db: Session, *, dict_type_code: str, tenant_no: Optional[str] = None
|
||
) -> List[Dict[str, str]]:
|
||
q = db.query(SysDictData).filter(
|
||
and_(SysDictData.dict_type == dict_type_code, SysDictData.del_flag == 0)
|
||
)
|
||
if tenant_no:
|
||
q = q.filter(SysDictData.tenant_no == tenant_no)
|
||
rows = q.order_by(SysDictData.create_time.desc()).all()
|
||
return [{"label": r.dict_label, "value": r.dict_value} for r in rows]
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
if total <= 0:
|
||
return 0
|
||
return int(math.ceil(total / float(page_size)))
|