更新登录页及优化相关功能

shierfeng-fastapi-backend/module_business/dao/master_dao.py

@@ -2,7 +2,7 @@ from datetime import datetime
 from typing import Any
 
 from pydantic import BaseModel
-from sqlalchemy import ColumnElement, Select, select, update
+from sqlalchemy import ColumnElement, Select, func, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm.attributes import InstrumentedAttribute
 
@@ -15,6 +15,7 @@ from module_business.entity.do.master_do import (
     BizPallet,
     BizProductionLine,
     BizStrain,
+    BizTunnel,
     BizUserSite,
     BizUserWarehouse,
     BizWarehouse,
@@ -31,6 +32,8 @@ from module_business.entity.vo.master_vo import (
     SitePageQueryModel,
     StrainModel,
     StrainPageQueryModel,
+    TunnelModel,
+    TunnelPageQueryModel,
     WarehouseModel,
     WarehousePageQueryModel,
 )
@@ -38,7 +41,7 @@ from utils.page_util import PageUtil
 
 
 class MasterDao:
-    """菌种、客户、现场、产线和仓库主数据数据库操作层。"""
+    """菌种、客户、现场、产线、隧道和仓库主数据数据库操作层。"""
 
     @staticmethod
     async def _paginate(
@@ -339,6 +342,67 @@ class MasterDao:
     async def update_line(cls, db: AsyncSession, line: LineModel, scope: TestBatchScope) -> None:
         await cls._update(db, BizProductionLine, BizProductionLine.line_id, 'line_id', line, scope)
 
+    @classmethod
+    async def list_tunnel(
+        cls,
+        db: AsyncSession,
+        query_object: TunnelPageQueryModel,
+        scope: TestBatchScope,
+        is_page: bool = True,
+    ) -> PageModel | list[dict[str, Any]]:
+        conditions = [batch_scope_condition(BizTunnel.test_batch_id, scope)]
+        if query_object.site_id:
+            conditions.append(BizTunnel.site_id == query_object.site_id)
+        if query_object.tunnel_code:
+            conditions.append(BizTunnel.tunnel_code.like(f'%{query_object.tunnel_code}%'))
+        if query_object.tunnel_name:
+            conditions.append(BizTunnel.tunnel_name.like(f'%{query_object.tunnel_name}%'))
+        if query_object.status:
+            conditions.append(BizTunnel.status == query_object.status)
+        statement = select(BizTunnel).where(*conditions).order_by(BizTunnel.tunnel_id)
+        return await cls._paginate(db, statement, query_object, is_page)
+
+    @classmethod
+    async def get_tunnel(
+        cls,
+        db: AsyncSession,
+        tunnel_id: int,
+        scope: TestBatchScope,
+        *,
+        for_update: bool = False,
+    ) -> BizTunnel | None:
+        return await cls._get(
+            db,
+            BizTunnel,
+            BizTunnel.tunnel_id,
+            tunnel_id,
+            batch_scope_condition(BizTunnel.test_batch_id, scope),
+            for_update=for_update,
+        )
+
+    @classmethod
+    async def find_tunnel_by_code(
+        cls,
+        db: AsyncSession,
+        tunnel_code: str,
+        site_id: int,
+    ) -> BizTunnel | None:
+        return await cls._find(
+            db,
+            BizTunnel,
+            BizTunnel.tunnel_code,
+            tunnel_code,
+            BizTunnel.site_id == site_id,
+        )
+
+    @classmethod
+    async def add_tunnel(cls, db: AsyncSession, tunnel: TunnelModel, test_batch_id: int | None) -> BizTunnel:
+        return await cls._add(db, BizTunnel, 'tunnel_id', tunnel, test_batch_id)
+
+    @classmethod
+    async def update_tunnel(cls, db: AsyncSession, tunnel: TunnelModel, scope: TestBatchScope) -> None:
+        await cls._update(db, BizTunnel, BizTunnel.tunnel_id, 'tunnel_id', tunnel, scope)
+
     @classmethod
     async def list_warehouse(
         cls,
@@ -475,6 +539,25 @@ class MasterDao:
         result = await db.execute(statement)
         return result.scalar_one_or_none()
 
+    @classmethod
+    async def count_tunnel_trace_members(
+        cls,
+        db: AsyncSession,
+        tunnel_id: int,
+        trace_no: str,
+        scope: TestBatchScope,
+    ) -> int:
+        result = await db.execute(
+            select(func.count())
+            .select_from(BizPalletInventory)
+            .where(
+                BizPalletInventory.tunnel_id == tunnel_id,
+                BizPalletInventory.trace_no == trace_no,
+                batch_scope_condition(BizPalletInventory.test_batch_id, scope),
+            )
+        )
+        return int(result.scalar_one() or 0)
+
     @classmethod
     async def add_pallet_inventory(
         cls,