SIENTIAPDE-1166

feat: add TTL index functionality to MongoDB activities and update requirements

- Updated MongoDB class to accept TTL index duration in seconds.
- Implemented create_collection_with_ttl_index method to create collections with TTL indexes.
- Added environment variable for TTL index duration in connectors_config.
- Updated tests to cover new TTL index functionality and ensure proper behavior.
- Bumped sientia-dataops-library version to 1.3.4 in requirements.txt.

orchestrator/activities/activities.py

@@ -55,6 +55,7 @@ class Activities(  # Couchbase,
         MongoDB.__init__(self,
                          connection_string=mongodb_config['connection_string'],
                          database_name=mongodb_config['database_name'],
+                         ttl_index_seconds=mongodb_config['ttl_index_seconds'],
                          logger=logger,
                          notification_handler=notification_handler)
 

orchestrator/activities/mongo_db.py

@@ -41,7 +41,7 @@ def clear_mongo_id(docs: list) -> list:
 
 
 class MongoDB(BaseActivity):
-    def __init__(self, connection_string: str, database_name: str,
+    def __init__(self, connection_string: str, database_name: str, ttl_index_seconds: int,
                  logger: Logger,
                  notification_handler: NotificationHandler):
         self.connection_string = connection_string
@@ -53,6 +53,8 @@ class MongoDB(BaseActivity):
 
         self.database = self.client[self.database_name]
 
+        self.ttl_index_seconds = ttl_index_seconds
+
         # Initialize MongoDB client here (omitted for brevity)
         logger.info("MongoDB connection initialized")
 
@@ -295,3 +297,57 @@ class MongoDB(BaseActivity):
             )
             self.error(trace, metadata=metadata)
             raise e
+
+    @activity.defn(name="create_collection_with_ttl_index")
+    async def create_collection_with_ttl_index(self, input_data: dict[str, Any]) -> None:
+        """
+        Create a collection with a TTL index.
+        input_data:
+            - collection_name (str): The name of the collection to create.
+            - ttl_index (str): The name of the TTL index to create.
+        """
+        pipelines = input_data.get("pipelines", {})
+        metadata = input_data.get("metadata", {})
+
+        self.info(
+            f"Creating collection with TTL index for pipelines: {list(pipelines.keys())}",
+            metadata=metadata)
+
+        collection_names = self.database.list_collection_names()
+
+        for pipeline_name, pipeline_config in pipelines.items():
+            collection = pipeline_config["topic"]
+
+            try:
+                # Check if collection exists
+                if collection not in collection_names:
+                    self.database.create_collection(collection)
+
+                collection = self.database[collection]
+                # Check if TTL index exists
+                existing_indexes = collection.list_indexes()
+                ttl_index_exists = False
+                for index in existing_indexes:
+                    if "inserted_at" in index["key"] and index.get("expireAfterSeconds") is not None:
+                        ttl_index_exists = True
+                        break
+
+                # Create TTL index if it doesn't exist
+                if not ttl_index_exists:
+                    collection.create_index(
+                        "inserted_at",
+                        expireAfterSeconds=self.ttl_index_seconds,
+                        background=True
+                    )
+            except Exception as e:
+                trace = traceback.format_exc()
+                self.send_notification(
+                    metadata=metadata,
+                    notification_id="MONGODB_CREATE_COLLECTION_ERROR",
+                    message=f"Failed to create collection {collection} with TTL index: {e}",
+                    block="create_collection_with_ttl_index",
+                    level=NotificationLevel.ERROR,
+                    attachment_content=trace
+                )
+                self.error(trace, metadata=metadata)
+                raise e