feat: Add CommaSeparatedStrList validation tests and update project versions

Author: a-klos

admin-api-lib/pyproject.toml

@@ -1,3 +1,14 @@
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry]
+name = "admin-api-lib"
+version = "1.0.1"
+description = "The admin backend is responsible for the document management. This includes deletion, upload and returning the source document."
+authors = ["STACKIT Data and AI Consulting <[email protected]>"]
+packages = [{ include = "admin_api_lib", from = "src" }]
+
 [tool.flake8]
 exclude= [".eggs", "./rag-core-library/*", "./src/admin_api_lib/models/*", "./src/admin_api_lib/rag_backend_client/*", "./src/admin_api_lib/extractor_api_client/*", ".git", ".hg", ".mypy_cache", ".tox", ".venv", ".devcontainer", "venv", "_build", "buck-out", "build", "dist", "**/__init__.py"]
 statistics = true
@@ -51,12 +62,6 @@ known_local_folder = ["admin_api_lib", "rag_core_lib"]
 [tool.pylint]
 max-line-length = 120
 
-[tool.poetry]
-name = "admin_api_lib"
-version = "0.0.1"
-description = "The admin backend is responsible for the document management. This includes deletion, upload and returning the source document."
-authors = ["STACKIT Data and AI Consulting <[email protected]>"]
-
 [tool.poetry.group.dev.dependencies]
 debugpy = "^1.8.1"
 pytest = "^8.2.1"
@@ -85,11 +90,6 @@ black = "^23.9.1"
 # flake8-logging-format = "^2024.24.12"
 # flake8-docstrings = "^1.7.0"
 
-
-[build-system]
-requires = ["poetry-core"]
-build-backend = "poetry.core.masonry.api"
-
 [tool.poetry.dependencies]
 rag-core-lib = {path = "../rag-core-lib"}
 python = "^3.11"
@@ -103,3 +103,9 @@ tqdm = "^4.66.4"
 langfuse = "^2.39.1"
 redis = "^5.0.8"
 pyyaml = "^6.0.2"
+
+[tool.pytest.ini_options]
+log_cli = 1
+log_cli_level = "DEBUG"
+pythonpath = "src"
+testpaths = "src/tests"

admin-api-lib/src/admin_api_lib/dependency_container.py

@@ -148,7 +148,7 @@ class DependencyContainer(DeclarativeContainer):
     )
 
     summary_enhancer = List(
-        Singleton(PageSummaryEnhancer, summarizer),
+        Singleton(PageSummaryEnhancer, summarizer, chunker_settings),
     )
     untraced_information_enhancer = Singleton(
         GeneralEnhancer,

admin-api-lib/src/admin_api_lib/impl/api_endpoints/default_confluence_loader.py

@@ -3,8 +3,10 @@
 import logging
 from asyncio import run
 from threading import Thread
+import threading
 
 from fastapi import HTTPException, status
+from langchain_core.documents import Document
 
 from admin_api_lib.api_endpoints.confluence_loader import ConfluenceLoader
 from admin_api_lib.api_endpoints.document_deleter import DocumentDeleter
@@ -100,10 +102,16 @@ async def aload_from_confluence(self) -> None:
         HTTPException
             If the Confluence loader is not configured or if a load is already in progress.
         """
-        if not (self._settings.url.strip() and self._settings.space_key.strip() and self._settings.token.strip()):
-            raise HTTPException(
-                status.HTTP_501_NOT_IMPLEMENTED, "The confluence loader is not configured! Required fields are missing."
-            )
+        for index in range(len(self._settings.url)):
+            if not (
+                self._settings.url[index].strip()
+                and self._settings.space_key[index].strip()
+                and self._settings.token[index].strip()
+            ):
+                raise HTTPException(
+                    status.HTTP_501_NOT_IMPLEMENTED,
+                    "The confluence loader is not configured! Required fields are missing.",
+                )
 
         if self._background_thread is not None and self._background_thread.is_alive():
             raise HTTPException(
@@ -113,51 +121,69 @@ async def aload_from_confluence(self) -> None:
         self._background_thread.start()
 
     async def _aload_from_confluence(self) -> None:
-        params = self._settings_mapper.map_settings_to_params(self._settings)
-        try:
-            self._key_value_store.upsert(self._settings.document_name, Status.PROCESSING)
-            information_pieces = self._extractor_api.extract_from_confluence_post(params)
-            documents = [self._information_mapper.extractor_information_piece2document(x) for x in information_pieces]
-            chunked_documents = self._chunker.chunk(documents)
-            rag_information_pieces = [
-                self._information_mapper.document2rag_information_piece(doc) for doc in chunked_documents
-            ]
-        except Exception as e:
-            self._key_value_store.upsert(self._settings.document_name, Status.ERROR)
-            logger.error("Error while loading from Confluence: %s", str(e))
-            raise HTTPException(
-                status.HTTP_500_INTERNAL_SERVER_ERROR, f"Error loading from Confluence: {str(e)}"
-            ) from e
-
-        await self._delete_previous_information_pieces()
-        self._key_value_store.upsert(self._settings.document_name, Status.UPLOADING)
-        self._upload_information_pieces(rag_information_pieces)
-
-    async def _delete_previous_information_pieces(self):
+        async def process_confluence(index):
+            logger.info("Loading from Confluence %s", self._settings.url[index])
+            self._sanitize_document_name(index=index)
+
+            params = self._settings_mapper.map_settings_to_params(self._settings, index)
+            try:
+                self._key_value_store.upsert(self._settings.document_name[index], Status.PROCESSING)
+                information_pieces = self._extractor_api.extract_from_confluence_post(params)
+                documents = [
+                    self._information_mapper.extractor_information_piece2document(x) for x in information_pieces
+                ]
+                documents = await self._aenhance_langchain_documents(documents)
+                chunked_documents = self._chunker.chunk(documents)
+                rag_information_pieces = [
+                    self._information_mapper.document2rag_information_piece(doc) for doc in chunked_documents
+                ]
+            except Exception as e:
+                self._key_value_store.upsert(self._settings.document_name[index], Status.ERROR)
+
+                logger.error("Error while loading from Confluence: %s", str(e))
+                raise HTTPException(
+                    status.HTTP_500_INTERNAL_SERVER_ERROR, f"Error loading from Confluence: {str(e)}"
+                ) from e
+
+            await self._delete_previous_information_pieces(index=index)
+            self._key_value_store.upsert(self._settings.document_name[index], Status.UPLOADING)
+            self._upload_information_pieces(rag_information_pieces, index=index)
+
+        threads = []
+        for idx in range(len(self._settings.url)):
+            t = threading.Thread(target=lambda idx=idx: run(process_confluence(idx)))
+            threads.append(t)
+            t.start()
+        for t in threads:
+            t.join()
+
+    async def _delete_previous_information_pieces(self, index=0):
         try:
-            await self._document_deleter.adelete_document(self._settings.document_name)
+            await self._document_deleter.adelete_document(self._settings.document_name[index])
         except HTTPException as e:
             logger.error(
                 (
                     "Error while trying to delete documents with id: %s before uploading %s."
                     "NOTE: Still continuing with upload."
                 ),
-                self._settings.document_name,
+                self._settings.document_name[index],
                 e,
             )
 
-    def _upload_information_pieces(self, rag_api_documents):
+    def _upload_information_pieces(self, rag_api_documents, index=0):
         try:
             self._rag_api.upload_information_piece(rag_api_documents)
-            self._key_value_store.upsert(self._settings.document_name, Status.READY)
+            self._key_value_store.upsert(self._settings.document_name[index], Status.READY)
             logger.info("Confluence loaded successfully")
         except Exception as e:
-            self._key_value_store.upsert(self._settings.document_name, Status.ERROR)
+            self._key_value_store.upsert(self._settings.document_name[index], Status.ERROR)
             logger.error("Error while uploading Confluence to the database: %s", str(e))
             raise HTTPException(500, f"Error loading from Confluence: {str(e)}") from e
 
-    def _sanitize_document_name(self) -> None:
-        document_name = self._settings.document_name if self._settings.document_name else self._settings.url
+    def _sanitize_document_name(self, index) -> None:
+        document_name = (
+            self._settings.document_name[index] if self._settings.document_name[index] else self._settings.url[index]
+        )
         document_name = document_name.replace("http://", "").replace("https://", "")
 
-        self._settings.document_name = sanitize_document_name(document_name)
+        self._settings.document_name[index] = sanitize_document_name(document_name)

admin-api-lib/src/admin_api_lib/impl/information_enhancer/page_summary_enhancer.py

@@ -29,9 +29,15 @@ class PageSummaryEnhancer(SummaryEnhancer):
 
     async def _acreate_summary(self, information: list[Document], config: Optional[RunnableConfig]) -> list[Document]:
         # group infos by page, defaulting to page 1 if no page metadata
+        if self._chunker_settings:
+            filtered_information = [
+                info for info in information if len(info.page_content) > self._chunker_settings.max_size
+            ]
+        else:
+            filtered_information = information
         grouped = [
-            [info for info in information if info.metadata.get("page", self.DEFAULT_PAGE_NR) == page]
-            for page in {info_piece.metadata.get("page", self.DEFAULT_PAGE_NR) for info_piece in information}
+            [info for info in filtered_information if info.metadata.get("page", self.DEFAULT_PAGE_NR) == page]
+            for page in {info_piece.metadata.get("page", self.DEFAULT_PAGE_NR) for info_piece in filtered_information}
         ]
 
         summary_tasks = [self._asummarize_page(info_group, config) for info_group in tqdm(grouped)]

admin-api-lib/src/admin_api_lib/impl/information_enhancer/summary_enhancer.py

@@ -3,6 +3,7 @@
 from abc import abstractmethod
 from typing import Optional
 
+from admin_api_lib.impl.settings.chunker_settings import ChunkerSettings
 from langchain_core.documents import Document
 from langchain_core.runnables import RunnableConfig, ensure_config
 
@@ -26,7 +27,7 @@ class SummaryEnhancer(InformationEnhancer):
 
     INFORMATION_METADATA_TYPE = "type"
 
-    def __init__(self, summarizer: Summarizer):
+    def __init__(self, summarizer: Summarizer, chunker_settings: ChunkerSettings = None):
         """
         Initialize the SummaryEnhancer with a given Summarizer instance.
 
@@ -37,6 +38,7 @@ def __init__(self, summarizer: Summarizer):
         """
         super().__init__()
         self._summarizer = summarizer
+        self._chunker_settings = chunker_settings
 
     @staticmethod
     def _is_relevant(information: Document) -> bool:

admin-api-lib/src/admin_api_lib/impl/key_db/file_status_key_value_store.py

@@ -76,7 +76,6 @@ def upsert(self, file_name: str, file_status: Status) -> None:
         None
         """
         self.remove(file_name)
-
         self._redis.sadd(self.STORAGE_KEY, FileStatusKeyValueStore._to_str(file_name, file_status))
 
     def remove(self, file_name: str) -> None:

admin-api-lib/src/admin_api_lib/impl/mapper/confluence_settings_mapper.py

@@ -10,7 +10,7 @@ class ConfluenceSettingsMapper:
     """Mapper class for converting ConfluenceSettings to ConfluenceParameters."""
 
     @staticmethod
-    def map_settings_to_params(settings: ConfluenceSettings) -> ConfluenceParameters:
+    def map_settings_to_params(settings: ConfluenceSettings, index) -> ConfluenceParameters:
         """
         Map ConfluenceSettings to ConfluenceParameters.
 
@@ -25,11 +25,12 @@ def map_settings_to_params(settings: ConfluenceSettings) -> ConfluenceParameters
             The parameters object for API consumption.
         """
         return ConfluenceParameters(
-            url=settings.url,
-            token=settings.token,
-            space_key=settings.space_key,
-            include_attachments=settings.include_attachments,
-            keep_markdown_format=settings.keep_markdown_format,
-            keep_newlines=settings.keep_newlines,
-            document_name=settings.document_name,
+            url=settings.url[index],
+            token=settings.token[index],
+            space_key=settings.space_key[index],
+            include_attachments=settings.include_attachments[index],
+            keep_markdown_format=settings.keep_markdown_format[index],
+            keep_newlines=settings.keep_newlines[index],
+            document_name=settings.document_name[index],
+            confluence_kwargs=[{"key": "verify_ssl", "value": settings.verify_ssl[index]}],
         )