python-fastapi-mm-zhcs-yj-api/routers/api/pattern/template.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 common.websocketManager import *
import json
import traceback

router = APIRouter()

def template_id_get_template_info(db,id):
    query = db.query(TpPatternTemplate)
    query = query.filter_by(template_id = id,del_flag = '0')
    return query.first()



@router.post("/create")
async def create_template(
    user_id=Depends(valid_access_token),
    body = Depends(remove_xss_json),
    db: Session = Depends(get_db)
):
    try:
        new_template = TpPatternTemplate(
            template_id=new_guid(),
            name=body['name'],
            value=body['value'],
            order_num=body['order_num'],
            visible=body['visible'],
            create_by = user_id
        )
        db.add(new_template)
        db.commit()
        return {"code": 200, "msg": "创建成功", "data": None}
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")



@router.put("/update/{id}")
async def update_pattern(
    id :str ,
    user_id=Depends(valid_access_token),
    body=Depends(remove_xss_json),
    db: Session = Depends(get_db)
):
    try:
        update_template = template_id_get_template_info(db,id)
        if not update_template:
            return JSONResponse(status_code=404,content={"code":404,"msg":"template not found"})

        update_template.name = body['name']
        update_template.value = body['value']
        update_template.order_num=body['order_num']
        update_template.visible=body['visible']
        update_template.update_by = user_id
        db.commit()
        return {"code": 200, "msg": "更新成功"}
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@router.get("/info/{id}")
async def get_pattern_info(
    id: str,
    db: Session = Depends(get_db)
):
    try:

        info = template_id_get_template_info(db,id)
        if not info:
            return JSONResponse(status_code=404,content={"code":404,"msg":"template not found"})
        data = {"template_id": info.template_id,
                "name": info.name,
                "value": info.value,
                "order_num": info.order_num,
                "visible": info.visible,
                "create_time": info.create_time,
                "create_by": info.create_by}
        return {"code": 200, "msg": "获取成功", "data": data}
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@router.get("/list")
async def get_pattern_list(
    name: 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(TpPatternTemplate)
        query = query.filter_by(del_flag='0')
        if name:
            query = query.filter(TpPatternTemplate.name.like(f'%{name}%'))
        total_items = query.count()
        # 排序
        if pageSize is None:
            pageSize=total_items
        query = query.order_by(TpPatternTemplate.order_num.asc())
        # 执行分页查询
        lists = query.offset((page - 1) * pageSize).limit(pageSize).all()
        data = [{"template_id": info.template_id,
                "name": info.name,
                "value": info.value,
                "order_num": info.order_num,
                "visible": info.visible,
                "create_time": info.create_time,
                "create_by": info.create_by} for info in lists]
        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)}")

@router.delete("/delete/{id}")
async def delete_pattern(
    id: str,
    db: Session = Depends(get_db)
):
    try:
        # 检查图案是否存在
        info = template_id_get_template_info(db, id)
        if not info:
            return JSONResponse(status_code=404, content={"code": 404, "msg": "template not found"})
        info.del_flag='2'
        db.commit()
        return {"code": 200, "msg": "删除成功"}
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")