135 lines
5.2 KiB
Python
135 lines
5.2 KiB
Python
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.dependency.test_batch_dependency import TestBatchScopeDependency
|
|
from module_business.domain.test_batch_scope import TestBatchScope
|
|
from module_business.entity.vo.operation_vo import (
|
|
BusinessOptionModel,
|
|
FillingCreateModel,
|
|
FillingModel,
|
|
FillingPageQueryModel,
|
|
PalletScanResponseModel,
|
|
)
|
|
from module_business.service.filling_service import FillingService
|
|
from utils.response_util import ResponseUtil
|
|
|
|
filling_controller = APIRouterPro(
|
|
prefix='/business/filling',
|
|
tags=['业务管理-托盘装填'],
|
|
dependencies=[PreAuthDependency()],
|
|
)
|
|
|
|
|
|
@filling_controller.get(
|
|
'/list',
|
|
summary='获取托盘装填记录分页列表接口',
|
|
description='按当前用户仓库数据范围获取托盘装填记录分页列表',
|
|
response_model=PageResponseModel[FillingModel],
|
|
dependencies=[UserInterfaceAuthDependency('business:filling:list')],
|
|
)
|
|
async def get_business_filling_list(
|
|
request: Request,
|
|
filling_page_query: Annotated[FillingPageQueryModel, Query()],
|
|
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
|
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
|
test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()],
|
|
) -> Response:
|
|
result = await FillingService.get_filling_list_services(
|
|
query_db,
|
|
filling_page_query,
|
|
current_user,
|
|
test_scope,
|
|
is_page=True,
|
|
)
|
|
return ResponseUtil.success(model_content=result)
|
|
|
|
|
|
@filling_controller.get(
|
|
'/strain-options',
|
|
summary='获取装填菌种选项接口',
|
|
description='获取全部启用菌种的业务选项',
|
|
response_model=DataResponseModel[list[BusinessOptionModel]],
|
|
dependencies=[UserInterfaceAuthDependency('business:filling:query')],
|
|
)
|
|
async def get_business_filling_strain_options(
|
|
request: Request,
|
|
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
|
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
|
test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()],
|
|
) -> Response:
|
|
result = await FillingService.get_strain_options_services(query_db, current_user, test_scope)
|
|
return ResponseUtil.success(data=result)
|
|
|
|
|
|
@filling_controller.get(
|
|
'/scan/{qr_token}',
|
|
summary='扫描待装填托盘接口',
|
|
description='按托盘当前仓库数据范围判断空托盘是否可装填',
|
|
response_model=DataResponseModel[PalletScanResponseModel],
|
|
dependencies=[UserInterfaceAuthDependency('business:filling:query')],
|
|
)
|
|
async def scan_business_filling_pallet(
|
|
request: Request,
|
|
qr_token: Annotated[
|
|
str,
|
|
Path(min_length=1, max_length=128, pattern=r'\S', description='二维码固定标识'),
|
|
],
|
|
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
|
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
|
test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()],
|
|
) -> Response:
|
|
result = await FillingService.scan_filling_pallet_services(
|
|
query_db,
|
|
qr_token,
|
|
current_user,
|
|
test_scope,
|
|
)
|
|
return ResponseUtil.success(data=result)
|
|
|
|
|
|
@filling_controller.get(
|
|
'/{filling_id}',
|
|
summary='获取托盘装填记录详情接口',
|
|
description='按仓库数据范围获取指定托盘装填记录',
|
|
response_model=DataResponseModel[FillingModel],
|
|
dependencies=[UserInterfaceAuthDependency('business:filling:query')],
|
|
)
|
|
async def query_detail_business_filling(
|
|
request: Request,
|
|
filling_id: Annotated[int, Path(gt=0, description='装填记录ID')],
|
|
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
|
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
|
test_scope: Annotated[TestBatchScope, TestBatchScopeDependency()],
|
|
) -> Response:
|
|
result = await FillingService.filling_detail_services(query_db, filling_id, current_user, test_scope)
|
|
return ResponseUtil.success(data=result)
|
|
|
|
|
|
@filling_controller.post(
|
|
'',
|
|
summary='单托盘装填接口',
|
|
description='在仓库内即时确认单个空托盘装填',
|
|
response_model=ResponseBaseModel,
|
|
dependencies=[UserInterfaceAuthDependency('business:filling:add')],
|
|
)
|
|
@Log(title='托盘装填', business_type=BusinessType.INSERT)
|
|
async def add_business_filling(
|
|
request: Request,
|
|
add_filling: FillingCreateModel,
|
|
query_db: Annotated[AsyncSession, DBSessionDependency()],
|
|
current_user: Annotated[CurrentUserModel, CurrentUserDependency()],
|
|
test_scope: Annotated[TestBatchScope, TestBatchScopeDependency(write=True)],
|
|
) -> Response:
|
|
result = await FillingService.fill_pallet_services(query_db, add_filling, current_user, test_scope)
|
|
return ResponseUtil.success(msg=result.message)
|