85 lines
2.6 KiB
Python
85 lines
2.6 KiB
Python
"""样本储存证质检信息 API"""
|
|
from typing import Any, Dict, Optional
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import get_db
|
|
from app.core.menu_guard import require_any_menu_code
|
|
from app.core.response import error_response, success_response
|
|
from app.crud import sample_storage_cert_qc as svc
|
|
from app.models.models import SysUser
|
|
|
|
router = APIRouter(prefix="/web/admin", tags=["储存证质检信息"])
|
|
|
|
READ_MENUS = (
|
|
"biz:qcWorkOrder:list",
|
|
"biz:myQcTask:list",
|
|
"cell-storage:sample-info:list",
|
|
)
|
|
WRITE_MENUS = (
|
|
"biz:qcWorkOrder:list",
|
|
"biz:myQcTask:list",
|
|
)
|
|
|
|
|
|
def _get_tenant_no(db: Session, user: dict) -> str:
|
|
user_id = user.get("user_id")
|
|
if not user_id:
|
|
raise PermissionError("用户身份信息缺失，无法确定租户")
|
|
sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
|
|
if not sys_user or not sys_user.tenant_no:
|
|
raise PermissionError("用户租户信息缺失")
|
|
return sys_user.tenant_no
|
|
|
|
|
|
def _get_current_user_id(user: dict) -> str:
|
|
user_id = user.get("user_id")
|
|
if not user_id:
|
|
raise PermissionError("用户身份信息缺失")
|
|
return str(user_id)
|
|
|
|
|
|
class UpsertBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
payload: Dict[str, Any] = Field(default_factory=dict)
|
|
qc_work_order_id: Optional[str] = Field(None, alias="qcWorkOrderId")
|
|
|
|
|
|
@router.get("/sample-storage-cert-qc/{sample_code}")
|
|
async def get_sample_storage_cert_qc(
|
|
sample_code: str,
|
|
user: dict = Depends(require_any_menu_code(*READ_MENUS)),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
detail = svc.get_detail(db, tenant_no, sample_code)
|
|
if detail is None:
|
|
return error_response(404, "样本不存在")
|
|
return success_response(detail)
|
|
|
|
|
|
@router.put("/sample-storage-cert-qc/{sample_code}")
|
|
async def put_sample_storage_cert_qc(
|
|
sample_code: str,
|
|
body: UpsertBody,
|
|
user: dict = Depends(require_any_menu_code(*WRITE_MENUS)),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
user_id = _get_current_user_id(user)
|
|
try:
|
|
detail = svc.upsert(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
user_id=user_id,
|
|
sample_code=sample_code,
|
|
payload=body.payload,
|
|
qc_work_order_id=body.qc_work_order_id,
|
|
)
|
|
return success_response(detail, message="保存成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|