feat(crud): add cell product CRUD operations
This commit is contained in:
zlwu committed
1 parent
96a0c394c2
commit
b303afb272
1 file changed
+207
@@ -0,0 +1,207 @@
|
||||
"""细胞产品 CRUD（biz_cell_product）"""
|
||||
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 BizCellProduct
|
||||
|
||||
|
||||
def _row_to_dict(c: BizCellProduct) -> Dict[str, Any]:
|
||||
return {
|
||||
"id": c.id,
|
||||
"productCode": c.product_code,
|
||||
"productName": c.product_name,
|
||||
"productTypeId": c.product_type_id or "",
|
||||
"specId": c.spec_id or "",
|
||||
"batchNo": c.batch_no or "",
|
||||
"storageSlotId": c.storage_slot_id or "",
|
||||
"productStatus": c.product_status or "",
|
||||
"sourceType": c.source_type or "",
|
||||
"productionTaskId": c.production_task_id or "",
|
||||
"quantity": float(c.quantity) if c.quantity is not None else None,
|
||||
"unit": c.unit or "",
|
||||
"expiryDate": c.expiry_date.strftime("%Y-%m-%d %H:%M:%S") if c.expiry_date else "",
|
||||
"productImages": c.product_images or "",
|
||||
"tenantNo": c.tenant_no,
|
||||
"orderNum": c.order_num if c.order_num is not None else 0,
|
||||
"remark": c.remark or "",
|
||||
"createBy": c.create_by or "",
|
||||
"createTime": c.create_time.strftime("%Y-%m-%d %H:%M:%S") if c.create_time else "",
|
||||
"updateBy": c.update_by or "",
|
||||
"updateTime": c.update_time.strftime("%Y-%m-%d %H:%M:%S") if c.update_time else "",
|
||||
}
|
||||
|
||||
|
||||
def list_cell_products(
|
||||
db: Session,
|
||||
*,
|
||||
tenant_no: str,
|
||||
page: int = 1,
|
||||
page_size: int = 20,
|
||||
product_code: Optional[str] = None,
|
||||
product_name: Optional[str] = None,
|
||||
product_type_id: Optional[str] = None,
|
||||
product_status: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
def _filtered():
|
||||
q0 = db.query(BizCellProduct).filter(
|
||||
BizCellProduct.del_flag == 0,
|
||||
BizCellProduct.tenant_no == tenant_no,
|
||||
)
|
||||
pc = (product_code or "").strip()
|
||||
if pc:
|
||||
q0 = q0.filter(BizCellProduct.product_code.like(f"%{pc}%"))
|
||||
pn = (product_name or "").strip()
|
||||
if pn:
|
||||
q0 = q0.filter(BizCellProduct.product_name.like(f"%{pn}%"))
|
||||
if product_type_id:
|
||||
q0 = q0.filter(BizCellProduct.product_type_id == product_type_id)
|
||||
if product_status:
|
||||
q0 = q0.filter(BizCellProduct.product_status == product_status)
|
||||
return q0
|
||||
|
||||
total = _filtered().order_by(None).with_entities(func.count(BizCellProduct.id)).scalar() or 0
|
||||
page = max(1, page)
|
||||
page_size = min(max(1, page_size), 100)
|
||||
rows = (
|
||||
_filtered()
|
||||
.order_by(BizCellProduct.create_time.desc())
|
||||
.offset((page - 1) * page_size)
|
||||
.limit(page_size)
|
||||
.all()
|
||||
)
|
||||
return {"items": [_row_to_dict(x) for x in rows], "total": int(total)}
|
||||
|
||||
|
||||
def get_cell_product(db: Session, id: str, tenant_no: str) -> Optional[BizCellProduct]:
|
||||
return (
|
||||
db.query(BizCellProduct)
|
||||
.filter(
|
||||
and_(
|
||||
BizCellProduct.id == id,
|
||||
BizCellProduct.tenant_no == tenant_no,
|
||||
BizCellProduct.del_flag == 0,
|
||||
)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
|
||||
def create_cell_product(
|
||||
db: Session,
|
||||
*,
|
||||
tenant_no: str,
|
||||
user_id: Optional[str] = None,
|
||||
create_dept_id: Optional[str] = None,
|
||||
data: Any = None,
|
||||
) -> BizCellProduct:
|
||||
code = (data.product_code or "").strip()
|
||||
if not code:
|
||||
raise ValueError("产品编码不能为空")
|
||||
|
||||
existing = db.query(BizCellProduct).filter(
|
||||
BizCellProduct.tenant_no == tenant_no,
|
||||
BizCellProduct.product_code == code,
|
||||
BizCellProduct.del_flag == 0,
|
||||
).first()
|
||||
if existing:
|
||||
raise ValueError(f"产品编码 {code} 已存在")
|
||||
|
||||
obj = BizCellProduct(
|
||||
id=generate_uuid(),
|
||||
product_code=code,
|
||||
product_name=(data.product_name or "").strip(),
|
||||
product_type_id=(data.product_type_id or "").strip() or None,
|
||||
spec_id=(data.spec_id or "").strip() or None,
|
||||
batch_no=(data.batch_no or "").strip() or None,
|
||||
storage_slot_id=(data.storage_slot_id or "").strip() or None,
|
||||
product_status=data.product_status or "zk",
|
||||
source_type=data.source_type or "manual",
|
||||
quantity=data.quantity,
|
||||
unit=(data.unit or "").strip() or None,
|
||||
expiry_date=data.expiry_date,
|
||||
product_images=(data.product_images or "").strip() or None,
|
||||
tenant_no=tenant_no,
|
||||
order_num=0,
|
||||
remark=(data.remark or "").strip() or None,
|
||||
create_by=user_id,
|
||||
create_dept_id=create_dept_id or "",
|
||||
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_cell_product(
|
||||
db: Session,
|
||||
id: str,
|
||||
tenant_no: str,
|
||||
user_id: Optional[str] = None,
|
||||
data: Any = None,
|
||||
) -> BizCellProduct:
|
||||
obj = get_cell_product(db, id, tenant_no)
|
||||
if not obj:
|
||||
raise ValueError("细胞产品不存在")
|
||||
|
||||
if data.product_name is not None:
|
||||
obj.product_name = (data.product_name or "").strip()
|
||||
if data.product_type_id is not None:
|
||||
obj.product_type_id = (data.product_type_id or "").strip() or None
|
||||
if data.spec_id is not None:
|
||||
obj.spec_id = (data.spec_id or "").strip() or None
|
||||
if data.batch_no is not None:
|
||||
obj.batch_no = (data.batch_no or "").strip() or None
|
||||
if data.storage_slot_id is not None:
|
||||
obj.storage_slot_id = (data.storage_slot_id or "").strip() or None
|
||||
if data.product_status is not None:
|
||||
obj.product_status = data.product_status
|
||||
if data.quantity is not None:
|
||||
obj.quantity = data.quantity
|
||||
if data.unit is not None:
|
||||
obj.unit = (data.unit or "").strip() or None
|
||||
if data.expiry_date is not None:
|
||||
obj.expiry_date = data.expiry_date
|
||||
if data.product_images is not None:
|
||||
obj.product_images = (data.product_images or "").strip() or None
|
||||
if data.remark is not None:
|
||||
obj.remark = (data.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 delete_cell_product(
|
||||
db: Session,
|
||||
id: str,
|
||||
tenant_no: str,
|
||||
user_id: Optional[str] = None,
|
||||
) -> None:
|
||||
obj = get_cell_product(db, id, tenant_no)
|
||||
if not obj:
|
||||
raise ValueError("细胞产品不存在")
|
||||
obj.del_flag = 1
|
||||
obj.update_by = user_id
|
||||
obj.update_time = datetime.now()
|
||||
db.add(obj)
|
||||
db.commit()
|
||||
|
||||
|
||||
def total_pages(total: int, page_size: int) -> int:
|
||||
if total <= 0:
|
||||
return 0
|
||||
return int(math.ceil(total / float(page_size)))
|
||||
Reference in new issue
Block a user