announcement.py 1.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960
  1. """
  2. 公告接口(客户端使用)
  3. """
  4. from typing import Optional
  5. from fastapi import APIRouter, Depends, Query
  6. from sqlalchemy.orm import Session
  7. from sqlalchemy import or_
  8. from app.database import get_db
  9. from app.models.announcement import Announcement
  10. router = APIRouter()
  11. @router.get("")
  12. async def get_announcement(
  13. page: int = Query(1, ge=1),
  14. page_size: int = Query(10, ge=1, le=100),
  15. keyword: Optional[str] = Query(None, description="搜索关键词(标题、内容)"),
  16. db: Session = Depends(get_db),
  17. ):
  18. """获取公告列表(支持检索与分页),按发布时间倒序"""
  19. offset = (page - 1) * page_size
  20. query = db.query(Announcement).filter(Announcement.is_active == 1)
  21. if keyword:
  22. like_pattern = f"%{keyword}%"
  23. query = query.filter(
  24. or_(
  25. Announcement.title.like(like_pattern),
  26. Announcement.content.like(like_pattern),
  27. )
  28. )
  29. # 按发布时间倒序,其次按ID倒序,确保最新的在最前面
  30. # 这里不使用 nullslast,以避免部分数据库方言不支持导致的错误
  31. query = query.order_by(Announcement.publish_time.desc(), Announcement.id.desc())
  32. total = query.count()
  33. announcements = query.offset(offset).limit(page_size).all()
  34. return {
  35. "code": 200,
  36. "msg": "成功",
  37. "data": [
  38. {
  39. "id": a.id,
  40. "title": a.title,
  41. "content": a.content,
  42. "publish_time": a.publish_time.isoformat() if a.publish_time else None,
  43. }
  44. for a in announcements
  45. ],
  46. "total": total,
  47. "page": page,
  48. "page_size": page_size,
  49. }