104 lines
4.3 KiB
Python
104 lines
4.3 KiB
Python
"""Notification Repository Implementation using MongoDB"""
|
|
from application.ports.notification_repository import NotificationRepository
|
|
from domain.notifications import Notification
|
|
from infrastructure.adapters.persistence.mongodb import get_notifications_collection
|
|
from typing import List, Optional
|
|
from datetime import datetime
|
|
import uuid
|
|
|
|
|
|
class NotificationRepositoryMongo(NotificationRepository):
|
|
"""Implementación del repositorio de Notificaciones usando MongoDB"""
|
|
|
|
def __init__(self):
|
|
self.collection = get_notifications_collection()
|
|
|
|
def save(self, notification: Notification) -> Notification:
|
|
"""Guarda una nueva notificación"""
|
|
notification_dict = {
|
|
"id_notificacion": notification.id_notificacion,
|
|
"id_usuario": notification.id_usuario,
|
|
"tipo_notificacion": notification.tipo_notificacion,
|
|
"titulo": notification.titulo,
|
|
"mensaje": notification.mensaje,
|
|
"id_reporte": notification.id_reporte,
|
|
"estado_reporte": notification.estado_reporte,
|
|
"leida": notification.leida,
|
|
"fecha_creacion": notification.fecha_creacion or datetime.utcnow(),
|
|
"fecha_lectura": notification.fecha_lectura
|
|
}
|
|
result = self.collection.insert_one(notification_dict)
|
|
return notification
|
|
|
|
def find_by_id(self, notification_id: str) -> Optional[Notification]:
|
|
"""Obtiene una notificación por ID"""
|
|
doc = self.collection.find_one({"id_notificacion": notification_id})
|
|
if doc:
|
|
return self._to_domain(doc)
|
|
return None
|
|
|
|
def find_by_user_id(self, user_id: int) -> List[Notification]:
|
|
"""Obtiene todas las notificaciones de un usuario, ordenadas por fecha descendente"""
|
|
docs = self.collection.find({"id_usuario": user_id}).sort("fecha_creacion", -1)
|
|
return [self._to_domain(doc) for doc in docs]
|
|
|
|
def find_unread_by_user_id(self, user_id: int) -> List[Notification]:
|
|
"""Obtiene todas las notificaciones no leídas de un usuario"""
|
|
docs = self.collection.find({
|
|
"id_usuario": user_id,
|
|
"leida": False
|
|
}).sort("fecha_creacion", -1)
|
|
return [self._to_domain(doc) for doc in docs]
|
|
|
|
def find_all(self) -> List[Notification]:
|
|
"""Obtiene todas las notificaciones"""
|
|
docs = self.collection.find().sort("fecha_creacion", -1)
|
|
return [self._to_domain(doc) for doc in docs]
|
|
|
|
def mark_as_read(self, notification_id: str) -> bool:
|
|
"""Marca una notificación como leída"""
|
|
result = self.collection.update_one(
|
|
{"id_notificacion": notification_id},
|
|
{"$set": {
|
|
"leida": True,
|
|
"fecha_lectura": datetime.utcnow()
|
|
}}
|
|
)
|
|
return result.modified_count > 0
|
|
|
|
def mark_all_as_read(self, user_id: int) -> bool:
|
|
"""Marca todas las notificaciones de un usuario como leídas"""
|
|
result = self.collection.update_many(
|
|
{"id_usuario": user_id, "leida": False},
|
|
{"$set": {
|
|
"leida": True,
|
|
"fecha_lectura": datetime.utcnow()
|
|
}}
|
|
)
|
|
return result.modified_count > 0
|
|
|
|
def delete(self, notification_id: str) -> bool:
|
|
"""Elimina una notificación"""
|
|
result = self.collection.delete_one({"id_notificacion": notification_id})
|
|
return result.deleted_count > 0
|
|
|
|
def delete_all_by_user(self, user_id: int) -> bool:
|
|
"""Elimina todas las notificaciones de un usuario"""
|
|
result = self.collection.delete_many({"id_usuario": user_id})
|
|
return result.deleted_count > 0
|
|
|
|
def _to_domain(self, doc: dict) -> Notification:
|
|
"""Convierte un documento de MongoDB a un objeto de dominio"""
|
|
return Notification(
|
|
id_notificacion=doc.get("id_notificacion"),
|
|
id_usuario=doc.get("id_usuario"),
|
|
tipo_notificacion=doc.get("tipo_notificacion"),
|
|
titulo=doc.get("titulo"),
|
|
mensaje=doc.get("mensaje"),
|
|
id_reporte=doc.get("id_reporte"),
|
|
estado_reporte=doc.get("estado_reporte"),
|
|
leida=doc.get("leida", False),
|
|
fecha_creacion=doc.get("fecha_creacion"),
|
|
fecha_lectura=doc.get("fecha_lectura")
|
|
)
|