Coverage for app\routers\notifications.py: 44%

1from typing import List, Optional, Any 

2from datetime import datetime, UTC 

3from fastapi import APIRouter, Depends, HTTPException, Query, Path 

4from sqlalchemy.orm import Session 

5from sqlalchemy import desc 

6from pydantic import BaseModel, Field 

7 

8from ..db import Notification, NotificationType, User 

9from ..deps import get_db, get_current_user 

10from ..services.notification_service import NotificationService 

11 

12router = APIRouter() 

13 

14# --- Schemas --- 

15 

16class NotificationSchema(BaseModel): 

17 id: str 

18 type: str 

19 title: str 

20 message: str 

21 recipient_id: Optional[str] = None 

22 is_read: bool 

23 is_dismissed: bool 

24 created_at: datetime 

25 read_at: Optional[datetime] = None 

26 action_url: Optional[str] = None 

27 action_text: Optional[str] = None 

28 priority: str = "medium" # Mapped from type for frontend compatibility 

29 

30 class Config: 

31 from_attributes = True 

32 

33 @staticmethod 

34 def map_priority(notification_type: NotificationType) -> str: 

35 if notification_type == NotificationType.CRITICAL: 

36 return "high" 

37 elif notification_type == NotificationType.ERROR: 

38 return "high" 

39 elif notification_type == NotificationType.WARNING: 

40 return "medium" 

41 else: 

42 return "low" 

43 

44class PaginatedNotifications(BaseModel): 

45 items: List[NotificationSchema] 

46 total: int 

47 page: int 

48 size: int 

49 

50class NotificationCreate(BaseModel): 

51 title: str 

52 message: str 

53 priority: str = "medium" # high, medium, low 

54 user_id: Optional[str] = None 

55 is_read: bool = False 

56 

57class NotificationUpdate(BaseModel): 

58 is_read: Optional[bool] = None 

59 is_dismissed: Optional[bool] = None 

60 

61# --- Helpers --- 

62 

63def get_notification_type(priority: str) -> NotificationType: 

64 if priority == "high": 

65 return NotificationType.ERROR 

66 elif priority == "medium": 

67 return NotificationType.WARNING 

68 else: 

69 return NotificationType.INFO 

70 

71def map_to_schema(notification: Notification) -> NotificationSchema: 

72 schema = NotificationSchema.model_validate(notification) 

73 schema.priority = NotificationSchema.map_priority(notification.type) 

74 return schema 

75 

76# --- Endpoints --- 

77 

78@router.get("", response_model=PaginatedNotifications) 

79def list_notifications( 

80 skip: int = Query(0, alias="offset"), 

81 limit: int = 20, 

82 priority: Optional[str] = None, 

83 is_read: Optional[bool] = None, 

84 search: Optional[str] = None, 

85 db: Session = Depends(get_db), 

86 current_user: User = Depends(get_current_user) 

87): 

88 """List notifications for the current user.""" 

89 query = db.query(Notification) 

90 

91 # Filter by recipient (either me or broadcast) 

92 # Handle dict-like user object from legacy dependency 

93 user_id = current_user.id if hasattr(current_user, 'id') else current_user["id"] 

94 query = query.filter((Notification.recipient_id == user_id) | (Notification.recipient_id == None)) 

95 

96 if is_read is not None: 

97 query = query.filter(Notification.is_read == is_read) 

98 

99 if priority: 

100 # Map priority back to types 

101 if priority == "high": 

102 query = query.filter(Notification.type.in_([NotificationType.ERROR, NotificationType.CRITICAL])) 

103 elif priority == "medium": 

104 query = query.filter(Notification.type == NotificationType.WARNING) 

105 elif priority == "low": 

106 query = query.filter(Notification.type == NotificationType.INFO) 

107 

108 if search: 

109 query = query.filter((Notification.title.ilike(f"%{search}%")) | (Notification.message.ilike(f"%{search}%"))) 

110 

111 total = query.count() 

112 

113 # Sort by created_at desc 

114 query = query.order_by(desc(Notification.created_at)) 

115 

116 notifications = query.offset(skip).limit(limit).all() 

117 

118 items = [map_to_schema(n) for n in notifications] 

119 

120 return { 

121 "items": items, 

122 "total": total, 

123 "page": (skip // limit) + 1, 

124 "size": limit 

125 } 

126 

127@router.post("", response_model=NotificationSchema) 

128def create_notification( 

129 notification_in: NotificationCreate, 

130 db: Session = Depends(get_db), 

131 current_user: User = Depends(get_current_user) 

132): 

133 """Create a new notification (for testing or manual alerts).""" 

134 # Only admins can create notifications for others 

135 if notification_in.user_id and notification_in.user_id != current_user.id: 

136 if current_user.role.value not in ["superadmin", "moderator"]: 

137 raise HTTPException(status_code=403, detail="Not authorized to send notifications to others") 

138 

139 notif_type = get_notification_type(notification_in.priority) 

140 

141 notification = Notification( 

142 type=notif_type, 

143 title=notification_in.title, 

144 message=notification_in.message, 

145 recipient_id=notification_in.user_id, 

146 is_read=notification_in.is_read 

147 ) 

148 

149 db.add(notification) 

150 db.commit() 

151 db.refresh(notification) 

152 

153 return map_to_schema(notification) 

154 

155@router.put("/{id}", response_model=NotificationSchema) 

156def update_notification( 

157 id: str, 

158 update_data: NotificationUpdate, 

159 db: Session = Depends(get_db), 

160 current_user: User = Depends(get_current_user) 

161): 

162 """Update notification status.""" 

163 notification = db.query(Notification).filter(Notification.id == id).first() 

164 if not notification: 

165 raise HTTPException(status_code=404, detail="Notification not found") 

166 

167 # Check permission 

168 if notification.recipient_id and notification.recipient_id != current_user.id: 

169 if current_user.role.value != "superadmin": 

170 raise HTTPException(status_code=403, detail="Not authorized") 

171 

172 if update_data.is_read is not None: 

173 notification.is_read = update_data.is_read 

174 if update_data.is_read: 

175 notification.read_at = datetime.now(UTC) 

176 

177 if update_data.is_dismissed is not None: 

178 notification.is_dismissed = update_data.is_dismissed 

179 

180 db.commit() 

181 db.refresh(notification) 

182 return map_to_schema(notification) 

183 

184@router.put("/{id}/read", response_model=NotificationSchema) 

185def mark_notification_read( 

186 id: str, 

187 db: Session = Depends(get_db), 

188 current_user: User = Depends(get_current_user) 

189): 

190 """Mark notification as read.""" 

191 return update_notification(id, NotificationUpdate(is_read=True), db, current_user) 

192 

193@router.delete("/{id}") 

194def delete_notification( 

195 id: str, 

196 db: Session = Depends(get_db), 

197 current_user: User = Depends(get_current_user) 

198): 

199 """Delete a notification.""" 

200 notification = db.query(Notification).filter(Notification.id == id).first() 

201 if not notification: 

202 raise HTTPException(status_code=404, detail="Notification not found") 

203 

204 # Check permission 

205 if notification.recipient_id and notification.recipient_id != current_user.id: 

206 if current_user.role.value != "superadmin": 

207 raise HTTPException(status_code=403, detail="Not authorized") 

208 

209 db.delete(notification) 

210 db.commit() 

211 return {"status": "success"}