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.

112 lines
4.0 KiB

10 months ago
9 months ago
10 months ago
9 months ago
10 months ago
9 months ago
10 months ago
9 months ago
10 months ago
10 months ago
9 months ago
10 months ago
9 months ago
10 months ago
10 months ago
9 months ago
10 months ago
9 months ago
10 months ago
10 months ago
9 months ago
10 months ago
10 months ago
9 months ago
10 months ago
9 months ago
10 months ago
10 months ago
9 months ago
10 months ago
10 months ago
10 months ago
10 months ago
10 months ago
10 months ago
10 months ago
10 months ago
10 months ago
10 months ago
9 months ago
10 months ago
9 months ago
  1. from fastapi import Depends, APIRouter, status, Query, Path, HTTPException
  2. from internal.models import *
  3. from internal.database import (
  4. fetch_one,
  5. fetch_all,
  6. execute_query,
  7. response_success,
  8. raise_if_exists,
  9. raise_if_not_found,
  10. )
  11. from dependencies import get_current_active_user
  12. router = APIRouter(prefix="/blogs", tags=["博客管理"])
  13. # 获取列表
  14. @router.get("/list")
  15. async def blog_list():
  16. select_query = """
  17. SELECT blogs.blogtitle, blogs.blogcontent, blogs.create_at, blogs.img,
  18. blogs.readminite, blogs.readnum, blogs.wordcount, types.typename,
  19. JSON_ARRAYAGG(labels.labelname) AS labelnames
  20. FROM blogs
  21. LEFT JOIN `types` ON blogs.typeid = types.id
  22. LEFT JOIN blog_label ON blog_label.blogid = blogs.id
  23. LEFT JOIN labels ON blog_label.labelid = labels.id
  24. GROUP BY blogs.blogtitle, blogs.blogcontent, blogs.create_at, blogs.img,
  25. blogs.readminite, blogs.readnum, blogs.wordcount, types.typename;
  26. """
  27. blog_list = fetch_all(select_query)
  28. return response_success(blog_list, "blog get list success")
  29. # 博客新增
  30. @router.post("/add")
  31. async def blog_add(blog: Blog, _: User = Depends(get_current_active_user)):
  32. select_query = "SELECT * FROM blogs WHERE blogtitle = %s"
  33. existing_blog = fetch_one(select_query, (blog.blogtitle,))
  34. raise_if_exists(existing_blog, "blog is already exists")
  35. insert_query = (
  36. "INSERT INTO blogs (blogtitle,blogcontent,typeid,descr) VALUES (%s,%s,%s,%s)"
  37. )
  38. insert_data = (blog.blogtitle, blog.blogcontent, blog.typeid, blog.descr)
  39. execute_query(insert_query, insert_data)
  40. return response_success(data=blog, message="blog create success")
  41. # 博客删除
  42. @router.delete("/delete/{id}")
  43. async def blog_delete(id: str = Path(description="博客id")):
  44. select_query = "SELECT * FROM blogs WHERE id = %s"
  45. existing_blog = fetch_one(select_query, (id,))
  46. raise_if_not_found(existing_blog, "blog not found")
  47. insert_query = "DELETE FROM blogs WHERE id = %s"
  48. execute_query(insert_query, (id,))
  49. return response_success(message="blog delete success")
  50. # 博客修改
  51. @router.put("/update/{id}")
  52. async def blog_put(blog: Blog, id: str = Path(description="博客id")):
  53. select_query = "SELECT * FROM blogs WHERE id=%s"
  54. existing_blog = fetch_one(select_query, (id,))
  55. raise_if_not_found(existing_blog, "blog not found")
  56. update_query = (
  57. "UPDATE blogs SET blogtitle=%s,blogcontent=%s,typeid=%s,descr=%s WHERE id=%s;"
  58. )
  59. update_data = (blog.blogtitle, blog.blogcontent,
  60. blog.typeid, blog.descr, id)
  61. execute_query(update_query, update_data)
  62. return response_success("blog update sucess")
  63. # 博客模糊查询
  64. @router.get("/list/search")
  65. async def blog_list_search(
  66. blogtitle: str = Query(None, description="博客标题"),
  67. typename: str = Query(None, description="博客类型"),
  68. start_date: str = Query(None, description="开始时间"),
  69. end_date: str = Query(None, description="结束时间"),
  70. ):
  71. select_query = "SELECT blogs.id, blogtitle, blogcontent, typename, create_at, update_at, blogs.descr FROM blogs LEFT JOIN `types` ON blogs.typeid = types.id WHERE 1=1"
  72. params = []
  73. if blogtitle:
  74. select_query += " AND blogtitle LIKE %s"
  75. params.append(f"%{blogtitle}%")
  76. if typename:
  77. select_query += " AND typename LIKE %s"
  78. params.append(f"%{typename}%")
  79. if start_date:
  80. select_query += " AND create_at >= %s"
  81. params.append(start_date)
  82. if end_date:
  83. select_query += " AND create_at <= %s"
  84. params.append(end_date)
  85. select_query += " ORDER BY create_at DESC"
  86. blog_list = fetch_all(select_query, params=params, fetchall=True)
  87. return response_success(data=blog_list, message="blog serach succuessfully!")
  88. # 根据id查询博客
  89. @router.get("/list/search/{id}")
  90. async def get_id_blog(id: str = Path(description="博客id")):
  91. select_query = "SELECT * FROM blogs WHERE id=%s"
  92. blog_list = execute_query(select_query, (id,))
  93. return response_success(data=blog_list, message="blog search success")
  94. # 我就测试一下