80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""通知 API 端点 — 获取列表、标记已读"""
|
|
from __future__ import annotations
|
|
import uuid
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.models.notification import Notification
|
|
from app.schemas.notification import NotificationResponse, NotificationListResponse
|
|
|
|
router = APIRouter(prefix="/notifications", tags=["消息通知"])
|
|
|
|
|
|
@router.get("/", response_model=NotificationListResponse)
|
|
async def list_notifications(
|
|
user_id: str = Query(..., description="当前用户ID"),
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(20, ge=1, le=100),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""获取当前用户的通知列表（按时间倒序）"""
|
|
# 总数
|
|
count_stmt = select(func.count()).select_from(Notification).where(
|
|
Notification.user_id == user_id
|
|
)
|
|
total_result = await db.execute(count_stmt)
|
|
total = total_result.scalar() or 0
|
|
|
|
# 未读数
|
|
unread_stmt = select(func.count()).select_from(Notification).where(
|
|
Notification.user_id == user_id,
|
|
Notification.is_read.is_(False),
|
|
)
|
|
unread_result = await db.execute(unread_stmt)
|
|
unread_count = unread_result.scalar() or 0
|
|
|
|
# 列表
|
|
stmt = (
|
|
select(Notification)
|
|
.where(Notification.user_id == user_id)
|
|
.order_by(Notification.created_at.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
result = await db.execute(stmt)
|
|
notifications = result.scalars().all()
|
|
|
|
return NotificationListResponse(
|
|
notifications=[
|
|
NotificationResponse.model_validate(n) for n in notifications
|
|
],
|
|
total=total,
|
|
unread_count=unread_count,
|
|
)
|
|
|
|
|
|
@router.put("/{notification_id}/read", response_model=NotificationResponse)
|
|
async def mark_notification_read(
|
|
notification_id: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""标记单条通知为已读"""
|
|
nid = uuid.UUID(notification_id)
|
|
result = await db.execute(
|
|
select(Notification).where(Notification.id == nid)
|
|
)
|
|
notification = result.scalar_one_or_none()
|
|
if not notification:
|
|
from fastapi import HTTPException, status
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"通知不存在: {notification_id}",
|
|
)
|
|
|
|
notification.is_read = True
|
|
await db.commit()
|
|
await db.refresh(notification)
|
|
return NotificationResponse.model_validate(notification)
|