module-2 done. Pulumi config must be added

course/module-2/.gitignore

@@ -26,6 +26,7 @@ share/python-wheels/
 *.egg
 MANIFEST
 
+data/
 # PyInstaller
 #  Usually these files are written by a python script from a template
 #  before PyInstaller builds the exe, so as to inject date/other infos into it.

course/module-2/Makefile

@@ -0,0 +1,11 @@
+help:
+	@grep -E '^[a-zA-Z0-9 -]+:.*#'  Makefile | sort | while read -r l; do printf "\033[1;32m$$(echo $$l | cut -f 1 -d':')\033[00m:$$(echo $$l | cut -f 2- -d'#')\n"; done
+
+local-start: # Buil and start mongodb and mq.
+	docker-compose -f docker-compose.yml up --build -d
+
+local-start-cdc: # Start CDC system
+	python cdc.py
+
+local-test-cdc: #Test CDC system by inserting data to mongodb
+	python test_cdc.py

course/module-2/cdc.py

@@ -1,31 +1,41 @@
 import json
+import logging
 
 from bson import json_util
 
 from data_flow.mq import publish_to_rabbitmq
 from db.mongo import MongoDatabaseConnector
 
+# Configure logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
+
 
 def stream_process():
-    # Setup MongoDB connection
-    client = MongoDatabaseConnector()
-    db = client["scrabble"]
-
-    # Watch changes in a specific collection
-    changes = db.watch([{"$match": {"operationType": {"$in": ["insert"]}}}])
-    for change in changes:
-        data_type = change["ns"]["coll"]
-        entry_id = str(change["fullDocument"]["_id"])  # Convert ObjectId to string
-        change["fullDocument"].pop("_id")
-        change["fullDocument"]["type"] = data_type
-        change["fullDocument"]["entry_id"] = entry_id
-
-        # Use json_util to serialize the document
-        data = json.dumps(change["fullDocument"], default=json_util.default)
-        print("aici")
-
-        # Send data to rabbitmq
-        publish_to_rabbitmq(queue_name="test_queue", data=data)
+    try:
+        # Setup MongoDB connection
+        client = MongoDatabaseConnector()
+        db = client["scrabble"]
+        logging.info("Connected to MongoDB.")
+
+        # Watch changes in a specific collection
+        changes = db.watch([{"$match": {"operationType": {"$in": ["insert"]}}}])
+        for change in changes:
+            data_type = change["ns"]["coll"]
+            entry_id = str(change["fullDocument"]["_id"])  # Convert ObjectId to string
+            change["fullDocument"].pop("_id")
+            change["fullDocument"]["type"] = data_type
+            change["fullDocument"]["entry_id"] = entry_id
+
+            # Use json_util to serialize the document
+            data = json.dumps(change["fullDocument"], default=json_util.default)
+            logging.info(f"Change detected and serialized: {data}")
+
+            # Send data to rabbitmq
+            publish_to_rabbitmq(queue_name="test_queue", data=data)
+            logging.info("Data published to RabbitMQ.")
+
+    except Exception as e:
+        logging.error(f"An error occurred: {e}")
 
 
 if __name__ == "__main__":