174 lines
6.9 KiB
Python
174 lines
6.9 KiB
Python
"""Notification services implementing the business logic"""
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Optional, List, Dict
|
|
from domain.notifications import Notification
|
|
from application.ports.notification_repository import NotificationRepository
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CreateNotification:
|
|
"""Use case: Create a new notification"""
|
|
|
|
def __init__(self, notification_repo: NotificationRepository):
|
|
self.notification_repo = notification_repo
|
|
|
|
def execute(self, id_usuario: int, tipo_notificacion: str, titulo: str,
|
|
mensaje: str, id_reporte: Optional[str] = None,
|
|
estado_reporte: Optional[str] = None) -> Dict:
|
|
"""
|
|
Creates a new notification
|
|
|
|
Args:
|
|
id_usuario: User ID who receives the notification
|
|
tipo_notificacion: Type of notification
|
|
titulo: Notification title
|
|
mensaje: Notification message
|
|
id_reporte: Optional report ID related to notification
|
|
estado_reporte: Optional report status
|
|
|
|
Returns:
|
|
Dictionary with status and message
|
|
"""
|
|
try:
|
|
# Validate inputs
|
|
if not id_usuario or id_usuario <= 0:
|
|
return {"status": "error", "message": "Usuario inválido"}
|
|
if not tipo_notificacion or not titulo or not mensaje:
|
|
return {"status": "error", "message": "Campos obligatorios faltantes"}
|
|
|
|
# Create notification domain object
|
|
notification = Notification(
|
|
id_notificacion=str(uuid.uuid4()),
|
|
id_usuario=id_usuario,
|
|
tipo_notificacion=tipo_notificacion,
|
|
titulo=titulo,
|
|
mensaje=mensaje,
|
|
id_reporte=id_reporte,
|
|
estado_reporte=estado_reporte,
|
|
leida=False,
|
|
fecha_creacion=datetime.utcnow()
|
|
)
|
|
|
|
# Save to repository
|
|
saved_notification = self.notification_repo.save(notification)
|
|
logger.info(f"Notification created: {saved_notification.id_notificacion}")
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": "Notificación creada exitosamente",
|
|
"id_notificacion": saved_notification.id_notificacion
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error creating notification: {e}", exc_info=True)
|
|
return {"status": "error", "message": f"Error al crear notificación: {str(e)}"}
|
|
|
|
|
|
class GetNotificationById:
|
|
"""Use case: Get a notification by ID"""
|
|
|
|
def __init__(self, notification_repo: NotificationRepository):
|
|
self.notification_repo = notification_repo
|
|
|
|
def execute(self, notification_id: str) -> Optional[Notification]:
|
|
"""Retrieves a notification by ID"""
|
|
return self.notification_repo.find_by_id(notification_id)
|
|
|
|
|
|
class GetUserNotifications:
|
|
"""Use case: Get all notifications for a user"""
|
|
|
|
def __init__(self, notification_repo: NotificationRepository):
|
|
self.notification_repo = notification_repo
|
|
|
|
def execute(self, user_id: int) -> List[Notification]:
|
|
"""Retrieves all notifications for a user"""
|
|
return self.notification_repo.find_by_user_id(user_id)
|
|
|
|
|
|
class GetUnreadUserNotifications:
|
|
"""Use case: Get unread notifications for a user"""
|
|
|
|
def __init__(self, notification_repo: NotificationRepository):
|
|
self.notification_repo = notification_repo
|
|
|
|
def execute(self, user_id: int) -> List[Notification]:
|
|
"""Retrieves unread notifications for a user"""
|
|
return self.notification_repo.find_unread_by_user_id(user_id)
|
|
|
|
|
|
class MarkNotificationAsRead:
|
|
"""Use case: Mark a notification as read"""
|
|
|
|
def __init__(self, notification_repo: NotificationRepository):
|
|
self.notification_repo = notification_repo
|
|
|
|
def execute(self, notification_id: str) -> Dict:
|
|
"""Marks a notification as read"""
|
|
try:
|
|
success = self.notification_repo.mark_as_read(notification_id)
|
|
if success:
|
|
logger.info(f"Notification marked as read: {notification_id}")
|
|
return {"status": "success", "message": "Notificación marcada como leída"}
|
|
else:
|
|
return {"status": "error", "message": "Notificación no encontrada"}
|
|
except Exception as e:
|
|
logger.error(f"Error marking notification as read: {e}", exc_info=True)
|
|
return {"status": "error", "message": f"Error: {str(e)}"}
|
|
|
|
|
|
class MarkAllUserNotificationsAsRead:
|
|
"""Use case: Mark all notifications for a user as read"""
|
|
|
|
def __init__(self, notification_repo: NotificationRepository):
|
|
self.notification_repo = notification_repo
|
|
|
|
def execute(self, user_id: int) -> Dict:
|
|
"""Marks all notifications for a user as read"""
|
|
try:
|
|
success = self.notification_repo.mark_all_as_read(user_id)
|
|
logger.info(f"All notifications marked as read for user: {user_id}")
|
|
return {"status": "success", "message": "Todas las notificaciones marcadas como leídas"}
|
|
except Exception as e:
|
|
logger.error(f"Error marking notifications as read: {e}", exc_info=True)
|
|
return {"status": "error", "message": f"Error: {str(e)}"}
|
|
|
|
|
|
class DeleteNotification:
|
|
"""Use case: Delete a notification"""
|
|
|
|
def __init__(self, notification_repo: NotificationRepository):
|
|
self.notification_repo = notification_repo
|
|
|
|
def execute(self, notification_id: str) -> Dict:
|
|
"""Deletes a notification"""
|
|
try:
|
|
success = self.notification_repo.delete(notification_id)
|
|
if success:
|
|
logger.info(f"Notification deleted: {notification_id}")
|
|
return {"status": "success", "message": "Notificación eliminada"}
|
|
else:
|
|
return {"status": "error", "message": "Notificación no encontrada"}
|
|
except Exception as e:
|
|
logger.error(f"Error deleting notification: {e}", exc_info=True)
|
|
return {"status": "error", "message": f"Error: {str(e)}"}
|
|
|
|
|
|
class DeleteAllUserNotifications:
|
|
"""Use case: Delete all notifications for a user"""
|
|
|
|
def __init__(self, notification_repo: NotificationRepository):
|
|
self.notification_repo = notification_repo
|
|
|
|
def execute(self, user_id: int) -> Dict:
|
|
"""Deletes all notifications for a user"""
|
|
try:
|
|
success = self.notification_repo.delete_all_by_user(user_id)
|
|
logger.info(f"All notifications deleted for user: {user_id}")
|
|
return {"status": "success", "message": "Todas las notificaciones eliminadas"}
|
|
except Exception as e:
|
|
logger.error(f"Error deleting notifications: {e}", exc_info=True)
|
|
return {"status": "error", "message": f"Error: {str(e)}"}
|