Ajout de l'intégration RabbitMQ

Author: serge-eric-kalaga

.env.exemple

@@ -1,8 +1,15 @@
+# Email configuration
+# Email domain (e.g., gmail.com)
 EMAIL_DOMAIN=
+# SMTP server host
 EMAIL_HOST=
+# SMTP server port
 EMAIL_PORT=
+# SMTP username
 EMAIL_USERNAME=
+# SMTP password
 EMAIL_PASSWORD=
+# From email address (e.g., noreply@example.com)
 FROM_EMAIL=
 
 # Kafka configuration
@@ -15,10 +22,43 @@ KAFKA_CONSUMER_TOPIC=email_topic
 # Key for Kafka messages
 KAFKA_MESSAGE_KEY=
 
-# Example values:
+# RabbitMQ configuration
+# Set to True to enable RabbitMQ integration
+USE_RABBITMQ=
+# RabbitMQ connection URL (e.g., amqp://user:password@host:port/vhost)
+RABBITMQ_URL=
+# RabbitMQ exchange name
+RABBITMQ_EXCHANGE=email_exchange
+# RabbitMQ routing key (optional)
+RABBITMQ_ROUTING_KEY=email_routing_key
+# RabbitMQ queue name 
+RABBITMQ_QUEUE=email_queue
+# RabbitMQ default username
+RABBITMQ_DEFAULT_USER=
+# RabbitMQ default password
+RABBITMQ_DEFAULT_PASS=
+
+
+
+# ========== Example values: ==========
 # EMAIL_DOMAIN=example.com
 # EMAIL_HOST=smtp.example.com
 # EMAIL_PORT=587
 # EMAIL_USERNAME=user@example.com
 # EMAIL_PASSWORD=yourpassword
-# FROM_EMAIL=noreply@example.com
+# FROM_EMAIL=noreply@example.com
+
+# KAFKA_BOOTSTRAP_SERVERS=broker1:9092,broker2:9092
+# KAFKA_MESSAGE_KEY=email_notification
+# KAFKA_MESSAGE_KEY=email_notification
+# USE_KAFKA=True
+# KAFKA_MESSAGE_KEY=email_notification
+# KAFKA_MESSAGE_KEY=email_notificationemail_notification
+
+# USE_RABBITMQ=True
+# RABBITMQ_URL=amqp://user:password@rabbitmq_host:5672/vhost
+# RABBITMQ_EXCHANGE=email_exchange
+# RABBITMQ_ROUTING_KEY=email_routing_key
+# RABBITMQ_DEFAULT_USER=guest
+# RABBITMQ_DEFAULT_PASS=guest
+# KAFKA_MESSAGE_KEY=email_notification

README.md

@@ -282,33 +282,6 @@ curl http://localhost:9876/api/smtp-status
 }
 ```
 
-### Intégration du test SMTP en Python
-
-```python
-import requests
-
-def check_smtp_status():
-    try:
-        response = requests.get("http://localhost:9876/api/smtp-status")
-        data = response.json()
-        
-        if data["status"]:
-            print("✅ Serveur SMTP accessible")
-            return True
-        else:
-            print("❌ Serveur SMTP non accessible:", data["message"])
-            return False
-    except Exception as e:
-        print(f"Erreur lors de la vérification : {e}")
-        return False
-
-# Utilisation
-if check_smtp_status():
-    # Procéder à l'envoi d'emails
-    send_email("test@example.com", "Test", "Message de test")
-else:
-    print("Impossible d'envoyer des emails pour le moment")
-```
 
 ## 🔧 Gestion des erreurs
 
@@ -412,6 +385,41 @@ services:
     restart: unless-stopped
 ```
 
+## 🐰 Intégration RabbitMQ
+
+Le service peut consommer des messages RabbitMQ pour envoyer automatiquement des emails si l'intégration RabbitMQ est activée via les variables d'environnement (`USE_RABBITMQ`, `RABBITMQ_URL`, `RABBITMQ_EXCHANGE`, `RABBITMQ_ROUTING_KEY`, `RABBITMQ_QUEUE`).
+
+Format JSON attendu pour le message publié sur RabbitMQ :
+
+```json
+{
+  "receiver_email": "user@example.com",
+  "email_object": "Sujet",
+  "message_text": "Contenu du message"
+}
+```
+
+### Variables d'environnement RabbitMQ
+
+| Variable | Description | Exemple |
+|----------|-------------|---------|
+| `USE_RABBITMQ` | Activer l'intégration RabbitMQ | `True` |
+| `RABBITMQ_URL` | URL de connexion RabbitMQ | `amqp://admin:admin@rabbitmq:5672` |
+| `RABBITMQ_EXCHANGE` | Nom de l'exchange | `email_exchange` |
+| `RABBITMQ_ROUTING_KEY` | Routing key pour lier la queue | `email_routing_key` |
+| `RABBITMQ_QUEUE` | Nom de la queue | `email_queue` |
+| `RABBITMQ_DEFAULT_USER` | Utilisateur RabbitMQ | `admin` |
+| `RABBITMQ_DEFAULT_PASS` | Mot de passe RabbitMQ | `admin` |
+
+### Points importants pour l'intégration RabbitMQ
+
+- Le conteneur qui exécute le service doit être sur le même réseau Docker que RabbitMQ afin d'utiliser l'adresse interne (ex. `rabbitmq:5672`). Sans réseau partagé, la résolution de nom et la connexion échoueront.
+- Pour les clients externes (depuis la machine hôte), utilisez l'endpoint exposé de RabbitMQ (ex. `localhost:5672`) si les ports sont mappés.
+- Assurez-vous que `RABBITMQ_URL` pointe vers l'endpoint correct selon le contexte (interne au réseau Docker vs externe).
+- Vérifiez que l'exchange et la routing key configurés correspondent à ceux utilisés par vos producteurs de messages.
+- Le service déclare automatiquement l'exchange de type `TOPIC` et la queue durable, puis les lie avec la routing key spécifiée.
+
+
 ## 🚨 Sécurité et bonnes pratiques
 
 ### Recommandations de sécurité

app.py

@@ -1,6 +1,7 @@
 from fastapi import FastAPI
 from api import router as api_router
 from kafka_config import consume_messages
+from rabbitmq_config import consume_rabbitmq_messages
 from contextlib import asynccontextmanager
 
 
@@ -10,6 +11,7 @@ async def LifeSpan(app: FastAPI):
     import asyncio
 
     loop = asyncio.get_event_loop()
+    loop.run_in_executor(None, consume_rabbitmq_messages)
     loop.run_in_executor(None, consume_messages)
     yield
     print("-----------------APP ENDED----------------------")

rabbitmq.py

@@ -0,0 +1,76 @@
+import datetime
+import aio_pika
+import asyncio
+from utils import SETTINGS, send_email
+
+
+async def connect_to_rabbitmq(rabbitmq_url: str = SETTINGS.RABBITMQ_URL) -> tuple:
+    try:
+        print("Connecting to RabbitMQ...")
+        connection = await aio_pika.connect_robust(rabbitmq_url, timeout=5)
+
+        channel = await connection.channel()
+        exchange = await channel.declare_exchange(
+            SETTINGS.RABBITMQ_EXCHANGE, aio_pika.ExchangeType.TOPIC
+        )
+        queue = await channel.declare_queue(SETTINGS.RABBITMQ_QUEUE, durable=True)
+
+        await queue.bind(exchange, routing_key=SETTINGS.RABBITMQ_ROUTING_KEY)
+
+        print("Connected to RabbitMQ")
+        return connection, exchange, queue
+    except Exception as e:
+        import traceback
+
+        print(f"Error connecting to RabbitMQ: {e}")
+        traceback.print_exc()
+        return None, None, None
+
+
+async def consume_rabbitmq_messages_():
+    connection, exchange, queue = await connect_to_rabbitmq()
+    if not connection:
+        print("Failed to connect to RabbitMQ. Exiting consumer.")
+        return
+
+    async with connection:
+        async with queue.iterator() as queue_iter:
+            async for message in queue_iter:
+                async with message.process():
+                    try:
+                        payload = message.body.decode()
+                        print(f"Received message from RabbitMQ: {payload}")
+                        # Assuming payload is a JSON string with required fields
+                        import json
+
+                        data = json.loads(payload)
+                        receiver_email = data.get("receiver_email")
+                        message_text = data.get("message_text")
+                        email_object = data.get("email_object")
+
+                        if not all([receiver_email, message_text, email_object]):
+                            print(
+                                f"[{datetime.datetime.now()}] Incomplete email data received: {data}"
+                            )
+                            print("The correct format is:")
+                            print(
+                                '{"receiver_email": "email@example.com", "email_object": "Subject", "message_text": "Body"} or {"receiver_email": ["email@example.com"], "email_object": "Subject", "message_text": "Body"}'
+                            )
+                            print("Skipping this message.")
+                            continue
+
+                        send_email(receiver_email, message_text, email_object)
+                    except Exception as e:
+                        print(f"Error processing message: {e}")
+                    finally:
+                        await asyncio.sleep(3)  # Prevent tight loop
+
+
+def consume_rabbitmq_messages():
+    if not SETTINGS.USE_RABBITMQ:
+        print(
+            "RabbitMQ integration is disabled. Setting USE_RABBITMQ to True to enable it and restart the service."
+        )
+        return
+    print("Starting RabbitMQ consumer...")
+    asyncio.run(consume_rabbitmq_messages_())

rabbitmq_config.py

@@ -40,3 +40,4 @@ uvicorn==0.37.0
 watchfiles==1.1.0
 websockets==15.0.1
 kafka-python==2.2.15
+aio_pika==9.5.7

requirements.txt

@@ -25,6 +25,15 @@ class Settings(BaseSettings):
     KAFKA_CONSUMER_TOPIC: str = None
     KAFKA_MESSAGE_KEY: str | None = None
 
+    # RabbitMQ configurations
+    USE_RABBITMQ: bool = False
+    RABBITMQ_URL: str = None
+    RABBITMQ_EXCHANGE: str = "email_exchange"
+    RABBITMQ_ROUTING_KEY: str = None
+    RABBITMQ_QUEUE: str = "email_queue"
+    RABBITMQ_DEFAULT_USER: str = None
+    RABBITMQ_DEFAULT_PASS: str = None
+
     class Config:
         env_file = ".env"