You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

66 lines
2.6 KiB

7 months ago
10 months ago
9 months ago
7 months ago
10 months ago
8 months ago
11 months ago
9 months ago
10 months ago
7 months ago
8 months ago
9 months ago
7 months ago
8 months ago
9 months ago
7 months ago
8 months ago
9 months ago
7 months ago
9 months ago
8 months ago
9 months ago
7 months ago
9 months ago
8 months ago
9 months ago
8 months ago
7 months ago
8 months ago
9 months ago
  1. from fastapi import Depends, APIRouter, status, Query, Path, HTTPException,Request
  2. from internal.models import *
  3. from internal.database import fetch_one, fetch_all, execute_query, response_success, raise_if_exists,raise_if_not_found
  4. from dependencies import get_current_active_user
  5. from limiter_config import limiter
  6. router = APIRouter(
  7. prefix="/blogtypes",
  8. tags=['博客分类']
  9. )
  10. # 获取列表
  11. @router.get("/list")
  12. @limiter.limit("10/minute")
  13. async def type_list(request: Request,):
  14. select_query = "SELECT id,typename,descr FROM blogtypes;"
  15. type_list = fetch_all(select_query)
  16. return response_success(type_list, "type get list success")
  17. # 新增
  18. @router.post("/add")
  19. @limiter.limit("10/minute")
  20. async def type_add(request: Request,type:Type,_: User = Depends(get_current_active_user)):
  21. insert_query="""INSERT INTO blogtypes (typename,descr) VALUES(%s,%s)"""
  22. insert_value=(type.typename,type.descr)
  23. execute_query(insert_query,insert_value)
  24. return response_success(data=type,message="type create success")
  25. # 单条数据查询
  26. @router.get("/list/search")
  27. @limiter.limit("10/minute")
  28. async def type_search(request: Request,typename:str=Query(None,description="类型名称")):
  29. select_query="SELECT id,typename,descr FROM blogtypes WHERE 1=1 "
  30. params=[]
  31. if typename:
  32. select_query+="AND typename LIKE %s"
  33. params.append(f"%{typename}%")
  34. type_query=fetch_all(select_query,params=params,fetchall=True)
  35. return response_success(data=type_query,message="type search success")
  36. # 语录修改
  37. @router.put("/update/{id}")
  38. @limiter.limit("10/minute")
  39. async def type_put(request: Request,type:Type,id: str = Path(description="类型id"),_: User = Depends(get_current_active_user)):
  40. update_query = (
  41. "UPDATE blogtypes SET typename=%s,descr=%s WHERE id=%s;"
  42. )
  43. update_data = (type.typename,type.descr,id)
  44. execute_query(update_query, update_data)
  45. return response_success("type update sucess")
  46. # 语录删除
  47. @router.delete("/delete/{id}")
  48. @limiter.limit("10/minute")
  49. async def type_del(request: Request,id: str = Path(description="类型id"),_: User = Depends(get_current_active_user)):
  50. update_query = (
  51. "DELETE FROM blogtypes WHERE id=%s;"
  52. )
  53. update_data = (id)
  54. execute_query(update_query, update_data)
  55. return response_success()
  56. # 根据id查询语录
  57. @router.get("/search/{id}")
  58. @limiter.limit("10/minute")
  59. async def type_search_id(request: Request,id:str=Path(description="类型id")):
  60. select_query="SELECT * FROM blogtypes WHERE id=%s"
  61. type_query=fetch_one(select_query,(id,))
  62. return response_success(data=type_query,message="type search success")