42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from typing import Optional
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from backend.database import get_db
|
|
from backend import crud
|
|
from backend.schemas import Notification
|
|
|
|
router = APIRouter(prefix="/api/notifications", tags=["history"])
|
|
|
|
@router.get("")
|
|
async def list_notifications(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
channel_id: Optional[int] = None,
|
|
status: Optional[str] = Query(None, enum=["pending", "sent", "failed"]),
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""获取通知历史记录"""
|
|
notifications = await crud.get_notifications(db, skip, limit, channel_id, status)
|
|
return {
|
|
"notifications": notifications,
|
|
"skip": skip,
|
|
"limit": limit
|
|
}
|
|
|
|
@router.get("/{notification_id}")
|
|
async def get_notification(notification_id: int, db: AsyncSession = Depends(get_db)):
|
|
"""获取单条通知详情"""
|
|
from sqlalchemy import select
|
|
from backend.models import Notification
|
|
|
|
result = await db.execute(
|
|
select(Notification).where(Notification.id == notification_id)
|
|
)
|
|
notification = result.scalar_one_or_none()
|
|
|
|
if not notification:
|
|
from fastapi import HTTPException
|
|
raise HTTPException(status_code=404, detail="Notification not found")
|
|
|
|
return notification
|