python-fastapi-mm-zhcs-yj-api/routers/api/resourceProvison/MaterialReserveManagement/warehouse_movement.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from fastapi import APIRouter, Request, Depends, Query, HTTPException, status,WebSocket,WebSocketDisconnect
from common.security import valid_access_token,valid_websocket_token
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from sqlalchemy.sql import func
from common.auth_user import *
from sqlalchemy import  text
from pydantic import BaseModel
from common.BigDataCenterAPI import *
from database import get_db
from typing import List
from models import *
from utils import *
from utils.spatial import *
from utils.ry_system_util import *
from utils.resource_provision_util import *
from common.websocketManager import *
import json
import traceback

router = APIRouter()


@router.get("/list")
async def get_pattern_list(
    material_id: str = Query(None, description='物资id'),
    io_number: str = Query(None, description='出入库单号'),
    io_flag: str = Query(None, description='出入库标识'),
    page: int = Query(1, gt=0, description='页码'),
    pageSize: int = Query(None, gt=0, description='每页条目数量'),
    db: Session = Depends(get_db)
):
    try:
        query = db.query(ResourceProvisionWarehouseMovement)
        query = query.filter_by(del_flag='0')
        if material_id:
            query = query.filter(ResourceProvisionWarehouseMovement.material_id.like(f'%{material_id}%'))
        if io_number:
            query = query.filter(ResourceProvisionWarehouseMovement.io_number.like(f'%{io_number}%'))
        if io_flag:
            query = query.filter(ResourceProvisionWarehouseMovement.io_flag==io_flag)
        total_items = query.count()
        if pageSize is None:
            pageSize = total_items
            if total_items==0:
                pageSize = 1
        # 排序
        query = query.order_by(ResourceProvisionWarehouseMovement.create_time.desc())
        # 执行分页查询
        lists = query.offset((page - 1) * pageSize).limit(pageSize).all()
        data = [ ]
        for info in lists:
            material_info = material_id_get_material_info(db,material_id)
            data.append({"id": info.id,
                "product_info": f"ID: {info.material_id}/n{material_info.material_name}",
                "io_number": info.io_number,
                "io_flag": info.io_flag,
                "changed_stock": info.changed_stock,
                "remaining_stock": info.remaining_stock,
                "create_time":info.create_time})
        return {"code": 200, "msg": "查询成功", "data": data,
                "total": total_items,
                "page": page,
                "pageSize": pageSize,
                "totalPages": (total_items + pageSize - 1) // pageSize
                }
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")