Spaces:
Build error
Build error
File size: 2,836 Bytes
a282d4b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """
Notifications router — CRUD for user notifications with WebSocket push support.
"""
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import desc
from app.database.database import get_db
from app.database.models import Notification, User, generate_uuid
from datetime import datetime
router = APIRouter(prefix="/api/notifications", tags=["Notifications"])
def _resolve_user(db: Session, user_id: Optional[str]) -> str:
if user_id:
return user_id
user = db.query(User).first()
if not user:
raise HTTPException(status_code=404, detail="No users found.")
return user.id
@router.get("/")
def get_notifications(user_id: Optional[str] = None, limit: int = 20, db: Session = Depends(get_db)):
uid = _resolve_user(db, user_id)
notifications = (
db.query(Notification)
.filter(Notification.user_id == uid)
.order_by(desc(Notification.created_at))
.limit(limit)
.all()
)
return {
"notifications": [
{
"id": n.id,
"title": n.title,
"message": n.message,
"type": n.type,
"read": n.read_status,
"created_at": n.created_at.isoformat() if n.created_at else None,
}
for n in notifications
],
"unread_count": sum(1 for n in notifications if not n.read_status),
}
@router.patch("/{notification_id}/read")
def mark_notification_read(
notification_id: str,
user_id: Optional[str] = None,
db: Session = Depends(get_db)
):
uid = _resolve_user(db, user_id)
notif = db.query(Notification).filter(
Notification.id == notification_id,
Notification.user_id == uid
).first()
if not notif:
raise HTTPException(status_code=404, detail="Notification not found")
notif.read_status = True
db.commit()
return {"success": True}
@router.patch("/read-all")
def mark_all_read(user_id: Optional[str] = None, db: Session = Depends(get_db)):
uid = _resolve_user(db, user_id)
db.query(Notification).filter(
Notification.user_id == uid,
Notification.read_status == False
).update({"read_status": True})
db.commit()
return {"success": True}
@router.delete("/{notification_id}")
def delete_notification(
notification_id: str,
user_id: Optional[str] = None,
db: Session = Depends(get_db)
):
uid = _resolve_user(db, user_id)
notif = db.query(Notification).filter(
Notification.id == notification_id,
Notification.user_id == uid
).first()
if not notif:
raise HTTPException(status_code=404, detail="Notification not found")
db.delete(notif)
db.commit()
return {"success": True}
|