python-fastapi-mm-zhcs-yj-api/routers/prod_api/system/post/__init__.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from fastapi import APIRouter, Request, Depends, Query, HTTPException, status
from database import get_db
from sqlalchemy.orm import Session
from sqlalchemy import not_
from fastapi.responses import JSONResponse
from models import *
from utils import *
from utils.ry_system_util import *
from common.security import valid_access_token
from common.db import db_user
import traceback

router = APIRouter()

@router.post('')
async def postcreate(
    db: Session = Depends(get_db),
    user_id: int = Depends(valid_access_token),
    body = Depends(remove_xss_json)
):
    try:
        deptId = body['deptId']
        postCategory = body['postCategory']
        postCode = body['postCode']
        postName = body['postName']
        postSort = body['postSort']
        status = body['status']
        remark = body['remark']

        new_post = SysPost(
            dept_id=deptId,
            post_category=postCategory,
            post_code=postCode,
            post_name=postName,
            post_sort=postSort,
            status=status,
            remark=remark,
            create_by=user_id,
            create_dept=0
        )
        db.add(new_post)
        db.commit()
        db.refresh(new_post)

        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.get('/list')
async def get_list(
    request: Request,
    belongDeptId: str = Query(None, max_length=100),
    postCode: str = Query(None, max_length=100),
    postName:str = Query(None, max_length=100),
    postCategory:str = Query(None, max_length=100),
    deptId:str = Query(None, max_length=100),
    status: str =  Query(None, max_length=100),
    db: Session = Depends(get_db),
    body = Depends(remove_xss_json),
    user_id = Depends(valid_access_token)
):
    query = db.query(SysPost)
    if belongDeptId:
        query = query.filter(SysPost.dept_id == int(belongDeptId))
    if deptId:
        query = query.filter(SysPost.dept_id == int(deptId))
    if postCode:
        query = query.filter(SysPost.post_code.like(f'%{postCode}%'))
    if postName:
        query = query.filter(SysPost.post_name.like(f'%{postName}%'))
    if postCategory:
        query = query.filter(SysPost.post_category.like(f'%{postCategory}%'))
    if status:
        query = query.filter(SysPost.status.like(f'%{status}%'))

    def get_det_name(dept_id):
        dept_info = dept_id_get_dept_info(db, dept_id)
        return dept_info.dept_name if dept_info is not None else ''
     
    post_list = query.all()
    # 将模型实例转换为字典
    dept_list_dict = [{
            "postId": post.post_id,
            "postName": post.post_name,
            "postCode": post.post_code,
            "postCategory": post.post_category,
            "postSort": post.post_sort,
            "deptId": post.dept_id,
            "deptName": get_det_name(post.dept_id),
            "createDept": post.create_dept,
            "createBy": post.create_by,
            "status": post.status,
            "remark": post.remark,
            "createTime": post.create_time.strftime('%Y-%m-%d %H:%M:%S') if post.create_time else '',
        } for post in post_list]


    return {
        "code": 200,
        "rows": dept_list_dict,
        "msg": "操作成功"
    }


@router.get('/optionselect')
async def get_optionselect(
    deptId: int = Query(0),
    db: Session = Depends(get_db),
):
    try:
        # 构建查询
        query = db.query(SysPost)
        query = query.filter(SysPost.status == 0)

        # 计算总记录数
        posts = query.all()

        # 转换为字典
        posts = dept_id_get_dept_post(db, deptId)

        # 构建返回结果
        result = {
            "data": posts,
            "code": 200,
            "msg": "查询成功"
        }
        return result
    except Exception as e:
        # 处理异常
        raise HTTPException(status_code=500, detail=str(e))



@router.put('')
async def postupdate(
    db: Session = Depends(get_db),
    user_id: int = Depends(valid_access_token),
    body = Depends(remove_xss_json)
):
    try:
        postId = body['postId']

        query = db.query(SysPost)
        query = query.filter(SysPost.post_id == postId)
        post = query.first()
        if not post :
            return JSONResponse(status_code=410, content={
            'code': 410,
            'msg': f'岗位{postId}不存在'
        })

        post.dept_id = body['deptId']
        post.post_category = body['postCategory']
        post.post_code = body['postCode']
        post.post_name = body['postName']
        post.post_sort = body['postSort']
        post.status = body['status']
        post.remark = body['remark']

        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.get('/{postId}')
async def postId( postId:int,
    db: Session = Depends(get_db),
    user_id: int = Depends(valid_access_token)):

    try:
        # 构建查询
        query = db.query(SysPost)
        # 应用查询条件
        query = query.filter(SysPost.post_id==postId)

        posts = query.all()

        # 将查询结果转换为列表形式的字典
        post_list = post_list_to_dict(posts, db)

        # 返回结果
        return {
            "code": 200,
            "msg": "成功",
            "data": post_list[0]
        }

    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
    
@router.delete('/{post_id}')
async def delete(
    post_id: int,
    db: Session = Depends(get_db),
    body = Depends(remove_xss_json),
    user_id = Depends(valid_access_token)
):
    try:
        query = db.query(SysPost)
        query = query.filter(SysPost.post_id == post_id)

        post_to_delete = query.first()
        if not post_to_delete:
            detail = "岗位不存在"
            raise HTTPException(status_code=404, detail="岗位不存在")
        db.delete(post_to_delete)
        db.commit()
        return {
            "code": 200,
            "msg": "岗位删除成功"
        }
    except Exception as e:
        db.rollback()
        if str(e)=='':
            e = detail
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))