feat: add mysql and redis part

Author: zakahan

veadk/database/database_adapter.py

@@ -54,11 +54,32 @@ def query(self, query: str, index: str, top_k: int = 0) -> list:
             logger.error(f"Failed to search from Redis: index={index} error={e}")
             raise e
 
-    def delete_docs(self, index: str, ids: list[int]): ...
+    def delete_doc(self, index: str, id: str) -> bool:
+        logger.debug(f"Deleting document from Redis database: index={index} id={id}")
+        try:
+            # For Redis, we need to handle deletion differently since RedisDatabase.delete_doc
+            # takes a key and a single id
+            result = self.client.delete_doc(key=index, id=id)
+            return result
+        except Exception as e:
+            logger.error(
+                f"Failed to delete document from Redis database: index={index} id={id} error={e}"
+            )
+            return False
 
-    def list_docs(
-        self, index: str, offset: int = 0, limit: int = 100
-    ) -> list[dict]: ...
+    def list_docs(self, index: str, offset: int = 0, limit: int = 100) -> list[dict]:
+        logger.debug(f"Listing documents from Redis database: index={index}")
+        try:
+            # Get all documents from Redis
+            docs = self.client.list_docs(key=index)
+
+            # Apply offset and limit for pagination
+            return docs[offset : offset + limit]
+        except Exception as e:
+            logger.error(
+                f"Failed to list documents from Redis database: index={index} error={e}"
+            )
+            return []
 
 
 class RelationalDatabaseAdapter:
@@ -114,11 +135,27 @@ def query(self, query: str, index: str, top_k: int) -> list[str]:
 
         return [item["data"] for item in results]
 
-    def delete_docs(self, index: str, ids: list[int]): ...
+    def delete_doc(self, index: str, id: str) -> bool:
+        logger.debug(f"Deleting document from SQL database: table_name={index} id={id}")
+        try:
+            # Convert single id to list for the client method
+            result = self.client.delete_doc(table=index, ids=[int(id)])
+            return result
+        except Exception as e:
+            logger.error(
+                f"Failed to delete document from SQL database: table_name={index} id={id} error={e}"
+            )
+            return False
 
-    def list_docs(
-        self, index: str, offset: int = 0, limit: int = 100
-    ) -> list[dict]: ...
+    def list_docs(self, index: str, offset: int = 0, limit: int = 100) -> list[dict]:
+        logger.debug(f"Listing documents from SQL database: table_name={index}")
+        try:
+            return self.client.list_docs(table=index, offset=offset, limit=limit)
+        except Exception as e:
+            logger.error(
+                f"Failed to list documents from SQL database: table_name={index} error={e}"
+            )
+            return []
 
 
 class VectorDatabaseAdapter:

veadk/database/kv/redis_database.py

@@ -110,3 +110,47 @@ def delete(self, **kwargs):
         except Exception as e:
             logger.error(f"Failed to delete key `{key}`: {e}")
             raise e
+
+    def delete_doc(self, key: str, id: str) -> bool:
+        """Delete a specific document by ID from a Redis list.
+
+        Args:
+            key: The Redis key (list) to delete from
+            id: The ID of the document to delete
+
+        Returns:
+            bool: True if deletion was successful, False otherwise
+        """
+        try:
+            # Get all items in the list
+            items = self._client.lrange(key, 0, -1)
+
+            # Find the index of the item to delete
+            for i, item in enumerate(items):
+                # Assuming the item is stored as a JSON string with an 'id' field
+                # If it's just the content, we'll use the list index as ID
+                if str(i) == id:
+                    self._client.lrem(key, 1, item)
+                    return True
+
+            logger.warning(f"Document with id {id} not found in key {key}")
+            return False
+        except Exception as e:
+            logger.error(f"Failed to delete document with id {id} from key {key}: {e}")
+            return False
+
+    def list_docs(self, key: str) -> list[dict]:
+        """List all documents in a Redis list.
+
+        Args:
+            key: The Redis key (list) to list documents from
+
+        Returns:
+            list[dict]: List of documents with id and content
+        """
+        try:
+            items = self._client.lrange(key, 0, -1)
+            return [{"id": str(i), "content": item} for i, item in enumerate(items)]
+        except Exception as e:
+            logger.error(f"Failed to list documents from key {key}: {e}")
+            return []

veadk/database/relational/mysql_database.py

@@ -111,5 +111,62 @@ def delete(self, **kwargs):
             logger.error(f"Failed to drop table {table}: {e}")
             raise e
 
+    def delete_doc(self, table: str, ids: list[int]) -> bool:
+        """Delete documents by IDs from a MySQL table.
+
+        Args:
+            table: The table name to delete from
+            ids: List of document IDs to delete
+
+        Returns:
+            bool: True if deletion was successful, False otherwise
+        """
+        if not self.table_exists(table):
+            logger.warning(f"Table {table} does not exist. Skipping delete operation.")
+            return False
+
+        if not ids:
+            return True  # Nothing to delete
+
+        try:
+            with self._connection.cursor() as cursor:
+                # Create placeholders for the IDs
+                placeholders = ",".join(["%s"] * len(ids))
+                sql = f"DELETE FROM `{table}` WHERE id IN ({placeholders})"
+                cursor.execute(sql, ids)
+                self._connection.commit()
+                logger.info(f"Deleted {cursor.rowcount} documents from table {table}")
+                return True
+        except Exception as e:
+            logger.error(f"Failed to delete documents from table {table}: {e}")
+            return False
+
+    def list_docs(self, table: str, offset: int = 0, limit: int = 100) -> list[dict]:
+        """List documents from a MySQL table.
+
+        Args:
+            table: The table name to list documents from
+            offset: Offset for pagination
+            limit: Limit for pagination
+
+        Returns:
+            list[dict]: List of documents with id and content
+        """
+        if not self.table_exists(table):
+            logger.warning(f"Table {table} does not exist. Returning empty list.")
+            return []
+
+        try:
+            with self._connection.cursor() as cursor:
+                sql = f"SELECT id, data FROM `{table}` ORDER BY created_at DESC LIMIT %s OFFSET %s"
+                cursor.execute(sql, (limit, offset))
+                results = cursor.fetchall()
+                return [
+                    {"id": str(row["id"]), "content": row["data"]} for row in results
+                ]
+        except Exception as e:
+            logger.error(f"Failed to list documents from table {table}: {e}")
+            return []
+
     def is_empty(self):
         pass