from typing import Any, Literal from pydantic import BaseModel from sqlalchemy import ColumnElement, Select, delete, func, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import InstrumentedAttribute from common.vo import PageModel from module_business.entity.do.master_do import ( BizCustomer, BizCustomerSite, BizProductionLine, BizStrain, BizUserSite, BizUserWarehouse, BizWarehouse, ) from module_business.entity.do.operation_do import ( BizFillingRecord, BizInboundOrder, BizPalletEvent, BizPalletInventory, BizReturnOrder, BizShipmentOrder, BizUsageRecord, ) from module_business.entity.vo.master_vo import ( CustomerModel, CustomerPageQueryModel, LineModel, LinePageQueryModel, SiteModel, SitePageQueryModel, StrainModel, StrainPageQueryModel, WarehouseModel, WarehousePageQueryModel, ) from utils.page_util import PageUtil MasterResource = Literal['strain', 'customer', 'site', 'line', 'warehouse'] _REFERENCE_COLUMNS: dict[MasterResource, tuple[tuple[InstrumentedAttribute, ...], ...]] = { 'strain': ((BizFillingRecord.strain_id,),), 'customer': ((BizCustomerSite.customer_id,),), 'site': ( (BizProductionLine.site_id,), (BizUserSite.site_id,), (BizShipmentOrder.target_site_id,), (BizUsageRecord.site_id,), (BizReturnOrder.source_site_id,), (BizPalletInventory.site_id,), (BizPalletEvent.before_site_id, BizPalletEvent.after_site_id), ), 'line': ( (BizUsageRecord.line_id,), (BizPalletInventory.current_line_id,), (BizPalletEvent.before_line_id, BizPalletEvent.after_line_id), ), 'warehouse': ( (BizUserWarehouse.warehouse_id,), (BizInboundOrder.warehouse_id,), (BizFillingRecord.warehouse_id,), (BizShipmentOrder.source_warehouse_id,), (BizReturnOrder.target_warehouse_id,), (BizPalletInventory.warehouse_id,), (BizPalletEvent.before_warehouse_id, BizPalletEvent.after_warehouse_id), ), } class MasterDao: """菌种、客户、现场、产线和仓库主数据数据库操作层。""" @staticmethod async def _paginate( db: AsyncSession, statement: Select, query_object: BaseModel, is_page: bool, ) -> PageModel | list[dict[str, Any]]: return await PageUtil.paginate( db, statement, query_object.page_num, query_object.page_size, is_page, ) @staticmethod async def _get( db: AsyncSession, model: type[Any], id_column: InstrumentedAttribute, resource_id: int, ) -> Any | None: result = await db.execute(select(model).where(id_column == resource_id)) return result.scalar_one_or_none() @staticmethod async def _find( db: AsyncSession, model: type[Any], code_column: InstrumentedAttribute, code: str, *scope_predicates: ColumnElement[bool], ) -> Any | None: result = await db.execute(select(model).where(code_column == code, *scope_predicates)) return result.scalar_one_or_none() @staticmethod async def _add( db: AsyncSession, model: type[Any], id_field: str, page_object: BaseModel, ) -> Any: payload = page_object.model_dump(exclude={id_field}) record = model(**payload) db.add(record) await db.flush() return record @staticmethod async def _update( db: AsyncSession, model: type[Any], id_column: InstrumentedAttribute, id_field: str, page_object: BaseModel, ) -> None: payload = page_object.model_dump( exclude_unset=True, exclude={'create_by', 'create_time'}, ) resource_id = payload.pop(id_field) await db.execute(update(model).where(id_column == resource_id).values(**payload)) @staticmethod async def _delete( db: AsyncSession, model: type[Any], id_column: InstrumentedAttribute, resource_id: int, ) -> None: await db.execute(delete(model).where(id_column == resource_id)) @classmethod async def list_strain( cls, db: AsyncSession, query_object: StrainPageQueryModel, is_page: bool = True, ) -> PageModel | list[dict[str, Any]]: conditions = [] if query_object.strain_code: conditions.append(BizStrain.strain_code.like(f'%{query_object.strain_code}%')) if query_object.strain_name: conditions.append(BizStrain.strain_name.like(f'%{query_object.strain_name}%')) if query_object.status: conditions.append(BizStrain.status == query_object.status) statement = select(BizStrain).where(*conditions).order_by(BizStrain.strain_id) return await cls._paginate(db, statement, query_object, is_page) @classmethod async def get_strain(cls, db: AsyncSession, strain_id: int) -> BizStrain | None: return await cls._get(db, BizStrain, BizStrain.strain_id, strain_id) @classmethod async def find_strain_by_code(cls, db: AsyncSession, strain_code: str) -> BizStrain | None: return await cls._find(db, BizStrain, BizStrain.strain_code, strain_code) @classmethod async def add_strain(cls, db: AsyncSession, strain: StrainModel) -> BizStrain: return await cls._add(db, BizStrain, 'strain_id', strain) @classmethod async def update_strain(cls, db: AsyncSession, strain: StrainModel) -> None: await cls._update(db, BizStrain, BizStrain.strain_id, 'strain_id', strain) @classmethod async def delete_strain(cls, db: AsyncSession, strain_id: int) -> None: await cls._delete(db, BizStrain, BizStrain.strain_id, strain_id) @classmethod async def list_customer( cls, db: AsyncSession, query_object: CustomerPageQueryModel, is_page: bool = True, ) -> PageModel | list[dict[str, Any]]: conditions = [] if query_object.customer_code: conditions.append(BizCustomer.customer_code.like(f'%{query_object.customer_code}%')) if query_object.customer_name: conditions.append(BizCustomer.customer_name.like(f'%{query_object.customer_name}%')) if query_object.status: conditions.append(BizCustomer.status == query_object.status) statement = select(BizCustomer).where(*conditions).order_by(BizCustomer.customer_id) return await cls._paginate(db, statement, query_object, is_page) @classmethod async def get_customer(cls, db: AsyncSession, customer_id: int) -> BizCustomer | None: return await cls._get(db, BizCustomer, BizCustomer.customer_id, customer_id) @classmethod async def find_customer_by_code(cls, db: AsyncSession, customer_code: str) -> BizCustomer | None: return await cls._find(db, BizCustomer, BizCustomer.customer_code, customer_code) @classmethod async def add_customer(cls, db: AsyncSession, customer: CustomerModel) -> BizCustomer: return await cls._add(db, BizCustomer, 'customer_id', customer) @classmethod async def update_customer(cls, db: AsyncSession, customer: CustomerModel) -> None: await cls._update(db, BizCustomer, BizCustomer.customer_id, 'customer_id', customer) @classmethod async def delete_customer(cls, db: AsyncSession, customer_id: int) -> None: await cls._delete(db, BizCustomer, BizCustomer.customer_id, customer_id) @classmethod async def list_site( cls, db: AsyncSession, query_object: SitePageQueryModel, is_page: bool = True, ) -> PageModel | list[dict[str, Any]]: conditions = [] if query_object.customer_id: conditions.append(BizCustomerSite.customer_id == query_object.customer_id) if query_object.site_code: conditions.append(BizCustomerSite.site_code.like(f'%{query_object.site_code}%')) if query_object.site_name: conditions.append(BizCustomerSite.site_name.like(f'%{query_object.site_name}%')) if query_object.status: conditions.append(BizCustomerSite.status == query_object.status) statement = select(BizCustomerSite).where(*conditions).order_by(BizCustomerSite.site_id) return await cls._paginate(db, statement, query_object, is_page) @classmethod async def get_site(cls, db: AsyncSession, site_id: int) -> BizCustomerSite | None: return await cls._get(db, BizCustomerSite, BizCustomerSite.site_id, site_id) @classmethod async def find_site_by_code( cls, db: AsyncSession, site_code: str, customer_id: int, ) -> BizCustomerSite | None: return await cls._find( db, BizCustomerSite, BizCustomerSite.site_code, site_code, BizCustomerSite.customer_id == customer_id, ) @classmethod async def add_site(cls, db: AsyncSession, site: SiteModel) -> BizCustomerSite: return await cls._add(db, BizCustomerSite, 'site_id', site) @classmethod async def update_site(cls, db: AsyncSession, site: SiteModel) -> None: await cls._update(db, BizCustomerSite, BizCustomerSite.site_id, 'site_id', site) @classmethod async def delete_site(cls, db: AsyncSession, site_id: int) -> None: await cls._delete(db, BizCustomerSite, BizCustomerSite.site_id, site_id) @classmethod async def list_line( cls, db: AsyncSession, query_object: LinePageQueryModel, is_page: bool = True, ) -> PageModel | list[dict[str, Any]]: conditions = [] if query_object.site_id: conditions.append(BizProductionLine.site_id == query_object.site_id) if query_object.line_code: conditions.append(BizProductionLine.line_code.like(f'%{query_object.line_code}%')) if query_object.line_name: conditions.append(BizProductionLine.line_name.like(f'%{query_object.line_name}%')) if query_object.status: conditions.append(BizProductionLine.status == query_object.status) statement = select(BizProductionLine).where(*conditions).order_by(BizProductionLine.line_id) return await cls._paginate(db, statement, query_object, is_page) @classmethod async def get_line(cls, db: AsyncSession, line_id: int) -> BizProductionLine | None: return await cls._get(db, BizProductionLine, BizProductionLine.line_id, line_id) @classmethod async def find_line_by_code( cls, db: AsyncSession, line_code: str, site_id: int, ) -> BizProductionLine | None: return await cls._find( db, BizProductionLine, BizProductionLine.line_code, line_code, BizProductionLine.site_id == site_id, ) @classmethod async def add_line(cls, db: AsyncSession, line: LineModel) -> BizProductionLine: return await cls._add(db, BizProductionLine, 'line_id', line) @classmethod async def update_line(cls, db: AsyncSession, line: LineModel) -> None: await cls._update(db, BizProductionLine, BizProductionLine.line_id, 'line_id', line) @classmethod async def delete_line(cls, db: AsyncSession, line_id: int) -> None: await cls._delete(db, BizProductionLine, BizProductionLine.line_id, line_id) @classmethod async def list_warehouse( cls, db: AsyncSession, query_object: WarehousePageQueryModel, is_page: bool = True, ) -> PageModel | list[dict[str, Any]]: conditions = [] if query_object.warehouse_code: conditions.append(BizWarehouse.warehouse_code.like(f'%{query_object.warehouse_code}%')) if query_object.warehouse_name: conditions.append(BizWarehouse.warehouse_name.like(f'%{query_object.warehouse_name}%')) if query_object.status: conditions.append(BizWarehouse.status == query_object.status) statement = select(BizWarehouse).where(*conditions).order_by(BizWarehouse.warehouse_id) return await cls._paginate(db, statement, query_object, is_page) @classmethod async def get_warehouse(cls, db: AsyncSession, warehouse_id: int) -> BizWarehouse | None: return await cls._get(db, BizWarehouse, BizWarehouse.warehouse_id, warehouse_id) @classmethod async def find_warehouse_by_code(cls, db: AsyncSession, warehouse_code: str) -> BizWarehouse | None: return await cls._find(db, BizWarehouse, BizWarehouse.warehouse_code, warehouse_code) @classmethod async def add_warehouse(cls, db: AsyncSession, warehouse: WarehouseModel) -> BizWarehouse: return await cls._add(db, BizWarehouse, 'warehouse_id', warehouse) @classmethod async def update_warehouse(cls, db: AsyncSession, warehouse: WarehouseModel) -> None: await cls._update(db, BizWarehouse, BizWarehouse.warehouse_id, 'warehouse_id', warehouse) @classmethod async def delete_warehouse(cls, db: AsyncSession, warehouse_id: int) -> None: await cls._delete(db, BizWarehouse, BizWarehouse.warehouse_id, warehouse_id) @classmethod async def count_references( cls, db: AsyncSession, resource_type: MasterResource, resource_id: int, ) -> int: """统计主数据的所有直接外键引用;同表的多个位置字段按一行计数。""" references = _REFERENCE_COLUMNS.get(resource_type) if references is None: raise ValueError(f'不支持的主数据类型:{resource_type}') total = 0 for columns in references: predicates = [column == resource_id for column in columns] predicate = predicates[0] if len(predicates) == 1 else or_(*predicates) result = await db.execute(select(func.count('*')).select_from(columns[0].class_).where(predicate)) total += result.scalar() or 0 return total