From f3eda856b060630603f8c30c895405154e42d09e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Sun, 26 Jul 2026 10:41:14 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E4=B8=9A=E5=8A=A1=E5=9F=BA=E7=A1=80=E8=B5=84=E6=96=99=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/customer_controller.py | 123 +++ .../controller/line_controller.py | 121 +++ .../controller/site_controller.py | 123 +++ .../controller/strain_controller.py | 123 +++ .../controller/warehouse_controller.py | 125 +++ .../module_business/dao/master_dao.py | 378 +++++++++ .../module_business/service/master_service.py | 375 +++++++++ .../module_business/test_master_service.py | 746 ++++++++++++++++++ 8 files changed, 2114 insertions(+) create mode 100644 shierfeng-fastapi-backend/module_business/controller/customer_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/controller/line_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/controller/site_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/controller/strain_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py create mode 100644 shierfeng-fastapi-backend/module_business/dao/master_dao.py create mode 100644 shierfeng-fastapi-backend/module_business/service/master_service.py create mode 100644 shierfeng-fastapi-backend/tests/module_business/test_master_service.py diff --git a/shierfeng-fastapi-backend/module_business/controller/customer_controller.py b/shierfeng-fastapi-backend/module_business/controller/customer_controller.py new file mode 100644 index 0000000..1ff3d3f --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/customer_controller.py @@ -0,0 +1,123 @@ +from datetime import datetime +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.master_vo import CustomerModel, CustomerPageQueryModel, DeleteCustomerModel +from module_business.service.master_service import MasterService +from utils.response_util import ResponseUtil + +customer_controller = APIRouterPro( + prefix='/master/customer', tags=['基础资料-客户管理'], dependencies=[PreAuthDependency()] +) + + +@customer_controller.get( + '/list', + summary='获取客户分页列表接口', + description='用于获取客户分页列表', + response_model=PageResponseModel[CustomerModel], + dependencies=[UserInterfaceAuthDependency('master:customer:list')], +) +async def get_master_customer_list( + request: Request, + customer_page_query: Annotated[CustomerPageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + customer_page_query_result = await MasterService.list_customer(query_db, customer_page_query, is_page=True) + + return ResponseUtil.success(model_content=customer_page_query_result) + + +@customer_controller.get( + '/{customer_id}', + summary='获取客户详情接口', + description='用于获取指定客户的详细信息', + response_model=DataResponseModel[CustomerModel], + dependencies=[UserInterfaceAuthDependency('master:customer:query')], +) +async def query_detail_master_customer( + request: Request, + customer_id: Annotated[int, Path(gt=0, description='客户ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + customer_detail_result = await MasterService.detail_customer(query_db, customer_id) + + return ResponseUtil.success(data=customer_detail_result) + + +@customer_controller.post( + '', + summary='新增客户接口', + description='用于新增客户', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:customer:add')], +) +@Log(title='客户管理', business_type=BusinessType.INSERT) +async def add_master_customer( + request: Request, + add_customer: CustomerModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + current_time = datetime.now() + add_customer.create_by = current_user.user.user_name + add_customer.create_time = current_time + add_customer.update_by = current_user.user.user_name + add_customer.update_time = current_time + add_customer_result = await MasterService.add_customer(query_db, add_customer) + + return ResponseUtil.success(msg=add_customer_result.message) + + +@customer_controller.put( + '', + summary='编辑客户接口', + description='用于编辑客户', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:customer:edit')], +) +@Log(title='客户管理', business_type=BusinessType.UPDATE) +async def edit_master_customer( + request: Request, + edit_customer: CustomerModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + edit_customer.update_by = current_user.user.user_name + edit_customer.update_time = datetime.now() + edit_customer_result = await MasterService.edit_customer(query_db, edit_customer) + + return ResponseUtil.success(msg=edit_customer_result.message) + + +@customer_controller.delete( + '/{customer_ids}', + summary='删除客户接口', + description='用于删除客户', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:customer:remove')], +) +@Log(title='客户管理', business_type=BusinessType.DELETE) +async def delete_master_customer( + request: Request, + customer_ids: Annotated[ + str, + Path(pattern=r'^[1-9]\d*(,[1-9]\d*)*$', description='需要删除的客户ID'), + ], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + delete_customer = DeleteCustomerModel(customerIds=[int(customer_id) for customer_id in customer_ids.split(',')]) + delete_customer_result = await MasterService.remove_customer(query_db, delete_customer) + + return ResponseUtil.success(msg=delete_customer_result.message) diff --git a/shierfeng-fastapi-backend/module_business/controller/line_controller.py b/shierfeng-fastapi-backend/module_business/controller/line_controller.py new file mode 100644 index 0000000..04cf487 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/line_controller.py @@ -0,0 +1,121 @@ +from datetime import datetime +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.master_vo import DeleteLineModel, LineModel, LinePageQueryModel +from module_business.service.master_service import MasterService +from utils.response_util import ResponseUtil + +line_controller = APIRouterPro(prefix='/master/line', tags=['基础资料-生产线管理'], dependencies=[PreAuthDependency()]) + + +@line_controller.get( + '/list', + summary='获取生产线分页列表接口', + description='用于获取生产线分页列表', + response_model=PageResponseModel[LineModel], + dependencies=[UserInterfaceAuthDependency('master:line:list')], +) +async def get_master_line_list( + request: Request, + line_page_query: Annotated[LinePageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + line_page_query_result = await MasterService.list_line(query_db, line_page_query, is_page=True) + + return ResponseUtil.success(model_content=line_page_query_result) + + +@line_controller.get( + '/{line_id}', + summary='获取生产线详情接口', + description='用于获取指定生产线的详细信息', + response_model=DataResponseModel[LineModel], + dependencies=[UserInterfaceAuthDependency('master:line:query')], +) +async def query_detail_master_line( + request: Request, + line_id: Annotated[int, Path(gt=0, description='生产线ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + line_detail_result = await MasterService.detail_line(query_db, line_id) + + return ResponseUtil.success(data=line_detail_result) + + +@line_controller.post( + '', + summary='新增生产线接口', + description='用于新增生产线', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:line:add')], +) +@Log(title='生产线管理', business_type=BusinessType.INSERT) +async def add_master_line( + request: Request, + add_line: LineModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + current_time = datetime.now() + add_line.create_by = current_user.user.user_name + add_line.create_time = current_time + add_line.update_by = current_user.user.user_name + add_line.update_time = current_time + add_line_result = await MasterService.add_line(query_db, add_line) + + return ResponseUtil.success(msg=add_line_result.message) + + +@line_controller.put( + '', + summary='编辑生产线接口', + description='用于编辑生产线', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:line:edit')], +) +@Log(title='生产线管理', business_type=BusinessType.UPDATE) +async def edit_master_line( + request: Request, + edit_line: LineModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + edit_line.update_by = current_user.user.user_name + edit_line.update_time = datetime.now() + edit_line_result = await MasterService.edit_line(query_db, edit_line) + + return ResponseUtil.success(msg=edit_line_result.message) + + +@line_controller.delete( + '/{line_ids}', + summary='删除生产线接口', + description='用于删除生产线', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:line:remove')], +) +@Log(title='生产线管理', business_type=BusinessType.DELETE) +async def delete_master_line( + request: Request, + line_ids: Annotated[ + str, + Path(pattern=r'^[1-9]\d*(,[1-9]\d*)*$', description='需要删除的生产线ID'), + ], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + delete_line = DeleteLineModel(lineIds=[int(line_id) for line_id in line_ids.split(',')]) + delete_line_result = await MasterService.remove_line(query_db, delete_line) + + return ResponseUtil.success(msg=delete_line_result.message) diff --git a/shierfeng-fastapi-backend/module_business/controller/site_controller.py b/shierfeng-fastapi-backend/module_business/controller/site_controller.py new file mode 100644 index 0000000..c27d975 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/site_controller.py @@ -0,0 +1,123 @@ +from datetime import datetime +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.master_vo import DeleteSiteModel, SiteModel, SitePageQueryModel +from module_business.service.master_service import MasterService +from utils.response_util import ResponseUtil + +site_controller = APIRouterPro( + prefix='/master/site', tags=['基础资料-客户现场管理'], dependencies=[PreAuthDependency()] +) + + +@site_controller.get( + '/list', + summary='获取客户现场分页列表接口', + description='用于获取客户现场分页列表', + response_model=PageResponseModel[SiteModel], + dependencies=[UserInterfaceAuthDependency('master:site:list')], +) +async def get_master_site_list( + request: Request, + site_page_query: Annotated[SitePageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + site_page_query_result = await MasterService.list_site(query_db, site_page_query, is_page=True) + + return ResponseUtil.success(model_content=site_page_query_result) + + +@site_controller.get( + '/{site_id}', + summary='获取客户现场详情接口', + description='用于获取指定客户现场的详细信息', + response_model=DataResponseModel[SiteModel], + dependencies=[UserInterfaceAuthDependency('master:site:query')], +) +async def query_detail_master_site( + request: Request, + site_id: Annotated[int, Path(gt=0, description='客户现场ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + site_detail_result = await MasterService.detail_site(query_db, site_id) + + return ResponseUtil.success(data=site_detail_result) + + +@site_controller.post( + '', + summary='新增客户现场接口', + description='用于新增客户现场', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:site:add')], +) +@Log(title='客户现场管理', business_type=BusinessType.INSERT) +async def add_master_site( + request: Request, + add_site: SiteModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + current_time = datetime.now() + add_site.create_by = current_user.user.user_name + add_site.create_time = current_time + add_site.update_by = current_user.user.user_name + add_site.update_time = current_time + add_site_result = await MasterService.add_site(query_db, add_site) + + return ResponseUtil.success(msg=add_site_result.message) + + +@site_controller.put( + '', + summary='编辑客户现场接口', + description='用于编辑客户现场', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:site:edit')], +) +@Log(title='客户现场管理', business_type=BusinessType.UPDATE) +async def edit_master_site( + request: Request, + edit_site: SiteModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + edit_site.update_by = current_user.user.user_name + edit_site.update_time = datetime.now() + edit_site_result = await MasterService.edit_site(query_db, edit_site) + + return ResponseUtil.success(msg=edit_site_result.message) + + +@site_controller.delete( + '/{site_ids}', + summary='删除客户现场接口', + description='用于删除客户现场', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:site:remove')], +) +@Log(title='客户现场管理', business_type=BusinessType.DELETE) +async def delete_master_site( + request: Request, + site_ids: Annotated[ + str, + Path(pattern=r'^[1-9]\d*(,[1-9]\d*)*$', description='需要删除的客户现场ID'), + ], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + delete_site = DeleteSiteModel(siteIds=[int(site_id) for site_id in site_ids.split(',')]) + delete_site_result = await MasterService.remove_site(query_db, delete_site) + + return ResponseUtil.success(msg=delete_site_result.message) diff --git a/shierfeng-fastapi-backend/module_business/controller/strain_controller.py b/shierfeng-fastapi-backend/module_business/controller/strain_controller.py new file mode 100644 index 0000000..45183ea --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/strain_controller.py @@ -0,0 +1,123 @@ +from datetime import datetime +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.master_vo import DeleteStrainModel, StrainModel, StrainPageQueryModel +from module_business.service.master_service import MasterService +from utils.response_util import ResponseUtil + +strain_controller = APIRouterPro( + prefix='/master/strain', tags=['基础资料-菌种管理'], dependencies=[PreAuthDependency()] +) + + +@strain_controller.get( + '/list', + summary='获取菌种分页列表接口', + description='用于获取菌种分页列表', + response_model=PageResponseModel[StrainModel], + dependencies=[UserInterfaceAuthDependency('master:strain:list')], +) +async def get_master_strain_list( + request: Request, + strain_page_query: Annotated[StrainPageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + strain_page_query_result = await MasterService.list_strain(query_db, strain_page_query, is_page=True) + + return ResponseUtil.success(model_content=strain_page_query_result) + + +@strain_controller.get( + '/{strain_id}', + summary='获取菌种详情接口', + description='用于获取指定菌种的详细信息', + response_model=DataResponseModel[StrainModel], + dependencies=[UserInterfaceAuthDependency('master:strain:query')], +) +async def query_detail_master_strain( + request: Request, + strain_id: Annotated[int, Path(gt=0, description='菌种ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + strain_detail_result = await MasterService.detail_strain(query_db, strain_id) + + return ResponseUtil.success(data=strain_detail_result) + + +@strain_controller.post( + '', + summary='新增菌种接口', + description='用于新增菌种', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:strain:add')], +) +@Log(title='菌种管理', business_type=BusinessType.INSERT) +async def add_master_strain( + request: Request, + add_strain: StrainModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + current_time = datetime.now() + add_strain.create_by = current_user.user.user_name + add_strain.create_time = current_time + add_strain.update_by = current_user.user.user_name + add_strain.update_time = current_time + add_strain_result = await MasterService.add_strain(query_db, add_strain) + + return ResponseUtil.success(msg=add_strain_result.message) + + +@strain_controller.put( + '', + summary='编辑菌种接口', + description='用于编辑菌种', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:strain:edit')], +) +@Log(title='菌种管理', business_type=BusinessType.UPDATE) +async def edit_master_strain( + request: Request, + edit_strain: StrainModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + edit_strain.update_by = current_user.user.user_name + edit_strain.update_time = datetime.now() + edit_strain_result = await MasterService.edit_strain(query_db, edit_strain) + + return ResponseUtil.success(msg=edit_strain_result.message) + + +@strain_controller.delete( + '/{strain_ids}', + summary='删除菌种接口', + description='用于删除菌种', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:strain:remove')], +) +@Log(title='菌种管理', business_type=BusinessType.DELETE) +async def delete_master_strain( + request: Request, + strain_ids: Annotated[ + str, + Path(pattern=r'^[1-9]\d*(,[1-9]\d*)*$', description='需要删除的菌种ID'), + ], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + delete_strain = DeleteStrainModel(strainIds=[int(strain_id) for strain_id in strain_ids.split(',')]) + delete_strain_result = await MasterService.remove_strain(query_db, delete_strain) + + return ResponseUtil.success(msg=delete_strain_result.message) diff --git a/shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py b/shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py new file mode 100644 index 0000000..ab12ae5 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/controller/warehouse_controller.py @@ -0,0 +1,125 @@ +from datetime import datetime +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.master_vo import DeleteWarehouseModel, WarehouseModel, WarehousePageQueryModel +from module_business.service.master_service import MasterService +from utils.response_util import ResponseUtil + +warehouse_controller = APIRouterPro( + prefix='/master/warehouse', tags=['基础资料-仓库管理'], dependencies=[PreAuthDependency()] +) + + +@warehouse_controller.get( + '/list', + summary='获取仓库分页列表接口', + description='用于获取仓库分页列表', + response_model=PageResponseModel[WarehouseModel], + dependencies=[UserInterfaceAuthDependency('master:warehouse:list')], +) +async def get_master_warehouse_list( + request: Request, + warehouse_page_query: Annotated[WarehousePageQueryModel, Query()], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + warehouse_page_query_result = await MasterService.list_warehouse(query_db, warehouse_page_query, is_page=True) + + return ResponseUtil.success(model_content=warehouse_page_query_result) + + +@warehouse_controller.get( + '/{warehouse_id}', + summary='获取仓库详情接口', + description='用于获取指定仓库的详细信息', + response_model=DataResponseModel[WarehouseModel], + dependencies=[UserInterfaceAuthDependency('master:warehouse:query')], +) +async def query_detail_master_warehouse( + request: Request, + warehouse_id: Annotated[int, Path(gt=0, description='仓库ID')], + query_db: Annotated[AsyncSession, DBSessionDependency()], +) -> Response: + warehouse_detail_result = await MasterService.detail_warehouse(query_db, warehouse_id) + + return ResponseUtil.success(data=warehouse_detail_result) + + +@warehouse_controller.post( + '', + summary='新增仓库接口', + description='用于新增仓库', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:warehouse:add')], +) +@Log(title='仓库管理', business_type=BusinessType.INSERT) +async def add_master_warehouse( + request: Request, + add_warehouse: WarehouseModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + current_time = datetime.now() + add_warehouse.create_by = current_user.user.user_name + add_warehouse.create_time = current_time + add_warehouse.update_by = current_user.user.user_name + add_warehouse.update_time = current_time + add_warehouse_result = await MasterService.add_warehouse(query_db, add_warehouse) + + return ResponseUtil.success(msg=add_warehouse_result.message) + + +@warehouse_controller.put( + '', + summary='编辑仓库接口', + description='用于编辑仓库', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:warehouse:edit')], +) +@Log(title='仓库管理', business_type=BusinessType.UPDATE) +async def edit_master_warehouse( + request: Request, + edit_warehouse: WarehouseModel, + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + edit_warehouse.update_by = current_user.user.user_name + edit_warehouse.update_time = datetime.now() + edit_warehouse_result = await MasterService.edit_warehouse(query_db, edit_warehouse) + + return ResponseUtil.success(msg=edit_warehouse_result.message) + + +@warehouse_controller.delete( + '/{warehouse_ids}', + summary='删除仓库接口', + description='用于删除仓库', + response_model=ResponseBaseModel, + dependencies=[UserInterfaceAuthDependency('master:warehouse:remove')], +) +@Log(title='仓库管理', business_type=BusinessType.DELETE) +async def delete_master_warehouse( + request: Request, + warehouse_ids: Annotated[ + str, + Path(pattern=r'^[1-9]\d*(,[1-9]\d*)*$', description='需要删除的仓库ID'), + ], + query_db: Annotated[AsyncSession, DBSessionDependency()], + current_user: Annotated[CurrentUserModel, CurrentUserDependency()], +) -> Response: + delete_warehouse = DeleteWarehouseModel( + warehouseIds=[int(warehouse_id) for warehouse_id in warehouse_ids.split(',')] + ) + delete_warehouse_result = await MasterService.remove_warehouse(query_db, delete_warehouse) + + return ResponseUtil.success(msg=delete_warehouse_result.message) diff --git a/shierfeng-fastapi-backend/module_business/dao/master_dao.py b/shierfeng-fastapi-backend/module_business/dao/master_dao.py new file mode 100644 index 0000000..b21d110 --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/dao/master_dao.py @@ -0,0 +1,378 @@ +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 diff --git a/shierfeng-fastapi-backend/module_business/service/master_service.py b/shierfeng-fastapi-backend/module_business/service/master_service.py new file mode 100644 index 0000000..26e067e --- /dev/null +++ b/shierfeng-fastapi-backend/module_business/service/master_service.py @@ -0,0 +1,375 @@ +from typing import Any, Literal + +from pydantic import BaseModel +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from common.vo import CrudResponseModel, PageModel +from exceptions.exception import ServiceException, ServiceWarning +from module_business.dao.master_dao import MasterDao +from module_business.entity.vo.master_vo import ( + CustomerModel, + CustomerPageQueryModel, + DeleteCustomerModel, + DeleteLineModel, + DeleteSiteModel, + DeleteStrainModel, + DeleteWarehouseModel, + LineModel, + LinePageQueryModel, + SiteModel, + SitePageQueryModel, + StrainModel, + StrainPageQueryModel, + WarehouseModel, + WarehousePageQueryModel, +) + +MasterResource = Literal['strain', 'customer', 'site', 'line', 'warehouse'] + + +_RESOURCE_CONFIG: dict[MasterResource, dict[str, Any]] = { + 'strain': { + 'label': '菌种', + 'id_field': 'strain_id', + 'code_field': 'strain_code', + 'ids_field': 'strain_ids', + 'vo': StrainModel, + }, + 'customer': { + 'label': '客户', + 'id_field': 'customer_id', + 'code_field': 'customer_code', + 'ids_field': 'customer_ids', + 'vo': CustomerModel, + }, + 'site': { + 'label': '客户现场', + 'id_field': 'site_id', + 'code_field': 'site_code', + 'ids_field': 'site_ids', + 'vo': SiteModel, + }, + 'line': { + 'label': '生产线', + 'id_field': 'line_id', + 'code_field': 'line_code', + 'ids_field': 'line_ids', + 'vo': LineModel, + }, + 'warehouse': { + 'label': '仓库', + 'id_field': 'warehouse_id', + 'code_field': 'warehouse_code', + 'ids_field': 'warehouse_ids', + 'vo': WarehouseModel, + }, +} + + +class MasterService: + """菌种、客户、现场、产线和仓库主数据服务层。""" + + @staticmethod + async def _validate_parent(db: AsyncSession, resource: MasterResource, page_object: BaseModel) -> None: + if resource == 'site': + customer = await MasterDao.get_customer(db, page_object.customer_id) + if customer is None: + raise ServiceException(message='目标客户不存在') + if customer.status != '0': + raise ServiceException(message='目标客户已停用') + elif resource == 'line': + site = await MasterDao.get_site(db, page_object.site_id) + if site is None: + raise ServiceException(message='目标客户现场不存在') + if site.status != '0': + raise ServiceException(message='目标客户现场已停用') + + @staticmethod + async def _find_duplicate( + db: AsyncSession, + resource: MasterResource, + page_object: BaseModel, + ) -> Any | None: + if resource == 'site': + return await MasterDao.find_site_by_code( + db, + page_object.site_code, + page_object.customer_id, + ) + if resource == 'line': + return await MasterDao.find_line_by_code( + db, + page_object.line_code, + page_object.site_id, + ) + code_field = _RESOURCE_CONFIG[resource]['code_field'] + return await getattr(MasterDao, f'find_{resource}_by_code')( + db, + getattr(page_object, code_field), + ) + + @classmethod + async def _validate_unique( + cls, + db: AsyncSession, + resource: MasterResource, + page_object: BaseModel, + current_id: int | None = None, + ) -> None: + duplicate = await cls._find_duplicate(db, resource, page_object) + if duplicate is None: + return + + config = _RESOURCE_CONFIG[resource] + duplicate_id = getattr(duplicate, config['id_field'], None) + if current_id is None or duplicate_id != current_id: + code = getattr(page_object, config['code_field']) + raise ServiceException(message=f'{config["label"]}编码{code}已存在') + + @staticmethod + def _extract_ids(resource: MasterResource, value: int | BaseModel) -> list[int]: + ids = [value] if isinstance(value, int) else list(getattr(value, _RESOURCE_CONFIG[resource]['ids_field'])) + if not ids or any(resource_id <= 0 for resource_id in ids): + raise ServiceException(message='主数据ID必须为正整数') + if len(ids) != len(set(ids)): + raise ServiceException(message='主数据ID不能重复') + return ids + + @classmethod + async def _list( + cls, + db: AsyncSession, + resource: MasterResource, + query_object: BaseModel, + is_page: bool, + ) -> PageModel | list[dict[str, Any]]: + return await getattr(MasterDao, f'list_{resource}')(db, query_object, is_page) + + @classmethod + async def _detail(cls, db: AsyncSession, resource: MasterResource, resource_id: int) -> BaseModel: + config = _RESOURCE_CONFIG[resource] + record = await getattr(MasterDao, f'get_{resource}')(db, resource_id) + if record is None: + raise ServiceException(message=f'{config["label"]}不存在') + return config['vo'].model_validate(record) + + @classmethod + async def _add( + cls, + db: AsyncSession, + resource: MasterResource, + page_object: BaseModel, + ) -> CrudResponseModel: + config = _RESOURCE_CONFIG[resource] + try: + await cls._validate_parent(db, resource, page_object) + await cls._validate_unique(db, resource, page_object) + safe_object = page_object.model_copy(update={config['id_field']: None}) + await getattr(MasterDao, f'add_{resource}')(db, safe_object) + await db.commit() + return CrudResponseModel(is_success=True, message='新增成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message=f'新增{config["label"]}失败，编码已存在或关联数据无效') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def _edit( + cls, + db: AsyncSession, + resource: MasterResource, + page_object: BaseModel, + ) -> CrudResponseModel: + config = _RESOURCE_CONFIG[resource] + try: + resource_id = getattr(page_object, config['id_field']) + if resource_id is None: + raise ServiceException(message=f'{config["label"]}ID不能为空') + current = await getattr(MasterDao, f'get_{resource}')(db, resource_id) + if current is None: + raise ServiceException(message=f'{config["label"]}不存在') + await cls._validate_parent(db, resource, page_object) + await cls._validate_unique(db, resource, page_object, resource_id) + await getattr(MasterDao, f'update_{resource}')(db, page_object) + await db.commit() + return CrudResponseModel(is_success=True, message='更新成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message=f'修改{config["label"]}失败，编码已存在或关联数据无效') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def _remove( + cls, + db: AsyncSession, + resource: MasterResource, + value: int | BaseModel, + ) -> CrudResponseModel: + config = _RESOURCE_CONFIG[resource] + try: + resource_ids = cls._extract_ids(resource, value) + for resource_id in resource_ids: + if await MasterDao.count_references(db, resource, resource_id) > 0: + raise ServiceWarning(message=f'{config["label"]}已被引用，只能停用，不能删除') + if await getattr(MasterDao, f'get_{resource}')(db, resource_id) is None: + raise ServiceException(message=f'{config["label"]}不存在') + for resource_id in resource_ids: + await getattr(MasterDao, f'delete_{resource}')(db, resource_id) + await db.commit() + return CrudResponseModel(is_success=True, message='删除成功') + except IntegrityError: + await db.rollback() + raise ServiceException(message=f'删除{config["label"]}失败，该数据仍被引用') from None + except Exception: + await db.rollback() + raise + + @classmethod + async def list_strain( + cls, + db: AsyncSession, + query_object: StrainPageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + return await cls._list(db, 'strain', query_object, is_page) + + @classmethod + async def detail_strain(cls, db: AsyncSession, strain_id: int) -> StrainModel: + return await cls._detail(db, 'strain', strain_id) + + @classmethod + async def add_strain(cls, db: AsyncSession, strain: StrainModel) -> CrudResponseModel: + return await cls._add(db, 'strain', strain) + + @classmethod + async def edit_strain(cls, db: AsyncSession, strain: StrainModel) -> CrudResponseModel: + return await cls._edit(db, 'strain', strain) + + @classmethod + async def remove_strain( + cls, + db: AsyncSession, + value: int | DeleteStrainModel, + ) -> CrudResponseModel: + return await cls._remove(db, 'strain', value) + + @classmethod + async def list_customer( + cls, + db: AsyncSession, + query_object: CustomerPageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + return await cls._list(db, 'customer', query_object, is_page) + + @classmethod + async def detail_customer(cls, db: AsyncSession, customer_id: int) -> CustomerModel: + return await cls._detail(db, 'customer', customer_id) + + @classmethod + async def add_customer(cls, db: AsyncSession, customer: CustomerModel) -> CrudResponseModel: + return await cls._add(db, 'customer', customer) + + @classmethod + async def edit_customer(cls, db: AsyncSession, customer: CustomerModel) -> CrudResponseModel: + return await cls._edit(db, 'customer', customer) + + @classmethod + async def remove_customer( + cls, + db: AsyncSession, + value: int | DeleteCustomerModel, + ) -> CrudResponseModel: + return await cls._remove(db, 'customer', value) + + @classmethod + async def list_site( + cls, + db: AsyncSession, + query_object: SitePageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + return await cls._list(db, 'site', query_object, is_page) + + @classmethod + async def detail_site(cls, db: AsyncSession, site_id: int) -> SiteModel: + return await cls._detail(db, 'site', site_id) + + @classmethod + async def add_site(cls, db: AsyncSession, site: SiteModel) -> CrudResponseModel: + return await cls._add(db, 'site', site) + + @classmethod + async def edit_site(cls, db: AsyncSession, site: SiteModel) -> CrudResponseModel: + return await cls._edit(db, 'site', site) + + @classmethod + async def remove_site( + cls, + db: AsyncSession, + value: int | DeleteSiteModel, + ) -> CrudResponseModel: + return await cls._remove(db, 'site', value) + + @classmethod + async def list_line( + cls, + db: AsyncSession, + query_object: LinePageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + return await cls._list(db, 'line', query_object, is_page) + + @classmethod + async def detail_line(cls, db: AsyncSession, line_id: int) -> LineModel: + return await cls._detail(db, 'line', line_id) + + @classmethod + async def add_line(cls, db: AsyncSession, line: LineModel) -> CrudResponseModel: + return await cls._add(db, 'line', line) + + @classmethod + async def edit_line(cls, db: AsyncSession, line: LineModel) -> CrudResponseModel: + return await cls._edit(db, 'line', line) + + @classmethod + async def remove_line( + cls, + db: AsyncSession, + value: int | DeleteLineModel, + ) -> CrudResponseModel: + return await cls._remove(db, 'line', value) + + @classmethod + async def list_warehouse( + cls, + db: AsyncSession, + query_object: WarehousePageQueryModel, + is_page: bool = True, + ) -> PageModel | list[dict[str, Any]]: + return await cls._list(db, 'warehouse', query_object, is_page) + + @classmethod + async def detail_warehouse(cls, db: AsyncSession, warehouse_id: int) -> WarehouseModel: + return await cls._detail(db, 'warehouse', warehouse_id) + + @classmethod + async def add_warehouse(cls, db: AsyncSession, warehouse: WarehouseModel) -> CrudResponseModel: + return await cls._add(db, 'warehouse', warehouse) + + @classmethod + async def edit_warehouse(cls, db: AsyncSession, warehouse: WarehouseModel) -> CrudResponseModel: + return await cls._edit(db, 'warehouse', warehouse) + + @classmethod + async def remove_warehouse( + cls, + db: AsyncSession, + value: int | DeleteWarehouseModel, + ) -> CrudResponseModel: + return await cls._remove(db, 'warehouse', value) diff --git a/shierfeng-fastapi-backend/tests/module_business/test_master_service.py b/shierfeng-fastapi-backend/tests/module_business/test_master_service.py new file mode 100644 index 0000000..781cfce --- /dev/null +++ b/shierfeng-fastapi-backend/tests/module_business/test_master_service.py @@ -0,0 +1,746 @@ +import inspect +import unittest +from contextlib import ExitStack +from datetime import datetime +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +from fastapi import FastAPI +from fastapi.routing import APIRoute +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from common.aspect.interface_auth import CheckUserInterfaceAuth +from common.aspect.pre_auth import PreAuth +from common.vo import CrudResponseModel, DataResponseModel, PageModel, PageResponseModel +from exceptions.exception import ServiceException, ServiceWarning +from module_admin.service.login_service import LoginService +from module_business.controller import ( + customer_controller as customer_controller_module, +) +from module_business.controller import line_controller as line_controller_module +from module_business.controller import site_controller as site_controller_module +from module_business.controller import strain_controller as strain_controller_module +from module_business.controller import warehouse_controller as warehouse_controller_module +from module_business.dao.master_dao import MasterDao +from module_business.entity.do.master_do import ( + BizCustomer, + BizCustomerSite, + BizProductionLine, + BizStrain, + BizWarehouse, +) +from module_business.entity.vo.master_vo import ( + CustomerModel, + CustomerPageQueryModel, + DeleteCustomerModel, + DeleteLineModel, + DeleteSiteModel, + DeleteStrainModel, + DeleteWarehouseModel, + LineModel, + LinePageQueryModel, + SiteModel, + SitePageQueryModel, + StrainModel, + StrainPageQueryModel, + WarehouseModel, + WarehousePageQueryModel, +) +from module_business.service.master_service import MasterService +from utils.page_util import PageUtil + +RESOURCE_IDS = { + 'strain': 'strain_id', + 'customer': 'customer_id', + 'site': 'site_id', + 'line': 'line_id', + 'warehouse': 'warehouse_id', +} + + +def make_models(resource_id: int | None = None) -> dict[str, object]: + return { + 'strain': StrainModel(strainId=resource_id, strainCode='S001', strainName='测试菌种'), + 'customer': CustomerModel(customerId=resource_id, customerCode='C001', customerName='测试客户'), + 'site': SiteModel( + siteId=resource_id, + customerId=11, + siteCode='SITE001', + siteName='测试现场', + ), + 'line': LineModel( + lineId=resource_id, + siteId=22, + lineCode='L001', + lineName='测试产线', + ), + 'warehouse': WarehouseModel( + warehouseId=resource_id, + warehouseCode='W001', + warehouseName='测试仓库', + ), + } + + +def compile_sql(statement: Any) -> str: + return ' '.join(str(statement.compile(compile_kwargs={'literal_binds': True})).split()) + + +def integrity_error() -> IntegrityError: + return IntegrityError('statement', {}, RuntimeError('duplicate')) + + +class TestMasterService(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + self.strain = StrainModel(strainCode='S001', strainName='测试菌种') + self.line = LineModel(siteId=1, lineCode='L001', lineName='测试产线') + + def _patch_enabled_parent(self, stack: ExitStack, resource: str) -> None: + if resource == 'site': + stack.enter_context( + patch.object(MasterDao, 'get_customer', new=AsyncMock(return_value=SimpleNamespace(status='0'))) + ) + elif resource == 'line': + stack.enter_context( + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))) + ) + + async def test_duplicate_strain_code_is_rejected(self) -> None: + with ( + patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=object())), + self.assertRaises(ServiceException), + ): + await MasterService.add_strain(self.db, self.strain) + + async def test_line_must_belong_to_enabled_site(self) -> None: + with ( + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=None)), + self.assertRaises(ServiceException), + ): + await MasterService.add_line(self.db, self.line) + + async def test_referenced_master_can_only_be_disabled(self) -> None: + with ( + patch.object(MasterDao, 'count_references', new=AsyncMock(return_value=1)), + self.assertRaises(ServiceWarning), + ): + await MasterService.remove_strain(self.db, 1) + + async def test_duplicate_code_is_rejected_for_all_five_resources(self) -> None: + for resource, model in make_models().items(): + with self.subTest(resource=resource), ExitStack() as stack: + self._patch_enabled_parent(stack, resource) + stack.enter_context( + patch.object( + MasterDao, + f'find_{resource}_by_code', + new=AsyncMock(return_value=SimpleNamespace()), + ) + ) + with self.assertRaises(ServiceException): + await getattr(MasterService, f'add_{resource}')(self.db, model) + + async def test_edit_unique_check_excludes_current_record_for_all_resources(self) -> None: + for resource, model in make_models(7).items(): + id_field = RESOURCE_IDS[resource] + current = SimpleNamespace(**{id_field: 7}) + with self.subTest(resource=resource), ExitStack() as stack: + self.db.reset_mock() + self._patch_enabled_parent(stack, resource) + stack.enter_context(patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=current))) + stack.enter_context( + patch.object(MasterDao, f'find_{resource}_by_code', new=AsyncMock(return_value=current)) + ) + update_mock = stack.enter_context(patch.object(MasterDao, f'update_{resource}', new=AsyncMock())) + + result = await getattr(MasterService, f'edit_{resource}')(self.db, model) + + self.assertIsInstance(result, CrudResponseModel) + self.assertTrue(result.is_success) + update_mock.assert_awaited_once() + self.db.commit.assert_awaited_once() + + async def test_edit_duplicate_code_from_another_record_is_rejected_for_all_resources(self) -> None: + for resource, model in make_models(7).items(): + id_field = RESOURCE_IDS[resource] + with self.subTest(resource=resource), ExitStack() as stack: + self._patch_enabled_parent(stack, resource) + stack.enter_context( + patch.object( + MasterDao, + f'get_{resource}', + new=AsyncMock(return_value=SimpleNamespace(**{id_field: 7})), + ) + ) + stack.enter_context( + patch.object( + MasterDao, + f'find_{resource}_by_code', + new=AsyncMock(return_value=SimpleNamespace(**{id_field: 8})), + ) + ) + + with self.assertRaises(ServiceException): + await getattr(MasterService, f'edit_{resource}')(self.db, model) + + async def test_site_add_requires_existing_enabled_customer(self) -> None: + site = make_models()['site'] + for parent in (None, SimpleNamespace(status='1')): + with ( + self.subTest(parent=parent), + patch.object(MasterDao, 'get_customer', new=AsyncMock(return_value=parent)), + self.assertRaises(ServiceException), + ): + await MasterService.add_site(self.db, site) + + with ( + patch.object( + MasterDao, + 'get_customer', + new=AsyncMock(return_value=SimpleNamespace(status='0')), + ), + patch.object(MasterDao, 'find_site_by_code', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'add_site', new=AsyncMock()), + ): + result = await MasterService.add_site(self.db, site) + self.assertTrue(result.is_success) + + async def test_line_add_requires_existing_enabled_site(self) -> None: + line = make_models()['line'] + for parent in (None, SimpleNamespace(status='1')): + with ( + self.subTest(parent=parent), + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=parent)), + self.assertRaises(ServiceException), + ): + await MasterService.add_line(self.db, line) + + with ( + patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))), + patch.object(MasterDao, 'find_line_by_code', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'add_line', new=AsyncMock()), + ): + result = await MasterService.add_line(self.db, line) + self.assertTrue(result.is_success) + + async def test_site_and_line_edit_require_enabled_target_parent(self) -> None: + cases = ( + ('site', make_models(7)['site'], 'get_customer'), + ('line', make_models(7)['line'], 'get_site'), + ) + for resource, model, parent_method in cases: + for parent in (None, SimpleNamespace(status='1')): + with self.subTest(resource=resource, parent=parent), ExitStack() as stack: + stack.enter_context( + patch.object( + MasterDao, + f'get_{resource}', + new=AsyncMock(return_value=SimpleNamespace(**{RESOURCE_IDS[resource]: 7})), + ) + ) + stack.enter_context(patch.object(MasterDao, parent_method, new=AsyncMock(return_value=parent))) + with self.assertRaises(ServiceException): + await getattr(MasterService, f'edit_{resource}')(self.db, model) + + with self.subTest(resource=resource, parent='enabled'), ExitStack() as stack: + stack.enter_context( + patch.object( + MasterDao, + f'get_{resource}', + new=AsyncMock(return_value=SimpleNamespace(**{RESOURCE_IDS[resource]: 7})), + ) + ) + stack.enter_context( + patch.object( + MasterDao, + parent_method, + new=AsyncMock(return_value=SimpleNamespace(status='0')), + ) + ) + stack.enter_context( + patch.object(MasterDao, f'find_{resource}_by_code', new=AsyncMock(return_value=None)) + ) + stack.enter_context(patch.object(MasterDao, f'update_{resource}', new=AsyncMock())) + result = await getattr(MasterService, f'edit_{resource}')(self.db, model) + self.assertTrue(result.is_success) + + async def test_detail_missing_is_explicit_for_all_resources(self) -> None: + for resource in RESOURCE_IDS: + with ( + self.subTest(resource=resource), + patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=None)), + self.assertRaises(ServiceException), + ): + await getattr(MasterService, f'detail_{resource}')(self.db, 99) + + async def test_reference_delete_protection_covers_all_resources(self) -> None: + for resource in RESOURCE_IDS: + with ( + self.subTest(resource=resource), + patch.object(MasterDao, 'count_references', new=AsyncMock(return_value=1)), + ): + self.db.reset_mock() + with self.assertRaises(ServiceWarning): + await getattr(MasterService, f'remove_{resource}')(self.db, 1) + self.db.rollback.assert_awaited_once() + self.db.commit.assert_not_awaited() + + async def test_unreferenced_existing_records_are_deleted_for_all_resources(self) -> None: + for resource in RESOURCE_IDS: + with self.subTest(resource=resource), ExitStack() as stack: + self.db.reset_mock() + count_mock = stack.enter_context( + patch.object(MasterDao, 'count_references', new=AsyncMock(return_value=0)) + ) + stack.enter_context( + patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=SimpleNamespace())) + ) + delete_mock = stack.enter_context(patch.object(MasterDao, f'delete_{resource}', new=AsyncMock())) + + result = await getattr(MasterService, f'remove_{resource}')(self.db, 1) + + self.assertTrue(result.is_success) + count_mock.assert_awaited_once_with(self.db, resource, 1) + delete_mock.assert_awaited_once_with(self.db, 1) + self.db.commit.assert_awaited_once() + self.db.rollback.assert_not_awaited() + + async def test_remove_missing_record_is_explicit_for_all_resources(self) -> None: + for resource in RESOURCE_IDS: + with self.subTest(resource=resource), ExitStack() as stack: + stack.enter_context(patch.object(MasterDao, 'count_references', new=AsyncMock(return_value=0))) + stack.enter_context(patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=None))) + with self.assertRaises(ServiceException): + await getattr(MasterService, f'remove_{resource}')(self.db, 1) + + async def test_batch_remove_failure_rolls_back_without_delete_or_commit(self) -> None: + delete_model = DeleteStrainModel(strainIds=[1, 2]) + with ( + patch.object(MasterDao, 'count_references', new=AsyncMock(side_effect=[0, 1])), + patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=SimpleNamespace())), + patch.object(MasterDao, 'delete_strain', new=AsyncMock()) as delete_mock, + self.assertRaises(ServiceWarning), + ): + await MasterService.remove_strain(self.db, delete_model) + + delete_mock.assert_not_awaited() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_awaited_once() + + async def test_add_ignores_client_primary_key_before_calling_dao(self) -> None: + model = make_models(77)['strain'] + with ( + patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'add_strain', new=AsyncMock()) as add_mock, + ): + await MasterService.add_strain(self.db, model) + + persisted_model = add_mock.await_args.args[1] + self.assertIsNone(persisted_model.strain_id) + + async def test_integrity_errors_from_add_update_delete_and_commit_are_stable(self) -> None: + for phase in ('add', 'update', 'delete', 'commit'): + db = MagicMock(spec=AsyncSession) + with self.subTest(phase=phase), ExitStack() as stack: + if phase in {'add', 'commit'}: + stack.enter_context( + patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)) + ) + stack.enter_context( + patch.object( + MasterDao, + 'add_strain', + new=AsyncMock(side_effect=integrity_error() if phase == 'add' else None), + ) + ) + if phase == 'commit': + db.commit.side_effect = integrity_error() + call = MasterService.add_strain(db, make_models()['strain']) + elif phase == 'update': + stack.enter_context( + patch.object( + MasterDao, + 'get_strain', + new=AsyncMock(return_value=SimpleNamespace(strain_id=7)), + ) + ) + stack.enter_context( + patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)) + ) + stack.enter_context( + patch.object(MasterDao, 'update_strain', new=AsyncMock(side_effect=integrity_error())) + ) + call = MasterService.edit_strain(db, make_models(7)['strain']) + else: + stack.enter_context(patch.object(MasterDao, 'count_references', new=AsyncMock(return_value=0))) + stack.enter_context( + patch.object(MasterDao, 'get_strain', new=AsyncMock(return_value=SimpleNamespace())) + ) + stack.enter_context( + patch.object(MasterDao, 'delete_strain', new=AsyncMock(side_effect=integrity_error())) + ) + call = MasterService.remove_strain(db, 7) + + with self.assertRaises(ServiceException) as raised: + await call + + self.assertIsNotNone(raised.exception.message) + self.assertNotIn('duplicate', raised.exception.message) + db.rollback.assert_awaited_once() + if phase != 'commit': + db.commit.assert_not_awaited() + + async def test_non_integrity_exception_rolls_back_and_is_reraised(self) -> None: + error = RuntimeError('unexpected') + with ( + patch.object(MasterDao, 'find_strain_by_code', new=AsyncMock(return_value=None)), + patch.object(MasterDao, 'add_strain', new=AsyncMock(side_effect=error)), + self.assertRaises(RuntimeError) as raised, + ): + await MasterService.add_strain(self.db, make_models()['strain']) + + self.assertIs(error, raised.exception) + self.db.rollback.assert_awaited_once() + self.db.commit.assert_not_awaited() + + +class TestMasterDao(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.db = MagicMock(spec=AsyncSession) + + async def test_list_queries_cover_vo_filters_and_stable_primary_key_order(self) -> None: + cases = ( + ( + 'strain', + StrainPageQueryModel(strainCode='S001', strainName='菌种', status='0', pageNum=2, pageSize=5), + ('biz_strain.strain_code LIKE', 'biz_strain.strain_name LIKE', 'biz_strain.status =', 'strain_id'), + ), + ( + 'customer', + CustomerPageQueryModel(customerCode='C001', customerName='客户', status='0', pageNum=2, pageSize=5), + ( + 'biz_customer.customer_code LIKE', + 'biz_customer.customer_name LIKE', + 'biz_customer.status =', + 'customer_id', + ), + ), + ( + 'site', + SitePageQueryModel( + customerId=11, + siteCode='SITE001', + siteName='现场', + status='0', + pageNum=2, + pageSize=5, + ), + ( + 'biz_customer_site.customer_id = 11', + 'biz_customer_site.site_code LIKE', + 'biz_customer_site.site_name LIKE', + 'biz_customer_site.status =', + 'site_id', + ), + ), + ( + 'line', + LinePageQueryModel(siteId=22, lineCode='L001', lineName='产线', status='0', pageNum=2, pageSize=5), + ( + 'biz_production_line.site_id = 22', + 'biz_production_line.line_code LIKE', + 'biz_production_line.line_name LIKE', + 'biz_production_line.status =', + 'line_id', + ), + ), + ( + 'warehouse', + WarehousePageQueryModel(warehouseCode='W001', warehouseName='仓库', status='0', pageNum=2, pageSize=5), + ( + 'biz_warehouse.warehouse_code LIKE', + 'biz_warehouse.warehouse_name LIKE', + 'biz_warehouse.status =', + 'warehouse_id', + ), + ), + ) + page = PageModel(rows=[], pageNum=2, pageSize=5, total=0, hasNext=False) + for resource, query_model, expected_tokens in cases: + with ( + self.subTest(resource=resource), + patch.object(PageUtil, 'paginate', new=AsyncMock(return_value=page)) as paginate_mock, + ): + result = await getattr(MasterDao, f'list_{resource}')(self.db, query_model, is_page=True) + statement = paginate_mock.await_args.args[1] + sql = compile_sql(statement) + self.assertIs(page, result) + for token in expected_tokens[:-1]: + self.assertIn(token, sql) + self.assertIn(f'ORDER BY {statement.get_final_froms()[0].name}.{expected_tokens[-1]}', sql) + self.assertEqual((2, 5, True), paginate_mock.await_args.args[2:]) + + async def test_get_queries_use_expected_table_and_primary_key(self) -> None: + cases = ( + ('strain', 'biz_strain', 'strain_id'), + ('customer', 'biz_customer', 'customer_id'), + ('site', 'biz_customer_site', 'site_id'), + ('line', 'biz_production_line', 'line_id'), + ('warehouse', 'biz_warehouse', 'warehouse_id'), + ) + for resource, table, id_field in cases: + result = MagicMock() + result.scalar_one_or_none.return_value = None + self.db.execute = AsyncMock(return_value=result) + + await getattr(MasterDao, f'get_{resource}')(self.db, 9) + + sql = compile_sql(self.db.execute.await_args.args[0]) + self.assertIn(f'FROM {table}', sql) + self.assertIn(f'{table}.{id_field} = 9', sql) + + async def test_unique_queries_use_global_or_parent_scoped_predicates(self) -> None: + cases = ( + ('strain', ('S001',), 'biz_strain', ('strain_code',)), + ('customer', ('C001',), 'biz_customer', ('customer_code',)), + ('warehouse', ('W001',), 'biz_warehouse', ('warehouse_code',)), + ('site', ('SITE001', 11), 'biz_customer_site', ('site_code', 'customer_id')), + ('line', ('L001', 22), 'biz_production_line', ('line_code', 'site_id')), + ) + for resource, args, table, predicates in cases: + result = MagicMock() + result.scalar_one_or_none.return_value = None + self.db.execute = AsyncMock(return_value=result) + + await getattr(MasterDao, f'find_{resource}_by_code')(self.db, *args) + + sql = compile_sql(self.db.execute.await_args.args[0]) + self.assertIn(f'FROM {table}', sql) + for predicate in predicates: + self.assertIn(f'{table}.{predicate} =', sql) + + async def test_add_ignores_client_primary_keys_flushes_and_never_commits(self) -> None: + expected_classes = { + 'strain': BizStrain, + 'customer': BizCustomer, + 'site': BizCustomerSite, + 'line': BizProductionLine, + 'warehouse': BizWarehouse, + } + for resource, model in make_models(77).items(): + with self.subTest(resource=resource): + self.db.reset_mock() + record = await getattr(MasterDao, f'add_{resource}')(self.db, model) + + self.assertIsInstance(record, expected_classes[resource]) + self.assertIsNone(getattr(record, RESOURCE_IDS[resource])) + self.db.add.assert_called_once_with(record) + self.db.flush.assert_awaited_once() + self.db.commit.assert_not_awaited() + self.db.rollback.assert_not_awaited() + + async def test_update_and_delete_statements_use_primary_key_and_preserve_creation_audit(self) -> None: + table_names = { + 'strain': 'biz_strain', + 'customer': 'biz_customer', + 'site': 'biz_customer_site', + 'line': 'biz_production_line', + 'warehouse': 'biz_warehouse', + } + created_at = datetime(2026, 1, 2, 3, 4, 5) + for resource, model in make_models(7).items(): + model.create_by = '客户端伪造创建人' + model.create_time = created_at + model.update_by = 'server-user' + with self.subTest(resource=resource): + self.db.reset_mock() + await getattr(MasterDao, f'update_{resource}')(self.db, model) + update_sql = compile_sql(self.db.execute.await_args.args[0]) + table = table_names[resource] + id_field = RESOURCE_IDS[resource] + self.assertIn(f'UPDATE {table} SET', update_sql) + self.assertIn(f'WHERE {table}.{id_field} = 7', update_sql) + self.assertNotIn('create_by', update_sql) + self.assertNotIn('create_time', update_sql) + self.db.commit.assert_not_awaited() + + self.db.reset_mock() + await getattr(MasterDao, f'delete_{resource}')(self.db, 7) + delete_sql = compile_sql(self.db.execute.await_args.args[0]) + self.assertIn(f'DELETE FROM {table}', delete_sql) + self.assertIn(f'{table}.{id_field} = 7', delete_sql) + self.db.commit.assert_not_awaited() + + async def test_count_references_covers_every_direct_foreign_key(self) -> None: + expected = { + 'strain': (('biz_filling_record', ('strain_id',)),), + 'customer': (('biz_customer_site', ('customer_id',)),), + 'site': ( + ('biz_production_line', ('site_id',)), + ('biz_user_site', ('site_id',)), + ('biz_shipment_order', ('target_site_id',)), + ('biz_usage_record', ('site_id',)), + ('biz_return_order', ('source_site_id',)), + ('biz_pallet_inventory', ('site_id',)), + ('biz_pallet_event', ('before_site_id', 'after_site_id')), + ), + 'line': ( + ('biz_usage_record', ('line_id',)), + ('biz_pallet_inventory', ('current_line_id',)), + ('biz_pallet_event', ('before_line_id', 'after_line_id')), + ), + 'warehouse': ( + ('biz_user_warehouse', ('warehouse_id',)), + ('biz_inbound_order', ('warehouse_id',)), + ('biz_filling_record', ('warehouse_id',)), + ('biz_shipment_order', ('source_warehouse_id',)), + ('biz_return_order', ('target_warehouse_id',)), + ('biz_pallet_inventory', ('warehouse_id',)), + ('biz_pallet_event', ('before_warehouse_id', 'after_warehouse_id')), + ), + } + for resource, references in expected.items(): + results = [] + for _ in references: + result = MagicMock() + result.scalar.return_value = 0 + results.append(result) + self.db.execute = AsyncMock(side_effect=results) + + count = await MasterDao.count_references(self.db, resource, 9) + + self.assertEqual(0, count) + self.assertEqual(len(references), self.db.execute.await_count) + for call, (table, columns) in zip(self.db.execute.await_args_list, references, strict=True): + sql = compile_sql(call.args[0]) + self.assertIn(f'FROM {table}', sql) + for column in columns: + self.assertIn(f'{table}.{column} = 9', sql) + self.db.commit.assert_not_awaited() + + +class TestMasterControllers(unittest.TestCase): + ROUTERS = { + 'strain': strain_controller_module.strain_controller, + 'customer': customer_controller_module.customer_controller, + 'site': site_controller_module.site_controller, + 'line': line_controller_module.line_controller, + 'warehouse': warehouse_controller_module.warehouse_controller, + } + MODULES = { + 'strain': strain_controller_module, + 'customer': customer_controller_module, + 'site': site_controller_module, + 'line': line_controller_module, + 'warehouse': warehouse_controller_module, + } + + def test_openapi_contains_all_resource_paths_methods_and_response_models(self) -> None: + app = FastAPI() + for router in self.ROUTERS.values(): + app.include_router(router) + schema = app.openapi() + + for resource in self.ROUTERS: + prefix = f'/master/{resource}' + self.assertIn(f'{prefix}/list', schema['paths']) + self.assertIn('get', schema['paths'][f'{prefix}/list']) + self.assertIn(prefix, schema['paths']) + self.assertIn('post', schema['paths'][prefix]) + self.assertIn('put', schema['paths'][prefix]) + self.assertIn(f'{prefix}/{{{resource}_ids}}', schema['paths']) + self.assertIn('delete', schema['paths'][f'{prefix}/{{{resource}_ids}}']) + self.assertIn(f'{prefix}/{{{resource}_id}}', schema['paths']) + self.assertIn('get', schema['paths'][f'{prefix}/{{{resource}_id}}']) + + def test_all_twenty_five_permission_dependencies_are_exact(self) -> None: + expected_permissions = { + (f'/master/{resource}/list', 'GET'): f'master:{resource}:list' for resource in self.ROUTERS + } + for resource in self.ROUTERS: + expected_permissions.update( + { + (f'/master/{resource}/{{{resource}_id}}', 'GET'): f'master:{resource}:query', + (f'/master/{resource}', 'POST'): f'master:{resource}:add', + (f'/master/{resource}', 'PUT'): f'master:{resource}:edit', + (f'/master/{resource}/{{{resource}_ids}}', 'DELETE'): f'master:{resource}:remove', + } + ) + + actual_permissions = {} + for router in self.ROUTERS.values(): + for route in router.routes: + if not isinstance(route, APIRoute): + continue + method = next(iter(route.methods)) + auth_dependencies = [ + dependency.dependency + for dependency in route.dependencies + if isinstance(dependency.dependency, CheckUserInterfaceAuth) + ] + self.assertEqual(1, len(auth_dependencies)) + self.assertTrue(any(isinstance(dependency.dependency, PreAuth) for dependency in route.dependencies)) + actual_permissions[(route.path, method)] = auth_dependencies[0].perm + + self.assertEqual(expected_permissions, actual_permissions) + self.assertEqual(25, len(actual_permissions)) + + def test_write_routes_have_log_and_current_user_dependency(self) -> None: + for router in self.ROUTERS.values(): + for route in router.routes: + if not isinstance(route, APIRoute) or route.methods.isdisjoint({'POST', 'PUT', 'DELETE'}): + continue + self.assertTrue(hasattr(route.endpoint, '__wrapped__')) + self.assertTrue( + any( + dependency.name == 'current_user' and dependency.call == LoginService.get_current_user + for dependency in route.dependant.dependencies + ) + ) + + def test_controller_source_has_three_log_and_current_user_declarations(self) -> None: + for resource, module in self.MODULES.items(): + with self.subTest(resource=resource): + source = inspect.getsource(module) + self.assertEqual(3, source.count('@Log(')) + self.assertEqual(3, source.count('CurrentUserDependency()')) + self.assertNotIn('ApiCache', source) + + def test_route_response_model_contracts(self) -> None: + model_by_resource = { + 'strain': StrainModel, + 'customer': CustomerModel, + 'site': SiteModel, + 'line': LineModel, + 'warehouse': WarehouseModel, + } + for resource, router in self.ROUTERS.items(): + for route in router.routes: + if not isinstance(route, APIRoute): + continue + if route.path.endswith('/list'): + self.assertEqual(PageResponseModel[model_by_resource[resource]], route.response_model) + elif route.methods == {'GET'}: + self.assertEqual(DataResponseModel[model_by_resource[resource]], route.response_model) + else: + self.assertEqual('ResponseBaseModel', route.response_model.__name__) + + +class TestDeleteModelCoverage(unittest.TestCase): + def test_all_delete_models_are_used_by_service_contract(self) -> None: + models = ( + DeleteStrainModel(strainIds=[1]), + DeleteCustomerModel(customerIds=[1]), + DeleteSiteModel(siteIds=[1]), + DeleteLineModel(lineIds=[1]), + DeleteWarehouseModel(warehouseIds=[1]), + ) + self.assertEqual(5, len(models)) + self.assertTrue(all(next(iter(model.model_dump().values())) == [1] for model in models)) + + +if __name__ == '__main__': + unittest.main()