功能：实现托盘两阶段退回

shierfeng-fastapi-backend/module_business/controller/return_controller.py

@@ -0,0 +1,155 @@
+"""托盘两阶段退回接口。"""
+
+from typing import Annotated
+
+from fastapi import Path, Query, Request, Response
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.annotation.log_annotation import Log
+from common.aspect.db_seesion import DBSessionDependency
+from common.aspect.interface_auth import UserInterfaceAuthDependency
+from common.aspect.pre_auth import CurrentUserDependency, PreAuthDependency
+from common.enums import BusinessType
+from common.router import APIRouterPro
+from common.vo import DataResponseModel, PageResponseModel, ResponseBaseModel
+from module_admin.entity.vo.user_vo import CurrentUserModel
+from module_business.entity.vo.operation_vo import (
+    BusinessOptionModel,
+    PalletScanResponseModel,
+    ReturnConfirmModel,
+    ReturnModel,
+    ReturnPageQueryModel,
+    ReturnRequestModel,
+    ReturnScanQueryModel,
+)
+from module_business.service.return_service import ReturnService
+from utils.response_util import ResponseUtil
+
+return_controller = APIRouterPro(
+    prefix='/business/return',
+    tags=['业务管理-托盘退回'],
+    dependencies=[PreAuthDependency()],
+)
+
+
+@return_controller.get(
+    '/list',
+    summary='获取托盘退回单分页列表接口',
+    description='按当前用户来源现场或目标仓库数据范围获取退回单',
+    response_model=PageResponseModel[ReturnModel],
+    dependencies=[UserInterfaceAuthDependency('business:return:list')],
+)
+async def get_business_return_list(
+    request: Request,
+    return_page_query: Annotated[ReturnPageQueryModel, Query()],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    result = await ReturnService.get_return_list_services(
+        query_db,
+        return_page_query,
+        current_user,
+        is_page=True,
+    )
+    return ResponseUtil.success(model_content=result)
+
+
+@return_controller.get(
+    '/warehouse-options',
+    summary='获取退回目标仓库选项接口',
+    description='返回全部启用仓库作为客户发起退回的目标字典',
+    response_model=DataResponseModel[list[BusinessOptionModel]],
+    dependencies=[UserInterfaceAuthDependency('business:return:query')],
+)
+async def get_business_return_warehouse_options(
+    request: Request,
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    result = await ReturnService.get_warehouse_options_services(query_db, current_user)
+    return ResponseUtil.success(data=result)
+
+
+@return_controller.get(
+    '/scan/{qr_token}',
+    summary='扫描退回托盘接口',
+    description='按申请现场或确认仓库数据范围重新查询托盘退回状态',
+    response_model=DataResponseModel[PalletScanResponseModel],
+    dependencies=[UserInterfaceAuthDependency('business:return:query')],
+)
+async def scan_business_return_pallet(
+    request: Request,
+    qr_token: Annotated[
+        str,
+        Path(min_length=1, max_length=128, pattern=r'\S', description='二维码固定标识'),
+    ],
+    return_scan_query: Annotated[ReturnScanQueryModel, Query()],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    result = await ReturnService.scan_return_pallet_services(
+        query_db,
+        qr_token,
+        return_scan_query,
+        current_user,
+    )
+    return ResponseUtil.success(data=result)
+
+
+@return_controller.post(
+    '/request',
+    summary='发起托盘退回申请接口',
+    description='原子冻结同一客户现场的全部退回托盘并保存库存快照',
+    response_model=ResponseBaseModel,
+    dependencies=[UserInterfaceAuthDependency('business:return:request')],
+)
+@Log(title='托盘退回', business_type=BusinessType.INSERT)
+async def request_business_return(
+    request: Request,
+    request_command: ReturnRequestModel,
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    result = await ReturnService.request_return_services(query_db, request_command, current_user)
+    return ResponseUtil.success(msg=result.message)
+
+
+@return_controller.get(
+    '/{return_id}',
+    summary='获取托盘退回单详情接口',
+    description='按来源现场或目标仓库数据范围获取指定退回单',
+    response_model=DataResponseModel[ReturnModel],
+    dependencies=[UserInterfaceAuthDependency('business:return:query')],
+)
+async def query_detail_business_return(
+    request: Request,
+    return_id: Annotated[int, Path(gt=0, description='退回单ID')],
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    result = await ReturnService.return_detail_services(query_db, return_id, current_user)
+    return ResponseUtil.success(data=result)
+
+
+@return_controller.put(
+    '/{return_id}/confirm',
+    summary='确认托盘退回接口',
+    description='核对整单扫码集合和申请快照后原子移入目标仓库',
+    response_model=ResponseBaseModel,
+    dependencies=[UserInterfaceAuthDependency('business:return:confirm')],
+)
+@Log(title='托盘退回', business_type=BusinessType.UPDATE)
+async def confirm_business_return(
+    request: Request,
+    return_id: Annotated[int, Path(gt=0, description='退回单ID')],
+    confirm_command: ReturnConfirmModel,
+    query_db: Annotated[AsyncSession, DBSessionDependency()],
+    current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
+) -> Response:
+    result = await ReturnService.confirm_return_services(
+        query_db,
+        return_id,
+        confirm_command,
+        current_user,
+    )
+    return ResponseUtil.success(msg=result.message)

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

@@ -1,7 +1,7 @@
 from datetime import datetime, time, timedelta
 from typing import Any
 
-from sqlalchemy import ColumnElement, Select, delete, false, select, update
+from sqlalchemy import ColumnElement, Select, delete, false, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm.attributes import InstrumentedAttribute
 
@@ -94,6 +94,27 @@ class OperationDao:
             conditions.append(BizShipmentOrder.target_site_id.in_(scoped_site_ids) if scoped_site_ids else false())
         return conditions
 
+    @staticmethod
+    def _return_scope_conditions(
+        site_ids: list[int] | None,
+        warehouse_ids: list[int] | None,
+    ) -> list[ColumnElement[bool]]:
+        if site_ids is None and warehouse_ids is None:
+            return []
+
+        visibility_conditions: list[ColumnElement[bool]] = []
+        if site_ids is not None:
+            scoped_site_ids = sorted(set(site_ids))
+            visibility_conditions.append(
+                BizReturnOrder.source_site_id.in_(scoped_site_ids) if scoped_site_ids else false()
+            )
+        if warehouse_ids is not None:
+            scoped_warehouse_ids = sorted(set(warehouse_ids))
+            visibility_conditions.append(
+                BizReturnOrder.target_warehouse_id.in_(scoped_warehouse_ids) if scoped_warehouse_ids else false()
+            )
+        return [or_(*visibility_conditions)]
+
     @staticmethod
     async def _paginate(
         db: AsyncSession,
@@ -300,13 +321,32 @@ class OperationDao:
         cls,
         db: AsyncSession,
         return_id: int,
+        site_ids: list[int] | None = None,
+        warehouse_ids: list[int] | None = None,
     ) -> BizReturnOrder | None:
-        return await cls._get_for_update(
+        return_order = await cls._get_for_update(
             db,
             BizReturnOrder,
             BizReturnOrder.return_id,
             return_id,
+            *cls._return_scope_conditions(site_ids, warehouse_ids),
         )
+        if return_order is None:
+            return None
+
+        # 父单锁定后使用 locking read 获取当前明细，避免 MySQL RR 沿用旧快照。
+        item_result = await db.execute(
+            select(BizReturnItem)
+            .where(BizReturnItem.return_id == return_id)
+            .order_by(BizReturnItem.pallet_id, BizReturnItem.return_item_id)
+            .with_for_update()
+            .execution_options(populate_existing=True)
+        )
+        return_order.items = sorted(
+            item_result.scalars().all(),
+            key=lambda item: (item.pallet_id, item.return_item_id),
+        )
+        return return_order
 
     @classmethod
     async def get_active_filling_for_update(
@@ -387,12 +427,15 @@ class OperationDao:
         cls,
         db: AsyncSession,
         idempotency_key: str,
+        site_ids: list[int] | None = None,
+        warehouse_ids: list[int] | None = None,
     ) -> BizReturnOrder | None:
         return await cls._get_by_key(
             db,
             BizReturnOrder,
             BizReturnOrder.request_idempotency_key,
             idempotency_key,
+            *cls._return_scope_conditions(site_ids, warehouse_ids),
         )
 
     @classmethod
@@ -400,12 +443,15 @@ class OperationDao:
         cls,
         db: AsyncSession,
         idempotency_key: str,
+        site_ids: list[int] | None = None,
+        warehouse_ids: list[int] | None = None,
     ) -> BizReturnOrder | None:
         return await cls._get_by_key(
             db,
             BizReturnOrder,
             BizReturnOrder.confirm_idempotency_key,
             idempotency_key,
+            *cls._return_scope_conditions(site_ids, warehouse_ids),
         )
 
     @classmethod
@@ -718,8 +764,10 @@ class OperationDao:
         db: AsyncSession,
         query_object: ReturnPageQueryModel,
         is_page: bool = True,
+        site_ids: list[int] | None = None,
+        warehouse_ids: list[int] | None = None,
     ) -> PageModel | list[dict[str, Any]]:
-        conditions = []
+        conditions = cls._return_scope_conditions(site_ids, warehouse_ids)
         if query_object.return_no:
             conditions.append(BizReturnOrder.return_no.like(f'%{query_object.return_no}%'))
         if query_object.source_site_id:
@@ -743,6 +791,8 @@ class OperationDao:
         cls,
         db: AsyncSession,
         return_id: int,
+        site_ids: list[int] | None = None,
+        warehouse_ids: list[int] | None = None,
     ) -> BizReturnOrder | None:
         return await cls._get_order_detail(
             db,
@@ -752,6 +802,7 @@ class OperationDao:
             BizReturnItem,
             BizReturnItem.return_id,
             BizReturnItem.return_item_id,
+            *cls._return_scope_conditions(site_ids, warehouse_ids),
         )
 
     @classmethod

shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py

@@ -123,7 +123,7 @@ class TestDaoPublicContract(unittest.TestCase):
         expected = {
             'get_inbound_for_update': ('db', 'inbound_id', 'warehouse_ids'),
             'get_shipment_for_update': ('db', 'shipment_id', 'warehouse_ids', 'site_ids'),
-            'get_return_for_update': ('db', 'return_id'),
+            'get_return_for_update': ('db', 'return_id', 'site_ids', 'warehouse_ids'),
             'get_active_filling_for_update': ('db', 'filling_id'),
             'get_inbound_by_confirm_idempotency_key': ('db', 'idempotency_key'),
             'get_shipment_by_confirm_idempotency_key': (
@@ -134,8 +134,18 @@ class TestDaoPublicContract(unittest.TestCase):
             ),
             'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'),
             'get_usage_by_idempotency_key': ('db', 'idempotency_key', 'site_ids'),
-            'get_return_by_request_idempotency_key': ('db', 'idempotency_key'),
+            'get_return_by_request_idempotency_key': (
-            'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'),
+                'db',
+                'idempotency_key',
+                'site_ids',
+                'warehouse_ids',
+            ),
+            'get_return_by_confirm_idempotency_key': (
+                'db',
+                'idempotency_key',
+                'site_ids',
+                'warehouse_ids',
+            ),
             'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'),
             'add_inbound': ('db', 'inbound'),
             'update_inbound': ('db', 'inbound'),
@@ -151,7 +161,7 @@ class TestDaoPublicContract(unittest.TestCase):
             'replace_shipment_items': ('db', 'shipment_id', 'items'),
             'get_usage_by_id': ('db', 'usage_id', 'site_ids'),
             'add_usage_record': ('db', 'usage'),
-            'get_return_by_id': ('db', 'return_id'),
+            'get_return_by_id': ('db', 'return_id', 'site_ids', 'warehouse_ids'),
             'add_return_order': ('db', 'return_order'),
             'update_return_order': ('db', 'return_order'),
             'add_return_items': ('db', 'items'),
@@ -228,8 +238,22 @@ class TestDaoPublicContract(unittest.TestCase):
         self.assertNotIn('for_update', inspect.signature(OperationDao._get_by_key).parameters)
 
         return_list_signature = inspect.signature(OperationDao.get_return_list)
-        self.assertEqual(('db', 'query_object', 'is_page'), tuple(return_list_signature.parameters))
+        self.assertEqual(
+            ('db', 'query_object', 'is_page', 'site_ids', 'warehouse_ids'),
+            tuple(return_list_signature.parameters),
+        )
         self.assertIs(True, return_list_signature.parameters['is_page'].default)
+        for method_name in (
+            'get_return_for_update',
+            'get_return_by_request_idempotency_key',
+            'get_return_by_confirm_idempotency_key',
+            'get_return_by_id',
+            'get_return_list',
+        ):
+            with self.subTest(method=method_name):
+                signature = inspect.signature(getattr(OperationDao, method_name))
+                self.assertIsNone(signature.parameters['site_ids'].default)
+                self.assertIsNone(signature.parameters['warehouse_ids'].default)
 
     def test_dao_source_never_controls_transaction_boundary(self) -> None:
         for module in (inventory_dao_module, operation_dao_module):
@@ -423,13 +447,6 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
 
     async def test_for_update_queries_target_primary_keys_and_lock_rows(self) -> None:
         cases = (
-            (
-                OperationDao.get_return_for_update,
-                BizReturnOrder,
-                'return_id',
-                13,
-                (),
-            ),
             (
                 OperationDao.get_active_filling_for_update,
                 BizFillingRecord,