blog_api/routers/typemanage.py

from fastapi import Depends, APIRouter, status, Query, Path, HTTPException
from internal.models import *
from internal.database import fetch_one, fetch_all, execute_query, response_success, raise_if_exists,raise_if_not_found
from dependencies import get_current_active_user

router = APIRouter(
    prefix="/types",
    tags=['分类管理']
)
# 获取列表
@router.get("/list")
async def type_list():
    select_query = "SELECT id,typename,descr FROM types;"
    type_list = fetch_all(select_query)
    return response_success(type_list, "type get list success")

# 新增
@router.post("/add")
async def type_add(type:Type):
    insert_query="""INSERT INTO types (typename,descr) VALUES(%s,%s)"""
    insert_value=(type.typename,type.descr)
    execute_query(insert_query,insert_value)
    return response_success(data=type,message="type create success")

# 单条数据查询
@router.get("/list/search")
async def type_search(typename:str=Query(description="类型名称")):
    select_query="SELECT id,typename,descr FROM types WHERE 1=1 "
    params=[]
    if typename:
        select_query+="AND typename LIKE %s"
        params.append(f"%{typename}%")
    type_query=fetch_all(select_query,params=params,fetchall=True)
    return response_success(data=type_query,message="type search success")

# 语录修改
@router.put("/update/{id}")
async def type_put(type:Type,id: str = Path(description="类型id")):
    update_query = (
        "UPDATE types SET typename=%s,descr=%s WHERE id=%s;"
    )
    update_data = (type.typename,type.descr,id)  
    execute_query(update_query, update_data)
    return response_success("type update sucess")

# 语录删除
@router.delete("/delete/{id}")
async def type_del(id: str = Path(description="类型id")):
    update_query = (
        "DELETE FROM types WHERE id=%s;"
    )
    update_data = (id)  
    execute_query(update_query, update_data)
    return response_success()

# 根据id查询语录
@router.get("/list/search/{id}")
async def type_search_id(id:str=Path(description="类型id")):
    select_query="SELECT * FROM types WHERE id=%s"
    type_query=fetch_one(select_query,(id,))
    return response_success(data=type_query,message="type search success")