"""进销存主数据业务服务层 将 customer / supplier / warehouse 这类主数据 CRUD 编排从路由层下沉到此, 路由层只做参数校验与响应组装。 """ from datetime import datetime from typing import List, Optional, Type from fastapi import HTTPException from sqlalchemy import Select, select from sqlalchemy.ext.asyncio import AsyncSession from shared.models.identity import User from inventory.models import Customer, Supplier, Warehouse from ..schemas import ( CustomerCreate, CustomerResponse, SupplierCreate, SupplierResponse, WarehouseCreate, WarehouseResponse, ) def _generate_code(prefix: str) -> str: return f"{prefix}{datetime.now().strftime('%Y%m%d%H%M%S%f')}" async def _get_entity_or_404( db_session: AsyncSession, model: Type[Customer] | Type[Supplier] | Type[Warehouse], entity_id: int, detail: str, ): result = await db_session.execute(select(model).where(model.id == entity_id)) entity = result.scalar_one_or_none() if not entity: raise HTTPException(status_code=404, detail=detail) return entity def _build_customer_response(customer: Customer) -> CustomerResponse: return CustomerResponse.model_validate(customer) def _build_supplier_response(supplier: Supplier) -> SupplierResponse: return SupplierResponse.model_validate(supplier) def _build_warehouse_response(warehouse: Warehouse) -> WarehouseResponse: return WarehouseResponse.model_validate(warehouse) class MasterDataService: """客户 / 供应商 / 仓库主数据服务""" @staticmethod async def list_customers( db_session: AsyncSession, skip: int, limit: int, search: Optional[str], ) -> List[CustomerResponse]: query: Select = select(Customer).where(Customer.is_active == True) if search: query = query.where(Customer.name.ilike(f"%{search}%")) query = query.offset(skip).limit(limit).order_by(Customer.created_at.desc()) result = await db_session.execute(query) return [_build_customer_response(customer) for customer in result.scalars().all()] @staticmethod async def create_customer( db_session: AsyncSession, customer_data: CustomerCreate, current_user: User, ) -> CustomerResponse: data = customer_data.model_dump() if not data.get("code"): data["code"] = _generate_code("C") customer = Customer(**data) db_session.add(customer) await db_session.commit() await db_session.refresh(customer) return _build_customer_response(customer) @staticmethod async def update_customer( db_session: AsyncSession, customer_id: int, customer_data: CustomerCreate, current_user: User, ) -> CustomerResponse: customer = await _get_entity_or_404(db_session, Customer, customer_id, "客户不存在") for key, value in customer_data.model_dump().items(): setattr(customer, key, value) await db_session.commit() await db_session.refresh(customer) return _build_customer_response(customer) @staticmethod async def delete_customer( db_session: AsyncSession, customer_id: int, current_user: User, ) -> dict: customer = await _get_entity_or_404(db_session, Customer, customer_id, "客户不存在") customer.is_active = False await db_session.commit() return {"message": "客户已删除"} @staticmethod async def list_suppliers( db_session: AsyncSession, skip: int, limit: int, search: Optional[str], ) -> List[SupplierResponse]: query: Select = select(Supplier).where(Supplier.is_active == True) if search: query = query.where(Supplier.name.ilike(f"%{search}%")) query = query.offset(skip).limit(limit).order_by(Supplier.created_at.desc()) result = await db_session.execute(query) return [_build_supplier_response(supplier) for supplier in result.scalars().all()] @staticmethod async def create_supplier( db_session: AsyncSession, supplier_data: SupplierCreate, current_user: User, ) -> SupplierResponse: data = supplier_data.model_dump() if not data.get("code"): data["code"] = _generate_code("S") supplier = Supplier(**data) db_session.add(supplier) await db_session.commit() await db_session.refresh(supplier) return _build_supplier_response(supplier) @staticmethod async def update_supplier( db_session: AsyncSession, supplier_id: int, supplier_data: SupplierCreate, current_user: User, ) -> SupplierResponse: supplier = await _get_entity_or_404(db_session, Supplier, supplier_id, "供应商不存在") for key, value in supplier_data.model_dump().items(): setattr(supplier, key, value) await db_session.commit() await db_session.refresh(supplier) return _build_supplier_response(supplier) @staticmethod async def delete_supplier( db_session: AsyncSession, supplier_id: int, current_user: User, ) -> dict: supplier = await _get_entity_or_404(db_session, Supplier, supplier_id, "供应商不存在") supplier.is_active = False await db_session.commit() return {"message": "供应商已删除"} @staticmethod async def list_warehouses( db_session: AsyncSession, ) -> List[WarehouseResponse]: result = await db_session.execute( select(Warehouse).where(Warehouse.is_active == True).order_by(Warehouse.is_default.desc()) ) return [_build_warehouse_response(warehouse) for warehouse in result.scalars().all()] @staticmethod async def create_warehouse( db_session: AsyncSession, warehouse_data: WarehouseCreate, current_user: User, ) -> WarehouseResponse: data = warehouse_data.model_dump() if not data.get("code"): data["code"] = _generate_code("W") warehouse = Warehouse(**data) db_session.add(warehouse) await db_session.commit() await db_session.refresh(warehouse) return _build_warehouse_response(warehouse) master_data_service = MasterDataService()