Ajout de l'intégration Kafka

Author: serge-eric-kalaga

.env.exemple

@@ -5,6 +5,15 @@ EMAIL_USERNAME=
 EMAIL_PASSWORD=
 FROM_EMAIL=
 
+# Kafka configuration
+# Set to True to enable Kafka integration
+USE_KAFKA= 
+# Comma-separated list of Kafka bootstrap servers
+KAFKA_BOOTSTRAP_SERVERS= 
+# Topic to consume email messages from
+KAFKA_CONSUMER_TOPIC=email_topic 
+# Key for Kafka messages
+KAFKA_MESSAGE_KEY=
 
 # Example values:
 # EMAIL_DOMAIN=example.com

.gitignore

@@ -155,4 +155,7 @@ data/mdm/data/*
 #Postgres
 data/postgres/
 
-.env
+.env
+
+docker-compose-kafka.yaml
+docker-compose-rabbitmq.yaml

README.md

@@ -367,6 +367,51 @@ curl -X POST "http://localhost:9876/api/send-email" \
 echo "=== Fin du diagnostic ==="
 ```
 
+## 🔁 Intégration Kafka
+
+Le service peut consommer des messages Kafka pour envoyer automatiquement des emails si l'intégration Kafka est activée via les variables d'environnement (`USE_KAFKA`, `KAFKA_BOOTSTRAP_SERVERS`, `KAFKA_CONSUMER_TOPIC`, `KAFKA_MESSAGE_KEY`).
+
+Format JSON attendu pour le message (payload) avec la clé Kafka `email_topic` (vous pouvez modifier la clé via `KAFKA_MESSAGE_KEY`) :
+
+```json
+{
+  "receiver_email": "user@example.com",
+  "email_object": "Sujet",
+  "message_text": "Contenu du message"
+}
+```
+
+Points importants pour une intégration en production ou en environnement conteneurisé :
+
+- Le conteneur qui exécute le service doit être sur le même réseau Docker que le broker Kafka afin d'utiliser l'adresse interne du broker (ex. `broker:9093`). 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é du broker (ex. `localhost:9092`) si les ports sont mappés.
+- Assurez-vous que `KAFKA_BOOTSTRAP_SERVERS` pointe vers l'endpoint correct selon le contexte (interne au réseau Docker vs externe).
+- Vérifiez que le topic configuré (`KAFKA_CONSUMER_TOPIC`) correspond au topic sur lequel sont publiés les messages.
+
+Exemple minimal (attacher le service au réseau Docker interne du broker) :
+```yaml
+version: '3.8'
+
+networks:
+  local-kafka:
+    external: true
+
+services:
+  mail_service:
+    image: kalagaserge/mail_service
+    container_name: mail_service
+    ports:
+      - "9876:9876"
+    environment:
+      - USE_KAFKA=True
+      - KAFKA_BOOTSTRAP_SERVERS=broker:9093
+      - KAFKA_CONSUMER_TOPIC=email_topic
+      - KAFKA_MESSAGE_KEY=email_topic
+    networks:
+      - local-kafka
+    restart: unless-stopped
+```
+
 ## 🚨 Sécurité et bonnes pratiques
 
 ### Recommandations de sécurité

app.py

@@ -1,13 +1,26 @@
 from fastapi import FastAPI
-import uvicorn
 from api import router as api_router
+from kafka_config import consume_messages
+from contextlib import asynccontextmanager
+
+
+@asynccontextmanager
+async def LifeSpan(app: FastAPI):
+    print("-----------------APP STARTED----------------------")
+    import asyncio
+
+    loop = asyncio.get_event_loop()
+    loop.run_in_executor(None, consume_messages)
+    yield
+    print("-----------------APP ENDED----------------------")
 
 
 app = FastAPI(
     title="Mail Service",
     version="1.0.0",
     description="Service for sending emails",
     summary="This API provides endpoints for sending emails.",
+    lifespan=LifeSpan,
 )
 
 app.include_router(api_router, prefix="/api", tags=["Mail Service"])

docker-compose.yaml

@@ -9,3 +9,10 @@ services:
       - .:/app
     env_file:
       - .env
+    networks:
+      - local-kafka
+
+networks:
+  local-kafka:
+    name: local-kafka
+    external: true

kafka_config.py

@@ -0,0 +1,82 @@
+from kafka import KafkaConsumer
+from utils import SETTINGS, send_email
+from datetime import datetime
+import json
+import time
+
+
+def make_consumer(topic, bootstrap_servers, group_id="mail_service-group"):
+    """Create and return a Kafka consumer with exponential backoff on connection failure."""
+    backoff = 1
+    while True:
+        try:
+            c = KafkaConsumer(
+                topic,
+                bootstrap_servers=bootstrap_servers,
+                auto_offset_reset="earliest",
+                enable_auto_commit=True,
+                group_id=group_id,
+            )
+            print("Connected to Kafka")
+            return c
+        except Exception as e:
+            print(f"Kafka connection failed: {e}. retrying in {backoff}s")
+            time.sleep(backoff)
+            backoff = min(backoff * 2, 30)
+
+
+def consume_messages():
+    """Consume messages from a specified Kafka topic."""
+    if not SETTINGS.USE_KAFKA:
+        print(
+            "Kafka integration is disabled. Setting USE_KAFKA to True to enable it and restart the service."
+        )
+        return
+
+    print("Waiting for the producer to start...")
+
+    consumer = make_consumer(
+        topic=SETTINGS.KAFKA_CONSUMER_TOPIC,
+        bootstrap_servers=SETTINGS.KAFKA_BOOTSTRAP_SERVERS.split(","),
+    )
+
+    print("Consumer is ready and listening for messages...")
+
+    try:
+        while True:
+            for message in consumer:
+                ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
+                key = message.key.decode("utf-8") if message.key else "None"
+                data = message.value.decode("utf-8")
+
+                if key == "email_topic":
+                    # Get email details from the message
+                    email_data = json.loads(data)
+                    receiver_email = email_data.get("receiver_email")
+                    email_object = email_data.get("email_object")
+                    message_text = email_data.get("message_text")
+
+                    if not all([receiver_email, email_object, message_text]):
+                        print(f"[{ts}] Incomplete email data received: {email_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
+
+                    print(
+                        f"[{ts}] Sending email to {receiver_email} with subject '{email_object}'"
+                    )
+                    # Send the email
+                    send_email(
+                        receiver_email=receiver_email,
+                        message_text=message_text,
+                        email_object=email_object,
+                    )
+
+    except KeyboardInterrupt:
+        pass
+    finally:
+        consumer.close()
+        time.sleep(3)

rabbitmq.py

@@ -39,3 +39,4 @@ urllib3==2.5.0
 uvicorn==0.37.0
 watchfiles==1.1.0
 websockets==15.0.1
+kafka-python==2.2.15

requirements.txt

@@ -8,13 +8,23 @@
 
 
 class Settings(BaseSettings):
+    # App configurations
+    PYTHONUNBUFFERED: int = 1
+
+    # Email configurations
     EMAIL_DOMAIN: str = "gmail.com"
     EMAIL_HOST: str
     EMAIL_PORT: int = 587
     EMAIL_USERNAME: str
     EMAIL_PASSWORD: str
     FROM_EMAIL: str | None = None
 
+    # Kafka configurations
+    USE_KAFKA: bool = False
+    KAFKA_BOOTSTRAP_SERVERS: str = None
+    KAFKA_CONSUMER_TOPIC: str = None
+    KAFKA_MESSAGE_KEY: str | None = None
+
     class Config:
         env_file = ".env"