补充主数据删除能力契约

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

@@ -78,6 +78,14 @@ _REFERENCE_COLUMNS: dict[MasterResource, tuple[tuple[InstrumentedAttribute, ...]
     ),
 }
 
+_MASTER_ID_COLUMNS: dict[MasterResource, InstrumentedAttribute] = {
+    'strain': BizStrain.strain_id,
+    'customer': BizCustomer.customer_id,
+    'site': BizCustomerSite.site_id,
+    'line': BizProductionLine.line_id,
+    'warehouse': BizWarehouse.warehouse_id,
+}
+
 _PALLET_REMOVAL_BLOCKING_COLUMNS: tuple[InstrumentedAttribute, ...] = (
     BizInboundItem.pallet_id,
     BizFillingRecord.pallet_id,
@@ -139,7 +147,7 @@ class MasterDao:
         id_field: str,
         page_object: BaseModel,
     ) -> Any:
-        payload = page_object.model_dump(exclude={id_field})
+        payload = page_object.model_dump(exclude={id_field, 'can_remove'})
         record = model(**payload)
         db.add(record)
         await db.flush()
@@ -155,7 +163,7 @@ class MasterDao:
     ) -> None:
         payload = page_object.model_dump(
             exclude_unset=True,
-            exclude={'create_by', 'create_time'},
+            exclude={'can_remove', 'create_by', 'create_time'},
         )
         resource_id = payload.pop(id_field)
         await db.execute(update(model).where(id_column == resource_id).values(**payload))
@@ -169,6 +177,38 @@ class MasterDao:
     ) -> None:
         await db.execute(delete(model).where(id_column == resource_id))
 
+    @classmethod
+    async def get_master_removability(
+        cls,
+        db: AsyncSession,
+        resource_type: MasterResource,
+        resource_ids: list[int],
+    ) -> dict[int, bool]:
+        """按删除约束批量判断五类主数据是否允许删除。"""
+        removability = dict.fromkeys(resource_ids, False)
+        if not resource_ids:
+            return removability
+
+        id_column = _MASTER_ID_COLUMNS.get(resource_type)
+        references = _REFERENCE_COLUMNS.get(resource_type)
+        if id_column is None or references is None:
+            raise ValueError(f'不支持的主数据类型：{resource_type}')
+
+        has_no_references = []
+        for columns in references:
+            predicates = [column == id_column for column in columns]
+            predicate = predicates[0] if len(predicates) == 1 else or_(*predicates)
+            has_no_references.append(~select(columns[0]).where(predicate).exists())
+
+        statement = select(
+            id_column,
+            and_(*has_no_references).label('can_remove'),
+        ).where(id_column.in_(resource_ids))
+        result = await db.execute(statement)
+        for resource_id, can_remove in result.all():
+            removability[resource_id] = bool(can_remove)
+        return removability
+
     @classmethod
     async def list_strain(
         cls,