#!/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.enc import mpfun, sys_post_data, sys_user_post_data from common.auth_user import * from common.db import db_czrz 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, sign='' ) db.add(new_post) db.commit() db.refresh(new_post) sys_post_data.sign_table() 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'] post.sign = '' db.commit() sys_post_data.sign_table() 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))