91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
from flask import request
|
|
from models import Notif, NotifType, db
|
|
from typing import Optional, Dict, Any, List
|
|
from datetime import datetime, timedelta
|
|
from flask_login import current_user
|
|
from sqlalchemy import desc
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def create_notification(
|
|
notif_type: str,
|
|
user_id: int,
|
|
sender_id: Optional[int] = None,
|
|
details: Optional[Dict[str, Any]] = None
|
|
) -> Notif:
|
|
"""
|
|
Create a notification in the database.
|
|
|
|
Args:
|
|
notif_type: The type of notification (must match NotifType enum)
|
|
user_id: The ID of the user to notify
|
|
sender_id: Optional ID of the user who triggered the notification
|
|
details: Optional dictionary containing notification details
|
|
|
|
Returns:
|
|
The created Notif object
|
|
"""
|
|
logger.debug(f"Creating notification of type: {notif_type}")
|
|
logger.debug(f"Notification details: {details}")
|
|
|
|
try:
|
|
notif = Notif(
|
|
notif_type=notif_type,
|
|
user_id=user_id,
|
|
sender_id=sender_id,
|
|
timestamp=datetime.utcnow(),
|
|
details=details or {},
|
|
read=False
|
|
)
|
|
|
|
logger.debug(f"Created notification object: {notif}")
|
|
db.session.add(notif)
|
|
# Don't commit here - let the caller handle the transaction
|
|
logger.debug("Notification object added to session")
|
|
return notif
|
|
except Exception as e:
|
|
logger.error(f"Error creating notification: {str(e)}")
|
|
raise
|
|
|
|
def get_user_notifications(user_id: int, limit: int = 50, unread_only: bool = False) -> List[Notif]:
|
|
"""Get recent notifications for a specific user"""
|
|
query = Notif.query.filter_by(user_id=user_id)
|
|
if unread_only:
|
|
query = query.filter_by(read=False)
|
|
return query.order_by(desc(Notif.timestamp)).limit(limit).all()
|
|
|
|
def mark_notification_read(notif_id: int) -> bool:
|
|
"""Mark a notification as read"""
|
|
notif = Notif.query.get(notif_id)
|
|
if notif:
|
|
notif.read = True
|
|
db.session.commit()
|
|
return True
|
|
return False
|
|
|
|
def mark_all_notifications_read(user_id: int) -> int:
|
|
"""Mark all notifications as read for a user"""
|
|
result = Notif.query.filter_by(user_id=user_id, read=False).update({'read': True})
|
|
db.session.commit()
|
|
return result
|
|
|
|
def get_unread_count(user_id: int) -> int:
|
|
"""Get count of unread notifications for a user"""
|
|
return Notif.query.filter_by(user_id=user_id, read=False).count()
|
|
|
|
def delete_notification(notif_id: int) -> bool:
|
|
"""Delete a notification"""
|
|
notif = Notif.query.get(notif_id)
|
|
if notif:
|
|
db.session.delete(notif)
|
|
db.session.commit()
|
|
return True
|
|
return False
|
|
|
|
def delete_old_notifications(days: int = 30) -> int:
|
|
"""Delete notifications older than specified days"""
|
|
cutoff_date = datetime.utcnow() - timedelta(days=days)
|
|
result = Notif.query.filter(Notif.timestamp < cutoff_date).delete()
|
|
db.session.commit()
|
|
return result |