añadido los estados de reporte
This commit is contained in:
@@ -1 +1 @@
|
||||
"""RabbitMQ Consumer implementations"""
|
||||
"""RabbitMQ Consumer implementations"""
|
||||
|
||||
@@ -1,220 +1,220 @@
|
||||
"""Report RabbitMQ Consumer - Processes report events and saves to database"""
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from shutil import move
|
||||
|
||||
# Add src to path to import modules
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
from infrastructure.adapters.rabbitmq.consumer import RabbitMQConsumer
|
||||
from infrastructure.adapters.rabbitmq.messages import ReportMessage, ReportEventType
|
||||
from infrastructure.adapters.persistence.report_repository_mongo import ReportRepositoryMongo
|
||||
from infrastructure.adapters.persistence.user_repository_sql import UserRepositorySQL
|
||||
from infrastructure.adapters.file_storage import image_storage
|
||||
from domain.reports import Report
|
||||
from core.config import ConfSettings
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ReportConsumer:
|
||||
"""Consumer for report events from RabbitMQ"""
|
||||
|
||||
def __init__(self):
|
||||
self.repo = ReportRepositoryMongo()
|
||||
self.user_repo = UserRepositorySQL()
|
||||
self.consumer = RabbitMQConsumer(queue_name='reports_queue')
|
||||
self.consumer.set_callback(self.process_message)
|
||||
|
||||
def process_message(self, message_dict: dict):
|
||||
"""
|
||||
Processes a report event message from RabbitMQ
|
||||
|
||||
Args:
|
||||
message_dict: Dictionary containing the message data
|
||||
"""
|
||||
try:
|
||||
# Reconstruct the ReportMessage object
|
||||
message = ReportMessage.from_dict(message_dict)
|
||||
|
||||
if message.event_type == ReportEventType.CREATE:
|
||||
self._handle_create_report(message)
|
||||
elif message.event_type == ReportEventType.UPDATE_VISIBILITY:
|
||||
self._handle_update_visibility(message)
|
||||
elif message.event_type == ReportEventType.DELETE:
|
||||
self._handle_delete_report(message)
|
||||
else:
|
||||
logger.warning(f"Unknown event type: {message.event_type}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing report message: {e}", exc_info=True)
|
||||
# Rollback explícito en caso de error
|
||||
self.user_repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_create_report(self, message: ReportMessage):
|
||||
"""Handle report create event con manejo de transacciones cruzadas"""
|
||||
try:
|
||||
logger.info(f"Creating report: {message.id_reporte} from user {message.id_usuario}")
|
||||
|
||||
# Parse datetime string
|
||||
fecha_creacion = datetime.fromisoformat(message.fecha_creacion)
|
||||
|
||||
# Renombrar imagen temporal si existe
|
||||
final_image_filename = None
|
||||
if message.image_filename:
|
||||
try:
|
||||
# Renombrar de temp_userid_type a report_id
|
||||
temp_path = image_storage.get_image_path(message.image_filename)
|
||||
final_filename = f"{message.id_reporte}.webp"
|
||||
final_path = image_storage.get_image_path(final_filename)
|
||||
|
||||
if temp_path.exists():
|
||||
move(str(temp_path), str(final_path))
|
||||
final_image_filename = final_filename
|
||||
logger.info(f"Image renamed from {message.image_filename} to {final_filename}")
|
||||
else:
|
||||
logger.warning(f"Temporary image not found: {message.image_filename}")
|
||||
except Exception as img_error:
|
||||
logger.error(f"Error renaming image: {img_error}", exc_info=True)
|
||||
# Continuar sin imagen si falla el renombramiento
|
||||
|
||||
# Create Report domain object
|
||||
report = Report(
|
||||
id_reporte=message.id_reporte,
|
||||
id_usuario=message.id_usuario,
|
||||
tipo_reporte=message.tipo_reporte,
|
||||
descripcion=message.descripcion,
|
||||
ubicacion=message.ubicacion,
|
||||
lat=message.lat,
|
||||
lng=message.lng,
|
||||
image_filename=final_image_filename,
|
||||
visibilidad=message.visibilidad,
|
||||
fecha_creacion=fecha_creacion
|
||||
)
|
||||
|
||||
try:
|
||||
# Save to MongoDB repository
|
||||
saved_report = self.repo.save(report)
|
||||
logger.info(f"Report created successfully in MongoDB: {message.id_reporte}")
|
||||
except Exception as mongo_error:
|
||||
logger.error(f"Error saving report to MongoDB: {mongo_error}", exc_info=True)
|
||||
raise
|
||||
|
||||
try:
|
||||
# Increment user's report counter in MySQL
|
||||
self.user_repo.increment_reports(message.id_usuario)
|
||||
logger.info(f"Incremented report counter for user: {message.id_usuario}")
|
||||
except Exception as sql_error:
|
||||
logger.error(f"Error incrementing report counter: {sql_error}", exc_info=True)
|
||||
# Rollback SQL transaction
|
||||
self.user_repo.db.rollback()
|
||||
# Note: MongoDB save cannot be rolled back, log for manual cleanup
|
||||
logger.critical(f"INCONSISTENCY: Report {message.id_reporte} saved to MongoDB but user counter not incremented for user {message.id_usuario}")
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating report: {e}", exc_info=True)
|
||||
self.user_repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_update_visibility(self, message: ReportMessage):
|
||||
"""Handle report visibility update event con manejo de transacciones"""
|
||||
try:
|
||||
logger.info(f"Updating visibility for report: {message.id_reporte}")
|
||||
|
||||
# Find the report
|
||||
report = self.repo.find_by_id(message.id_reporte)
|
||||
if not report:
|
||||
logger.warning(f"Report not found: {message.id_reporte}")
|
||||
return
|
||||
|
||||
# Update visibility in MongoDB
|
||||
try:
|
||||
self.repo.update_visibility(message.id_reporte, message.visibilidad)
|
||||
logger.info(f"Report visibility updated: {message.id_reporte} -> {message.visibilidad}")
|
||||
except Exception as mongo_error:
|
||||
logger.error(f"Error updating report visibility in MongoDB: {mongo_error}", exc_info=True)
|
||||
raise
|
||||
|
||||
# Penalize author if visibility is very low (shadowban)
|
||||
if message.penalize_author and message.visibilidad < 20:
|
||||
try:
|
||||
user = self.user_repo.find_by_id(report.id_usuario)
|
||||
if user:
|
||||
# Reduce user's rating
|
||||
new_rating = max(0, user.calificacion - 5)
|
||||
self.user_repo.update_rating(report.id_usuario, new_rating)
|
||||
logger.info(f"Author penalized: user {report.id_usuario} rating reduced to {new_rating}")
|
||||
else:
|
||||
logger.warning(f"User not found for penalty: {report.id_usuario}")
|
||||
except Exception as penalty_error:
|
||||
logger.error(f"Error penalizing author: {penalty_error}", exc_info=True)
|
||||
self.user_repo.db.rollback()
|
||||
logger.critical(f"INCONSISTENCY: Report {message.id_reporte} visibility updated but author penalty failed")
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating report visibility: {e}", exc_info=True)
|
||||
self.user_repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_delete_report(self, message: ReportMessage):
|
||||
"""Handle report delete event"""
|
||||
try:
|
||||
logger.info(f"Deleting report: {message.id_reporte}")
|
||||
|
||||
# Obtener reportepara acceder a image_filename antes de eliminarlo
|
||||
report = self.repo.find_by_id(message.id_reporte)
|
||||
|
||||
# Eliminar del MongoDB
|
||||
success = self.repo.delete(message.id_reporte)
|
||||
if success:
|
||||
logger.info(f"Report deleted successfully from MongoDB: {message.id_reporte}")
|
||||
|
||||
# Eliminar imagen del almacenamiento
|
||||
if report and report.image_filename:
|
||||
deleted = image_storage.delete_image(report.image_filename)
|
||||
if deleted:
|
||||
logger.info(f"Image deleted: {report.image_filename}")
|
||||
else:
|
||||
logger.warning(f"Could not delete image: {report.image_filename}")
|
||||
else:
|
||||
logger.warning(f"Failed to delete report: {message.id_reporte}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting report: {e}", exc_info=True)
|
||||
self.user_repo.db.rollback()
|
||||
raise
|
||||
|
||||
def start(self):
|
||||
"""Start consuming messages"""
|
||||
logger.info("Starting Report Consumer...")
|
||||
logger.info("[*] Waiting for report events. Ctrl+C to exit.")
|
||||
try:
|
||||
self.consumer.start_consuming()
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Report Consumer stopped by user")
|
||||
except Exception as e:
|
||||
logger.error(f"Consumer error: {e}", exc_info=True)
|
||||
raise
|
||||
finally:
|
||||
# Asegurar cierre de sesión SQL
|
||||
if self.user_repo.db:
|
||||
try:
|
||||
self.user_repo.db.close()
|
||||
except Exception as e:
|
||||
logger.error(f"Error closing database session: {e}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
consumer = ReportConsumer()
|
||||
consumer.start()
|
||||
"""Report RabbitMQ Consumer - Processes report events and saves to database"""
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from shutil import move
|
||||
|
||||
# Add src to path to import modules
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
from infrastructure.adapters.rabbitmq.consumer import RabbitMQConsumer
|
||||
from infrastructure.adapters.rabbitmq.messages import ReportMessage, ReportEventType
|
||||
from infrastructure.adapters.persistence.report_repository_mongo import ReportRepositoryMongo
|
||||
from infrastructure.adapters.persistence.user_repository_sql import UserRepositorySQL
|
||||
from infrastructure.adapters.file_storage import image_storage
|
||||
from domain.reports import Report
|
||||
from core.config import ConfSettings
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ReportConsumer:
|
||||
"""Consumer for report events from RabbitMQ"""
|
||||
|
||||
def __init__(self):
|
||||
self.repo = ReportRepositoryMongo()
|
||||
self.user_repo = UserRepositorySQL()
|
||||
self.consumer = RabbitMQConsumer(queue_name='reports_queue')
|
||||
self.consumer.set_callback(self.process_message)
|
||||
|
||||
def process_message(self, message_dict: dict):
|
||||
"""
|
||||
Processes a report event message from RabbitMQ
|
||||
|
||||
Args:
|
||||
message_dict: Dictionary containing the message data
|
||||
"""
|
||||
try:
|
||||
# Reconstruct the ReportMessage object
|
||||
message = ReportMessage.from_dict(message_dict)
|
||||
|
||||
if message.event_type == ReportEventType.CREATE:
|
||||
self._handle_create_report(message)
|
||||
elif message.event_type == ReportEventType.UPDATE_VISIBILITY:
|
||||
self._handle_update_visibility(message)
|
||||
elif message.event_type == ReportEventType.DELETE:
|
||||
self._handle_delete_report(message)
|
||||
else:
|
||||
logger.warning(f"Unknown event type: {message.event_type}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing report message: {e}", exc_info=True)
|
||||
# Rollback explícito en caso de error
|
||||
self.user_repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_create_report(self, message: ReportMessage):
|
||||
"""Handle report create event con manejo de transacciones cruzadas"""
|
||||
try:
|
||||
logger.info(f"Creating report: {message.id_reporte} from user {message.id_usuario}")
|
||||
|
||||
# Parse datetime string
|
||||
fecha_creacion = datetime.fromisoformat(message.fecha_creacion)
|
||||
|
||||
# Renombrar imagen temporal si existe
|
||||
final_image_filename = None
|
||||
if message.image_filename:
|
||||
try:
|
||||
# Renombrar de temp_userid_type a report_id
|
||||
temp_path = image_storage.get_image_path(message.image_filename)
|
||||
final_filename = f"{message.id_reporte}.webp"
|
||||
final_path = image_storage.get_image_path(final_filename)
|
||||
|
||||
if temp_path.exists():
|
||||
move(str(temp_path), str(final_path))
|
||||
final_image_filename = final_filename
|
||||
logger.info(f"Image renamed from {message.image_filename} to {final_filename}")
|
||||
else:
|
||||
logger.warning(f"Temporary image not found: {message.image_filename}")
|
||||
except Exception as img_error:
|
||||
logger.error(f"Error renaming image: {img_error}", exc_info=True)
|
||||
# Continuar sin imagen si falla el renombramiento
|
||||
|
||||
# Create Report domain object
|
||||
report = Report(
|
||||
id_reporte=message.id_reporte,
|
||||
id_usuario=message.id_usuario,
|
||||
tipo_reporte=message.tipo_reporte,
|
||||
descripcion=message.descripcion,
|
||||
ubicacion=message.ubicacion,
|
||||
lat=message.lat,
|
||||
lng=message.lng,
|
||||
image_filename=final_image_filename,
|
||||
visibilidad=message.visibilidad,
|
||||
fecha_creacion=fecha_creacion
|
||||
)
|
||||
|
||||
try:
|
||||
# Save to MongoDB repository
|
||||
saved_report = self.repo.save(report)
|
||||
logger.info(f"Report created successfully in MongoDB: {message.id_reporte}")
|
||||
except Exception as mongo_error:
|
||||
logger.error(f"Error saving report to MongoDB: {mongo_error}", exc_info=True)
|
||||
raise
|
||||
|
||||
try:
|
||||
# Increment user's report counter in MySQL
|
||||
self.user_repo.increment_reports(message.id_usuario)
|
||||
logger.info(f"Incremented report counter for user: {message.id_usuario}")
|
||||
except Exception as sql_error:
|
||||
logger.error(f"Error incrementing report counter: {sql_error}", exc_info=True)
|
||||
# Rollback SQL transaction
|
||||
self.user_repo.db.rollback()
|
||||
# Note: MongoDB save cannot be rolled back, log for manual cleanup
|
||||
logger.critical(f"INCONSISTENCY: Report {message.id_reporte} saved to MongoDB but user counter not incremented for user {message.id_usuario}")
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating report: {e}", exc_info=True)
|
||||
self.user_repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_update_visibility(self, message: ReportMessage):
|
||||
"""Handle report visibility update event con manejo de transacciones"""
|
||||
try:
|
||||
logger.info(f"Updating visibility for report: {message.id_reporte}")
|
||||
|
||||
# Find the report
|
||||
report = self.repo.find_by_id(message.id_reporte)
|
||||
if not report:
|
||||
logger.warning(f"Report not found: {message.id_reporte}")
|
||||
return
|
||||
|
||||
# Update visibility in MongoDB
|
||||
try:
|
||||
self.repo.update_visibility(message.id_reporte, message.visibilidad)
|
||||
logger.info(f"Report visibility updated: {message.id_reporte} -> {message.visibilidad}")
|
||||
except Exception as mongo_error:
|
||||
logger.error(f"Error updating report visibility in MongoDB: {mongo_error}", exc_info=True)
|
||||
raise
|
||||
|
||||
# Penalize author if visibility is very low (shadowban)
|
||||
if message.penalize_author and message.visibilidad < 20:
|
||||
try:
|
||||
user = self.user_repo.find_by_id(report.id_usuario)
|
||||
if user:
|
||||
# Reduce user's rating
|
||||
new_rating = max(0, user.calificacion - 5)
|
||||
self.user_repo.update_rating(report.id_usuario, new_rating)
|
||||
logger.info(f"Author penalized: user {report.id_usuario} rating reduced to {new_rating}")
|
||||
else:
|
||||
logger.warning(f"User not found for penalty: {report.id_usuario}")
|
||||
except Exception as penalty_error:
|
||||
logger.error(f"Error penalizing author: {penalty_error}", exc_info=True)
|
||||
self.user_repo.db.rollback()
|
||||
logger.critical(f"INCONSISTENCY: Report {message.id_reporte} visibility updated but author penalty failed")
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating report visibility: {e}", exc_info=True)
|
||||
self.user_repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_delete_report(self, message: ReportMessage):
|
||||
"""Handle report delete event"""
|
||||
try:
|
||||
logger.info(f"Deleting report: {message.id_reporte}")
|
||||
|
||||
# Obtener reportepara acceder a image_filename antes de eliminarlo
|
||||
report = self.repo.find_by_id(message.id_reporte)
|
||||
|
||||
# Eliminar del MongoDB
|
||||
success = self.repo.delete(message.id_reporte)
|
||||
if success:
|
||||
logger.info(f"Report deleted successfully from MongoDB: {message.id_reporte}")
|
||||
|
||||
# Eliminar imagen del almacenamiento
|
||||
if report and report.image_filename:
|
||||
deleted = image_storage.delete_image(report.image_filename)
|
||||
if deleted:
|
||||
logger.info(f"Image deleted: {report.image_filename}")
|
||||
else:
|
||||
logger.warning(f"Could not delete image: {report.image_filename}")
|
||||
else:
|
||||
logger.warning(f"Failed to delete report: {message.id_reporte}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting report: {e}", exc_info=True)
|
||||
self.user_repo.db.rollback()
|
||||
raise
|
||||
|
||||
def start(self):
|
||||
"""Start consuming messages"""
|
||||
logger.info("Starting Report Consumer...")
|
||||
logger.info("[*] Waiting for report events. Ctrl+C to exit.")
|
||||
try:
|
||||
self.consumer.start_consuming()
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Report Consumer stopped by user")
|
||||
except Exception as e:
|
||||
logger.error(f"Consumer error: {e}", exc_info=True)
|
||||
raise
|
||||
finally:
|
||||
# Asegurar cierre de sesión SQL
|
||||
if self.user_repo.db:
|
||||
try:
|
||||
self.user_repo.db.close()
|
||||
except Exception as e:
|
||||
logger.error(f"Error closing database session: {e}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
consumer = ReportConsumer()
|
||||
consumer.start()
|
||||
|
||||
@@ -1,157 +1,157 @@
|
||||
"""User RabbitMQ Consumer - Processes user events and saves to database"""
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
|
||||
# Add src to path to import modules
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
from infrastructure.adapters.rabbitmq.consumer import RabbitMQConsumer
|
||||
from infrastructure.adapters.rabbitmq.messages import UserMessage, UserEventType
|
||||
from infrastructure.adapters.persistence.user_repository_sql import UserRepositorySQL
|
||||
from domain.users import User
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class UserConsumer:
|
||||
"""Consumer for user events from RabbitMQ"""
|
||||
|
||||
def __init__(self):
|
||||
self.repo = UserRepositorySQL()
|
||||
self.consumer = RabbitMQConsumer(queue_name='users_queue')
|
||||
self.consumer.set_callback(self.process_message)
|
||||
|
||||
def process_message(self, message_dict: dict):
|
||||
"""
|
||||
Processes a user event message from RabbitMQ
|
||||
|
||||
Args:
|
||||
message_dict: Dictionary containing the message data
|
||||
"""
|
||||
try:
|
||||
# Reconstruct the UserMessage object
|
||||
message = UserMessage.from_dict(message_dict)
|
||||
|
||||
if message.event_type == UserEventType.CREATE:
|
||||
self._handle_create_user(message)
|
||||
elif message.event_type == UserEventType.UPDATE:
|
||||
self._handle_update_user(message)
|
||||
elif message.event_type == UserEventType.DELETE:
|
||||
self._handle_delete_user(message)
|
||||
else:
|
||||
logger.warning(f"Unknown event type: {message.event_type}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing user message: {e}", exc_info=True)
|
||||
# Rollback en caso de error en el procesamiento del mensaje
|
||||
self.repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_create_user(self, message: UserMessage):
|
||||
"""Handle user create event"""
|
||||
try:
|
||||
logger.info(f"Creating user: {message.email}")
|
||||
|
||||
# Parse datetime strings
|
||||
fecha_nacimiento = datetime.fromisoformat(message.fecha_nacimiento)
|
||||
fecha_creacion = datetime.fromisoformat(message.fecha_creacion)
|
||||
|
||||
# Create User domain object
|
||||
user = User(
|
||||
user_id=0, # Will be auto-generated by DB
|
||||
nombre=message.nombre,
|
||||
apellido=message.apellido,
|
||||
email=message.email,
|
||||
fecha_nacimiento=fecha_nacimiento,
|
||||
fecha_creacion=fecha_creacion,
|
||||
calificacion=message.calificacion,
|
||||
numero_reportes=message.numero_reportes,
|
||||
url_foto_perfil=message.url_foto_perfil,
|
||||
biografia=message.biografia
|
||||
)
|
||||
|
||||
# Save to repository with transaction handling
|
||||
saved_user = self.repo.save(user)
|
||||
logger.info(f"User created successfully: {saved_user.user_id} - {saved_user.email}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating user: {e}", exc_info=True)
|
||||
self.repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_update_user(self, message: UserMessage):
|
||||
"""Handle user update event"""
|
||||
try:
|
||||
logger.info(f"Updating user: {message.user_id}")
|
||||
|
||||
# Find the user
|
||||
user = self.repo.find_by_id(message.user_id)
|
||||
if not user:
|
||||
logger.warning(f"User not found: {message.user_id}")
|
||||
return
|
||||
|
||||
# Update fields if provided
|
||||
if message.nombre:
|
||||
user.nombre = message.nombre
|
||||
if message.apellido:
|
||||
user.apellido = message.apellido
|
||||
if message.url_foto_perfil is not None:
|
||||
user.url_foto_perfil = message.url_foto_perfil
|
||||
if message.biografia is not None:
|
||||
user.biografia = message.biografia
|
||||
|
||||
# Save to repository with transaction handling
|
||||
updated_user = self.repo.update(user)
|
||||
logger.info(f"User updated successfully: {message.user_id}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating user: {e}", exc_info=True)
|
||||
self.repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_delete_user(self, message: UserMessage):
|
||||
"""Handle user delete event"""
|
||||
try:
|
||||
logger.info(f"Deleting user: {message.user_id}")
|
||||
|
||||
success = self.repo.delete(message.user_id)
|
||||
if success:
|
||||
logger.info(f"User deleted successfully: {message.user_id}")
|
||||
else:
|
||||
logger.warning(f"Failed to delete user: {message.user_id}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting user: {e}", exc_info=True)
|
||||
self.repo.db.rollback()
|
||||
raise
|
||||
|
||||
def start(self):
|
||||
"""Start consuming messages"""
|
||||
logger.info("Starting User Consumer...")
|
||||
logger.info("[*] Waiting for user events. Ctrl+C to exit.")
|
||||
try:
|
||||
self.consumer.start_consuming()
|
||||
except KeyboardInterrupt:
|
||||
logger.info("User Consumer stopped by user")
|
||||
except Exception as e:
|
||||
logger.error(f"Consumer error: {e}", exc_info=True)
|
||||
raise
|
||||
finally:
|
||||
# Asegurar cierre de sesión
|
||||
if self.repo.db:
|
||||
try:
|
||||
self.repo.db.close()
|
||||
except Exception as e:
|
||||
logger.error(f"Error closing database session: {e}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
consumer = UserConsumer()
|
||||
consumer.start()
|
||||
"""User RabbitMQ Consumer - Processes user events and saves to database"""
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime
|
||||
|
||||
# Add src to path to import modules
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
|
||||
from infrastructure.adapters.rabbitmq.consumer import RabbitMQConsumer
|
||||
from infrastructure.adapters.rabbitmq.messages import UserMessage, UserEventType
|
||||
from infrastructure.adapters.persistence.user_repository_sql import UserRepositorySQL
|
||||
from domain.users import User
|
||||
|
||||
# Set up logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class UserConsumer:
|
||||
"""Consumer for user events from RabbitMQ"""
|
||||
|
||||
def __init__(self):
|
||||
self.repo = UserRepositorySQL()
|
||||
self.consumer = RabbitMQConsumer(queue_name='users_queue')
|
||||
self.consumer.set_callback(self.process_message)
|
||||
|
||||
def process_message(self, message_dict: dict):
|
||||
"""
|
||||
Processes a user event message from RabbitMQ
|
||||
|
||||
Args:
|
||||
message_dict: Dictionary containing the message data
|
||||
"""
|
||||
try:
|
||||
# Reconstruct the UserMessage object
|
||||
message = UserMessage.from_dict(message_dict)
|
||||
|
||||
if message.event_type == UserEventType.CREATE:
|
||||
self._handle_create_user(message)
|
||||
elif message.event_type == UserEventType.UPDATE:
|
||||
self._handle_update_user(message)
|
||||
elif message.event_type == UserEventType.DELETE:
|
||||
self._handle_delete_user(message)
|
||||
else:
|
||||
logger.warning(f"Unknown event type: {message.event_type}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing user message: {e}", exc_info=True)
|
||||
# Rollback en caso de error en el procesamiento del mensaje
|
||||
self.repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_create_user(self, message: UserMessage):
|
||||
"""Handle user create event"""
|
||||
try:
|
||||
logger.info(f"Creating user: {message.email}")
|
||||
|
||||
# Parse datetime strings
|
||||
fecha_nacimiento = datetime.fromisoformat(message.fecha_nacimiento)
|
||||
fecha_creacion = datetime.fromisoformat(message.fecha_creacion)
|
||||
|
||||
# Create User domain object
|
||||
user = User(
|
||||
user_id=0, # Will be auto-generated by DB
|
||||
nombre=message.nombre,
|
||||
apellido=message.apellido,
|
||||
email=message.email,
|
||||
fecha_nacimiento=fecha_nacimiento,
|
||||
fecha_creacion=fecha_creacion,
|
||||
calificacion=message.calificacion,
|
||||
numero_reportes=message.numero_reportes,
|
||||
url_foto_perfil=message.url_foto_perfil,
|
||||
biografia=message.biografia
|
||||
)
|
||||
|
||||
# Save to repository with transaction handling
|
||||
saved_user = self.repo.save(user)
|
||||
logger.info(f"User created successfully: {saved_user.user_id} - {saved_user.email}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating user: {e}", exc_info=True)
|
||||
self.repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_update_user(self, message: UserMessage):
|
||||
"""Handle user update event"""
|
||||
try:
|
||||
logger.info(f"Updating user: {message.user_id}")
|
||||
|
||||
# Find the user
|
||||
user = self.repo.find_by_id(message.user_id)
|
||||
if not user:
|
||||
logger.warning(f"User not found: {message.user_id}")
|
||||
return
|
||||
|
||||
# Update fields if provided
|
||||
if message.nombre:
|
||||
user.nombre = message.nombre
|
||||
if message.apellido:
|
||||
user.apellido = message.apellido
|
||||
if message.url_foto_perfil is not None:
|
||||
user.url_foto_perfil = message.url_foto_perfil
|
||||
if message.biografia is not None:
|
||||
user.biografia = message.biografia
|
||||
|
||||
# Save to repository with transaction handling
|
||||
updated_user = self.repo.update(user)
|
||||
logger.info(f"User updated successfully: {message.user_id}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating user: {e}", exc_info=True)
|
||||
self.repo.db.rollback()
|
||||
raise
|
||||
|
||||
def _handle_delete_user(self, message: UserMessage):
|
||||
"""Handle user delete event"""
|
||||
try:
|
||||
logger.info(f"Deleting user: {message.user_id}")
|
||||
|
||||
success = self.repo.delete(message.user_id)
|
||||
if success:
|
||||
logger.info(f"User deleted successfully: {message.user_id}")
|
||||
else:
|
||||
logger.warning(f"Failed to delete user: {message.user_id}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting user: {e}", exc_info=True)
|
||||
self.repo.db.rollback()
|
||||
raise
|
||||
|
||||
def start(self):
|
||||
"""Start consuming messages"""
|
||||
logger.info("Starting User Consumer...")
|
||||
logger.info("[*] Waiting for user events. Ctrl+C to exit.")
|
||||
try:
|
||||
self.consumer.start_consuming()
|
||||
except KeyboardInterrupt:
|
||||
logger.info("User Consumer stopped by user")
|
||||
except Exception as e:
|
||||
logger.error(f"Consumer error: {e}", exc_info=True)
|
||||
raise
|
||||
finally:
|
||||
# Asegurar cierre de sesión
|
||||
if self.repo.db:
|
||||
try:
|
||||
self.repo.db.close()
|
||||
except Exception as e:
|
||||
logger.error(f"Error closing database session: {e}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
consumer = UserConsumer()
|
||||
consumer.start()
|
||||
|
||||
Reference in New Issue
Block a user