Refactor #5

README.md

@@ -84,9 +84,13 @@ Not every file or folder is listed, but the important stuff is here.
 * `src/`
     * `ctp_slack_bot/`
         * `core/`: fundamental components like configuration (using pydantic), logging setup (loguru), and custom exceptions
+            * `config.py`: application settings model
         * `db/`: data connection and interface logic
             * `repositories/`: data collection/table interface logic
+                * `mongo_db_vectorized_repository_base.py`: base implementation of a repository corresponding to a MongoDB collection with a search index
+                * `vectorized_chunk_repository.py`: repository interface for `VectorizedChunk`s
         * `models/`: data models
+        * `mime_type_handlers`: parsers for converting bytes of different MIME types to `Chunk`s
         * `services/`: business logic
             * `answer_retrieval_service.py`: obtains an answer to a question from a language model using relevant context
             * `application_health_service.py`: collects the health status of the application components
@@ -94,11 +98,11 @@ Not every file or folder is listed, but the important stuff is here.
             * `context_retrieval_service.py`: queries for relevant context from the database to answer a question
             * `embeddings_model_service.py`: converts text to embeddings
             * `event_brokerage_service.py`: brokers events between decoupled components
+            * `google_drive_service.py`: interfaces with Google Drive
             * `language_model_service.py`: answers questions using relevant context
             * `question_dispatch_service.py`: listens for questions and retrieves relevant context to get answers
             * `schedule_service.py`: runs background jobs
             * `slack_service.py`: handles events from Slack and sends back responses
-            * `vector_database_service.py`: stores and queries chunks
             * `vectorization_service.py`: converts chunks into chunks with embeddings
         * `tasks/`: scheduled tasks to run in the background
         * `utils/`: reusable utilities

notebooks/google_drive_web_vtt_vectorizer_and_storer.ipynb

@@ -27,9 +27,8 @@
     "display_html = partial(display_html, raw=True)\n",
     "\n",
     "container = Container()\n",
-    "mime_type_handlers = container.mime_type_handlers()\n",
-    "mongo_db = await container.mongo_db()\n",
     "google_drive_service = container.google_drive_service()\n",
+    "mime_type_handlers = container.mime_type_handlers()\n",
     "vectorization_service = container.vectorization_service()\n",
     "vectorized_chunk_repository = await container.vectorized_chunk_repository()"
    ]
@@ -58,7 +57,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "## Upload"
+    "## Get the Files"
    ]
   },
   {
@@ -69,12 +68,26 @@
    "source": [
     "item_metadata = google_drive_service.list_directory(\"\", True)\n",
     "display_html(f\"<p>Found {len(item_metadata)} files/folders.</p>\")\n",
-    "display_html(\"\".join(chain(\"<ul>\", (f\"<li>{escape(metadata.folder_path)}/{escape(metadata.name)}</li>\" for metadata in item_metadata), \"</ul>\")))\n",
-    "\n",
+    "display_html(\"\".join(chain(\"<ul>\", (f\"<li>{escape(metadata.folder_path)}/{escape(metadata.name)}</li>\" for metadata in item_metadata), \"</ul>\")))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
     "recent_metadata = tuple(filter(lambda metadata: MODIFICATION_TIME_CUTOFF <= metadata.modified_time, item_metadata))\n",
     "display_html(f\"<p>{len(item_metadata)} files/folders pass the modification time (<em>{MODIFICATION_TIME_CUTOFF}</em>) cut-off.</p>\")\n",
-    "display_html(\"\".join(chain(\"<ul>\", (f\"<li>{escape(metadata.folder_path)}/{escape(metadata.name)}</li>\" for metadata in recent_metadata), \"</ul>\")))\n",
-    "\n",
+    "display_html(\"\".join(chain(\"<ul>\", (f\"<li>{escape(metadata.folder_path)}/{escape(metadata.name)}</li>\" for metadata in recent_metadata), \"</ul>\")))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
     "metadata_to_process = tuple(filter(lambda metadata: metadata.mime_type == MIME_TYPE, recent_metadata))\n",
     "display_html(f\"<p>{len(item_metadata)} files/folders pass the modification time (<em>{MODIFICATION_TIME_CUTOFF}</em>) cut-off and MIME type (<em>{MIME_TYPE}</em>) criterion.</p>\")\n",
     "display_html(\"\".join(chain(\"<ul>\", (f\"<li>{escape(metadata.folder_path)}/{escape(metadata.name)}</li>\" for metadata in metadata_to_process), \"</ul>\")))"
@@ -87,7 +100,7 @@
    "outputs": [],
    "source": [
     "web_vtt_parser = mime_type_handlers[MIME_TYPE]\n",
-    "display_html(f\"<p>{escape(str(type(web_vtt_parser)))}</p><code>{id(web_vtt_parser)}</code>\")"
+    "display_html(f\"<p><code>{escape(type(web_vtt_parser).__name__)}</code> instance identifier <code>{id(web_vtt_parser)}</code>\")"
    ]
   },
   {
@@ -109,6 +122,13 @@
     "display_html(f\"Processed {len(web_vtts)} files.\")"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Chunk, Vectorize, and Store"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -128,9 +148,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "## Close MongoDB Connection\n",
-    "\n",
-    "Don’t forget to clean up…"
+    "## Clean Up"
    ]
   },
   {

src/ctp_slack_bot/app.py

@@ -1,12 +1,25 @@
-from aiohttp.web import Application as WebApplication, AppRunner as WebAppRunner, Response, TCPSite
-from asyncio import all_tasks, CancelledError, create_task, current_task, get_running_loop, run
-from json import dumps
+from asyncio import all_tasks, CancelledError, create_task, current_task, gather, get_running_loop, run
 from loguru import logger
 from signal import SIGINT, SIGTERM
-from typing import Any, Callable
+from typing import Callable
 
-from ctp_slack_bot.containers import Container
-from ctp_slack_bot.core.logging import setup_logging
+from containers import Container
+from core.logging import setup_logging
+
+
+async def handle_shutdown_signal() -> None:
+    logger.info("Received shutdown signal.")
+    for task in all_tasks():
+        if task is not current_task() and not task.done():
+            task.cancel()
+            logger.trace("Cancelled task {}.", task.get_name())
+    logger.info("Cancelled all tasks.")
+
+
+def create_shutdown_signal_handler() -> Callable[[], None]:
+    def shutdown_signal_handler() -> None:
+        create_task(handle_shutdown_signal())
+    return shutdown_signal_handler
 
 
 async def main() -> None:
@@ -17,54 +30,31 @@ async def main() -> None:
     # Set up dependency injection container.
     container = Container()
     container.wire(packages=["ctp_slack_bot"])
+    logger.debug("Created dependency injection container with providers: {}", '; '.join(container.providers))
 
-    # Instantiate services which should be active from the beginning.
-    application_health_service = await container.application_health_service()
+    # Initialize/instantiate services which should be active from the beginning.
     container.content_ingestion_service()
     container.question_dispatch_service()
+    http_server = await container.http_server()
+    socket_mode_handler = await container.socket_mode_handler()
     container.schedule_service()
+    logger.debug("Initialized services.")
 
-    async def health(request): # TODO: Abstract away
-        health_statuses = await application_health_service.get_health()
-        if all(health_statuses.values()):
-            return Response(text=dumps(dict(health_statuses)), content_type="application/json")
-        else:
-            return Response(body=dumps(dict(health_statuses)), content_type="application/json", status=503)
-    http_server = WebApplication()
-    http_server.router.add_get("/health", health)
-    web_app_runner = WebAppRunner(http_server)
-    await web_app_runner.setup()
-    website = TCPSite(web_app_runner, "0.0.0.0", 8080) # TODO: Un-hard-code
-    await website.start()
-
-    async def handle_shutdown_signal() -> None:
-        logger.info("Received shutdown signal.")
-        await web_app_runner.cleanup()
-        for task in all_tasks():
-            if task is not current_task() and not task.done():
-                task.cancel()
-                logger.trace("Cancelled task {}.", task.get_name())
-        logger.info("Cancelled all tasks.")
-
-    def create_shutdown_signal_handler() -> Callable[[], None]:
-        def shutdown_signal_handler() -> None:
-            create_task(handle_shutdown_signal())
-        return shutdown_signal_handler
-
-    # Start the Slack socket mode handler in the background.
-    socket_mode_handler = await container.socket_mode_handler()
-    slack_bolt_task = create_task(socket_mode_handler.start_async())
+    # Install the shutdown signal handler.
     shutdown_signal_handler = create_shutdown_signal_handler()
     loop = get_running_loop()
     loop.add_signal_handler(SIGINT, shutdown_signal_handler)
     loop.add_signal_handler(SIGTERM, shutdown_signal_handler)
+
+    # Start the HTTP server and Slack socket mode handler in the background; clean up resources when shut down.
     try:
-        logger.info("Starting Slack Socket Mode handler…")
-        await slack_bolt_task
+        logger.info("Starting HTTP server and Slack Socket Mode handler…")
+        await gather(http_server.start(), socket_mode_handler.start_async())
     except CancelledError:
         logger.info("Shutting down application…")
     finally:
         await socket_mode_handler.close_async()
+        logger.info("Stopped Slack Socket Mode handler.")
         await container.shutdown_resources()
 
 

src/ctp_slack_bot/containers.py

@@ -1,16 +1,19 @@
 from dependency_injector.containers import DeclarativeContainer
-from dependency_injector.providers import Callable, Dict, List, Resource, Singleton
+from dependency_injector.providers import Coroutine, Callable, Dict, List, Resource, Self, Singleton
 from importlib import import_module
+from itertools import chain
 from openai import AsyncOpenAI
 from pkgutil import iter_modules
 from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
 from slack_bolt.async_app import AsyncApp
 from types import ModuleType
+from typing import Sequence
 
+from ctp_slack_bot.controllers import ControllerBase, ControllerRegistry
 from ctp_slack_bot.core import Settings
 from ctp_slack_bot.db.mongo_db import MongoDBResource
 from ctp_slack_bot.db.repositories.mongo_db_vectorized_chunk_repository import MongoVectorizedChunkRepositoryResource
-from ctp_slack_bot.mime_type_handlers.base import MimeTypeHandlerRegistry
+from ctp_slack_bot.mime_type_handlers import MimeTypeHandlerRegistry
 from ctp_slack_bot.services.answer_retrieval_service import AnswerRetrievalService
 from ctp_slack_bot.services.application_health_service import ApplicationHealthService
 from ctp_slack_bot.services.content_ingestion_service import ContentIngestionService
@@ -19,6 +22,7 @@ from ctp_slack_bot.services.embeddings_model_service import EmbeddingsModelServi
 from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
 from ctp_slack_bot.services.google_drive_service import GoogleDriveService
 from ctp_slack_bot.services.http_client_service import HTTPClientServiceResource
+from ctp_slack_bot.services.http_server_service import HTTPServerResource
 from ctp_slack_bot.services.language_model_service import LanguageModelService
 from ctp_slack_bot.services.question_dispatch_service import QuestionDispatchService
 from ctp_slack_bot.services.schedule_service import ScheduleServiceResource
@@ -26,21 +30,16 @@ from ctp_slack_bot.services.slack_service import SlackServiceResource
 from ctp_slack_bot.services.vectorization_service import VectorizationService
 
 
-def __scan_and_import_modules(package_name: str) -> None:
-    try:
-        package: Optional[ModuleType] = import_module(package_name)
-    except ImportError as e:
-        raise ValueError(f"Package {package_name} not found") from e
-    if not hasattr(package, '__path__'):
-        raise ValueError(f"{package_name} is not a package")
-    for _, module_name, is_pkg in iter_modules(package.__path__):
-        if not is_pkg:
-            import_module(f"{package.__name__}.{module_name}")
-
-__scan_and_import_modules("ctp_slack_bot.mime_type_handlers")
-
-
 class Container(DeclarativeContainer): # TODO: audit for potential async-related bugs.
+    async def __get_http_controllers(dependency_container: Self) -> Sequence[ControllerBase]:
+        return [controller_class(**{dependency_name: await dependency()
+                                    for dependency_name, dependency
+                                    in dependency_container.providers.items()
+                                    if dependency_name != "__self__" and dependency_name in controller_class.model_fields})
+                for controller_class
+                in ControllerRegistry.get_registry()] # TODO: mutable until async support is extended to tuples
+
+    __self__                    = Self()
     settings                    = Singleton(Settings)
     event_brokerage_service     = Singleton(EventBrokerageService)
     schedule_service            = Resource (ScheduleServiceResource,
@@ -102,3 +101,8 @@ class Container(DeclarativeContainer): # TODO: audit for potential async-related
     #                                         mime_type_handler_factory=mime_type_handler_factory)
     application_health_service  = Singleton(ApplicationHealthService,
                                             services=List(mongo_db, slack_service))
+    http_controllers            = Callable (__get_http_controllers,
+                                            dependency_container=__self__)
+    http_server                 = Resource (HTTPServerResource,
+                                            settings=settings,
+                                            controllers=http_controllers)

src/ctp_slack_bot/controllers/__init__.py

@@ -0,0 +1,2 @@
+from .application_health_controller import ApplicationHealthController
+from .base import ControllerBase, ControllerRegistry, Route

src/ctp_slack_bot/controllers/application_health_controller.py

@@ -0,0 +1,24 @@
+from aiohttp.web import json_response, Request, Response
+from pydantic import ConfigDict
+from typing import Self
+
+from .base import ControllerBase, ControllerRegistry, Route
+from ctp_slack_bot.services import ApplicationHealthService
+
+
+@ControllerRegistry.register()
+class ApplicationHealthController(ControllerBase):
+    """
+    Application health reporting endpoints.
+    """
+
+    application_health_service: ApplicationHealthService
+
+    @Route.get("/health")
+    async def get_health(self: Self, request: Request) -> Response:
+        health_statuses = await self.application_health_service.get_health()
+        return json_response(dict(health_statuses), status=200 if all(health_statuses.values()) else 503)
+
+    @property
+    def name(self: Self) -> str:
+        return "application_health_controller"

src/ctp_slack_bot/controllers/base.py

@@ -0,0 +1,55 @@
+from aiohttp.web import Request, Response
+from functools import partial
+from importlib import import_module
+from inspect import getmembers, ismethod
+from pydantic import BaseModel, ConfigDict
+from typing import Awaitable, Callable, ClassVar, Mapping, Self, Sequence, TypeVar
+
+from ctp_slack_bot.core import ApplicationComponentBase
+
+
+AsyncHandler = Callable[[Request], Awaitable[Response]]
+
+
+class Route(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    method: str
+    path: str
+    handler: AsyncHandler
+
+    @staticmethod
+    def get(path: str) -> Callable[[AsyncHandler], AsyncHandler]:
+        def decorator(function: AsyncHandler) -> AsyncHandler:
+            function._http_method = "GET"
+            function._http_path = path
+            return function
+        return decorator
+
+
+class ControllerBase(ApplicationComponentBase):
+
+    def get_routes(self: Self) -> Sequence[Route]:
+        return tuple(Route(method=method._http_method, path=method._http_path, handler=method)
+                     for name, method in getmembers(self, predicate=ismethod)
+                     if name != 'get_routes' and hasattr(method, "_http_method") and hasattr(method, "_http_path"))
+
+
+T = TypeVar('T', bound=ControllerBase)
+
+
+class ControllerRegistry:
+
+    _registry: ClassVar[list[T]] = []
+
+    @classmethod
+    def get_registry(cls) -> Mapping[T, Sequence[Route]]:
+        import_module(__package__)
+        return tuple(cls._registry)
+
+    @classmethod
+    def register(cls):
+        def decorator(controller_cls: T):
+            cls._registry.append(controller_cls)
+            return controller_cls
+        return decorator

src/ctp_slack_bot/core/__init__.py

@@ -1,2 +1,2 @@
-from ctp_slack_bot.core.abstractions import AbstractBaseModel, ApplicationComponentBase, HealthReportingApplicationComponentBase
-from ctp_slack_bot.core.config import Settings
+from .abstractions import AbstractBaseModel, ApplicationComponentBase, HealthReportingApplicationComponentBase
+from .config import Settings

src/ctp_slack_bot/core/logging.py

@@ -1,4 +1,4 @@
-from logging import __file__ as logging_file, basicConfig, currentframe, getLogger, Handler, INFO, LogRecord, WARNING
+from logging import __file__ as logging_file, basicConfig, currentframe, ERROR, getLogger, Handler, INFO, LogRecord, WARNING
 from loguru import logger
 from os import access, getenv, W_OK
 from sys import stderr
@@ -94,9 +94,10 @@ def setup_logging() -> None:
     basicConfig(handlers=[InterceptHandler()], level=0, force=True)
 
     # Update logging levels for some noisy libraries.
-    for logger_name in ("uvicorn", "uvicorn.error", "fastapi", "httpx", "pymongo"):
+    for logger_name in ("logging", "uvicorn", "uvicorn.error", "fastapi", "httpx", "pymongo"):
         getLogger(logger_name).setLevel(INFO)
     for logger_name in ("apscheduler"):
         getLogger(logger_name).setLevel(WARNING)
+    getLogger().setLevel(WARNING)
 
     logger.info(f"Logging configured with level {log_level}")

src/ctp_slack_bot/db/__init__.py

@@ -1 +1 @@
-from ctp_slack_bot.db.mongo_db import MongoDB
+from .mongo_db import MongoDB

src/ctp_slack_bot/db/mongo_db.py

@@ -47,7 +47,7 @@ class MongoDB(HealthReportingApplicationComponentBase):
             logger.error("Failed to initialize MongoDB client: {}", e)
             self._client = None
             self._db = None
-            raise
+            raise e
 
     async def ping(self: Self) -> bool:
         """Check if MongoDB connection is alive."""
@@ -89,7 +89,7 @@ class MongoDB(HealthReportingApplicationComponentBase):
             return collection
         except Exception as e:
             logger.error("Error accessing collection '{}': {}", name, e)
-            raise
+            raise e
 
     def close(self: Self) -> None:
         """Close the MongoDB connection."""
@@ -125,7 +125,7 @@ class MongoDBResource(AsyncResource):
                 logger.error("MongoDB connection test failed!")
         except Exception as e:
             logger.error("Error testing MongoDB connection: {}", e)
-            raise
+            raise e
 
     async def shutdown(self: Self, mongo_db: MongoDB) -> None:
         """Close MongoDB connection on shutdown."""

src/ctp_slack_bot/db/repositories/__init__.py

@@ -1,3 +1,3 @@
-from ctp_slack_bot.db.repositories.mongo_db_vectorized_chunk_repository import MongoVectorizedChunkRepository
-from ctp_slack_bot.db.repositories.vectorized_chunk_repository import VectorizedChunkRepository
-from ctp_slack_bot.db.repositories.vector_repository_base import VectorRepositoryBase
+from .mongo_db_vectorized_chunk_repository import MongoVectorizedChunkRepository
+from .vectorized_chunk_repository import VectorizedChunkRepository
+from .mongo_db_vector_repository_base import MongoDbVectorRepositoryBase

src/ctp_slack_bot/db/repositories/mongo_db_vector_repository_base.py

@@ -0,0 +1,39 @@
+from abc import abstractmethod
+from loguru import logger
+from motor.motor_asyncio import AsyncIOMotorCollection
+from pydantic import ConfigDict
+from pymongo.operations import SearchIndexModel
+from typing import Self
+
+from ctp_slack_bot.core import ApplicationComponentBase
+
+
+class MongoDbVectorRepositoryBase(ApplicationComponentBase):
+    """Basic MongoDB vector repository implementation"""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)
+
+    collection: AsyncIOMotorCollection
+
+    async def ensure_indices_exist(self: Self) -> None:
+        """Ensure that indices exist."""
+        await self.ensure_search_index_exists() 
+
+    async def ensure_search_index_exists(self: Self) -> None:
+        """
+        Ensure that a vector search index exists.
+        """
+        search_index_model = self._search_index_model
+        index_name = search_index_model.document["name"]
+        existing_indices = [index["name"] async for index in self.collection.list_search_indexes()]
+        logger.debug("{} existing indices were found: {}", len(existing_indices), existing_indices)
+        if index_name in existing_indices:
+            logger.debug("Index, {}, already exists; duplicate index will not be created.", index_name)
+            return
+        result = await self.collection.create_search_index(search_index_model)
+        logger.info("Vector search index, {}, created for collection {}.", result, self.collection.name)
+
+    @property
+    @abstractmethod
+    def _search_index_model(self: Self) -> SearchIndexModel:
+        pass

src/ctp_slack_bot/db/repositories/mongo_db_vectorized_chunk_repository.py

@@ -1,18 +1,20 @@
 from dependency_injector.resources import AsyncResource
 from loguru import logger
 from pymongo import ASCENDING, ReturnDocument
+from pymongo.operations import SearchIndexModel
 from typing import Any, Collection, Iterable, Mapping, Optional, Self, Sequence, Set
 
 from ctp_slack_bot.core import Settings
 from ctp_slack_bot.models import Chunk, VectorizedChunk, VectorQuery
 from ctp_slack_bot.db.mongo_db import MongoDB
-from ctp_slack_bot.db.repositories.vectorized_chunk_repository import VectorizedChunkRepository
-from ctp_slack_bot.db.repositories.vector_repository_base import VectorRepositoryBase
+from .mongo_db_vector_repository_base import MongoDbVectorRepositoryBase
+from .vectorized_chunk_repository import VectorizedChunkRepository
 
-
-class MongoVectorizedChunkRepository(VectorRepositoryBase, VectorizedChunkRepository):
+class MongoVectorizedChunkRepository(MongoDbVectorRepositoryBase, VectorizedChunkRepository):
     """MongoDB implementation of VectorizedChunkRepository"""
 
+    settings: Settings
+
     async def count_by_id(self: Self, parent_id: str, chunk_id: Optional[str] = None) -> int:
         if chunk_id is None:
             return await self.collection.count_documents({"parent_id": parent_id})
@@ -126,7 +128,6 @@ class MongoVectorizedChunkRepository(VectorRepositoryBase, VectorizedChunkReposi
         return result.deleted_count
 
     async def ensure_indices_exist(self: Self) -> None:
-        await super().ensure_indices_exist()
         index_name = "parent_chunk_unique"
         existing_indices = await self.collection.index_information()
         logger.debug("{} existing indices were found: {}", len(existing_indices), existing_indices)
@@ -134,11 +135,30 @@ class MongoVectorizedChunkRepository(VectorRepositoryBase, VectorizedChunkReposi
             logger.debug("Index, {}, already exists; duplicate index will not be created.", index_name)
         else:
             await self.collection.create_index([("parent_id", ASCENDING), ("chunk_id", ASCENDING)], unique=True, name=index_name)
+        await super().ensure_indices_exist()
 
     @property
     def name(self: Self) -> str:
         return "mongo_db_vectorized_chunk_repository"
 
+    @property
+    def _search_index_model(self: Self) -> SearchIndexModel:
+        return SearchIndexModel(
+            definition={
+                "fields": [
+                    {
+                        "type": "vector",
+                        "path": "embedding",
+                        "numDimensions": self.settings.vector_dimension,
+                        "similarity": "cosine",
+                        "quantization": "scalar"
+                    }
+                ]
+            },
+            name=self.settings.vectorized_chunks_search_index_name or f"{self.collection.name}_vector_index",
+            type="vectorSearch"
+        )
+
 
 class MongoVectorizedChunkRepositoryResource(AsyncResource):
     async def init(self: Self, settings: Settings, mongo_db: MongoDB) -> MongoVectorizedChunkRepository:

src/ctp_slack_bot/db/repositories/vector_repository_base.py

@@ -1,59 +0,0 @@
-from loguru import logger
-from motor.motor_asyncio import AsyncIOMotorCollection
-from pydantic import ConfigDict
-from pymongo.operations import SearchIndexModel
-from typing import Self
-
-from ctp_slack_bot.core import ApplicationComponentBase, Settings
-
-
-class VectorRepositoryBase(ApplicationComponentBase):
-    """MongoDB implementation of VectorizedChunkRepository"""
-
-    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)
-
-    settings: Settings
-    collection: AsyncIOMotorCollection
-
-    async def ensure_indices_exist(self: Self) -> None:
-        """Ensure that indices exist."""
-        await self.ensure_search_index_exists() 
-
-    async def ensure_search_index_exists(self: Self) -> None:
-        """
-        Ensure that a vector search index exists.
-        """
-        index_name = self.settings.vectorized_chunks_search_index_name or f"{self.collection.name}_vector_index"
-        try:
-            existing_indices = [index["name"] async for index in self.collection.list_search_indexes()]
-            logger.debug("{} existing indices were found: {}", len(existing_indices), existing_indices)
-            if index_name in existing_indices:
-                logger.debug("Index, {}, already exists; duplicate index will not be created.", index_name)
-                return
-
-            # Create search index model using MongoDB's recommended approach.
-            search_index_model = SearchIndexModel(
-                definition={
-                    "fields": [
-                        {
-                            "type": "vector",
-                            "path": "embedding",
-                            "numDimensions": self.settings.vector_dimension,
-                            "similarity": "cosine",
-                            "quantization": "scalar"
-                        }
-                    ]
-                },
-                name=index_name,
-                type="vectorSearch"
-            )
-            result = await self.collection.create_search_index(search_index_model)
-            logger.info("Vector search index, {}, created for collection {}.", result, self.collection.name)
-        except Exception as e:
-            if "command not found" in str(e).lower():
-                logger.warning("Vector search not supported by this MongoDB instance. Some functionality may be limited.")
-                await self.collection.create_index("embedding") # Create a fallback standard index on embedding field.
-                logger.info("Created standard index on field, {}, as fallback.", "embedding")
-            else:
-                logger.error("Failed to create any index: {}", e)
-                raise

src/ctp_slack_bot/mime_type_handlers/__init__.py

@@ -1,2 +1,2 @@
-from ctp_slack_bot.mime_type_handlers.base import MimeTypeHandler, MimeTypeHandlerRegistry
-from ctp_slack_bot.mime_type_handlers.text.vtt import WebVTTMimeTypeHandler
+from .base import MimeTypeHandler, MimeTypeHandlerRegistry
+from .text.vtt import WebVTTMimeTypeHandler

src/ctp_slack_bot/mime_type_handlers/text/__init__.py

@@ -1 +1 @@
-from ctp_slack_bot.mime_type_handlers.text.vtt import WebVTTMimeTypeHandler 
+from .vtt import WebVTTMimeTypeHandler 

src/ctp_slack_bot/mime_type_handlers/text/vtt.py

@@ -6,7 +6,7 @@ from types import MappingProxyType
 from typing import Any, ClassVar, Mapping, Optional, Self
 from webvtt import WebVTT
 
-from ctp_slack_bot.mime_type_handlers import MimeTypeHandler, MimeTypeHandlerRegistry
+from ctp_slack_bot.mime_type_handlers.base import MimeTypeHandler, MimeTypeHandlerRegistry
 from ctp_slack_bot.models import Content, WebVTTContent, WebVTTFrame
 
 

src/ctp_slack_bot/models/__init__.py

@@ -1,4 +1,4 @@
-from ctp_slack_bot.models.base import Chunk, Content, VectorizedChunk, VectorQuery
-from ctp_slack_bot.models.google_drive import GoogleDriveMetadata
-from ctp_slack_bot.models.slack import SlackEventPayload, SlackMessage, SlackReaction, SlackResponse, SlackUserTimestampPair
-from ctp_slack_bot.models.webvtt import WebVTTContent, WebVTTFrame
+from .base import Chunk, Content, VectorizedChunk, VectorQuery
+from .google_drive import GoogleDriveMetadata
+from .slack import SlackEventPayload, SlackMessage, SlackReaction, SlackResponse, SlackUserTimestampPair
+from .webvtt import WebVTTContent, WebVTTFrame

src/ctp_slack_bot/models/slack.py

@@ -4,7 +4,7 @@ from pydantic import BaseModel, ConfigDict, PositiveInt
 from types import MappingProxyType
 from typing import Any, Literal, Mapping, Optional, Self
 
-from ctp_slack_bot.models.base import Chunk, Content
+from .base import Chunk, Content
 
 
 class SlackEventPayload(BaseModel):

src/ctp_slack_bot/models/webvtt.py

@@ -6,8 +6,8 @@ from types import MappingProxyType
 from typing import Any, ClassVar, Literal, Mapping, Optional, Self
 from webvtt import Caption, WebVTT
 
-from ctp_slack_bot.models.base import Chunk, Content
 from ctp_slack_bot.utils import to_deep_immutable
+from .base import Chunk, Content
 
 
 class WebVTTFrame(BaseModel):

src/ctp_slack_bot/services/__init__.py

@@ -1,11 +1,13 @@
-from ctp_slack_bot.services.answer_retrieval_service import AnswerRetrievalService
-from ctp_slack_bot.services.application_health_service import ApplicationHealthService
-from ctp_slack_bot.services.content_ingestion_service import ContentIngestionService
-from ctp_slack_bot.services.context_retrieval_service import ContextRetrievalService
-from ctp_slack_bot.services.embeddings_model_service import EmbeddingsModelService
-from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
-from ctp_slack_bot.services.google_drive_service import GoogleDriveService
-from ctp_slack_bot.services.language_model_service import LanguageModelService
-from ctp_slack_bot.services.question_dispatch_service import QuestionDispatchService
-from ctp_slack_bot.services.slack_service import SlackService
-from ctp_slack_bot.services.vectorization_service import VectorizationService
+from .answer_retrieval_service import AnswerRetrievalService
+from .application_health_service import ApplicationHealthService
+from .content_ingestion_service import ContentIngestionService
+from .context_retrieval_service import ContextRetrievalService
+from .embeddings_model_service import EmbeddingsModelService
+from .event_brokerage_service import EventBrokerageService
+from .google_drive_service import GoogleDriveService
+from .http_server_service import HTTPServer
+from .language_model_service import LanguageModelService
+from .question_dispatch_service import QuestionDispatchService
+from. schedule_service import ScheduleService
+from .slack_service import SlackService
+from .vectorization_service import VectorizationService

src/ctp_slack_bot/services/answer_retrieval_service.py

@@ -5,8 +5,8 @@ from typing import Collection, Self
 from ctp_slack_bot.core import ApplicationComponentBase, Settings
 from ctp_slack_bot.enums import EventType
 from ctp_slack_bot.models import Chunk, SlackMessage, SlackResponse
-from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
-from ctp_slack_bot.services.language_model_service import LanguageModelService
+from .event_brokerage_service import EventBrokerageService
+from .language_model_service import LanguageModelService
 
 
 class AnswerRetrievalService(ApplicationComponentBase):

src/ctp_slack_bot/services/content_ingestion_service.py

@@ -6,8 +6,8 @@ from ctp_slack_bot.core import ApplicationComponentBase, Settings
 from ctp_slack_bot.db.repositories import VectorizedChunkRepository
 from ctp_slack_bot.enums import EventType
 from ctp_slack_bot.models import Chunk, Content, SlackMessage
-from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
-from ctp_slack_bot.services.vectorization_service import VectorizationService
+from .event_brokerage_service import EventBrokerageService
+from .vectorization_service import VectorizationService
 
 
 class ContentIngestionService(ApplicationComponentBase):

src/ctp_slack_bot/services/context_retrieval_service.py

@@ -5,7 +5,7 @@ from typing import Self, Sequence
 from ctp_slack_bot.core import ApplicationComponentBase, Settings
 from ctp_slack_bot.db.repositories import VectorizedChunkRepository
 from ctp_slack_bot.models import Chunk, SlackMessage, VectorQuery, VectorizedChunk
-from ctp_slack_bot.services.vectorization_service import VectorizationService
+from .vectorization_service import VectorizationService
 
 
 class ContextRetrievalService(ApplicationComponentBase):

src/ctp_slack_bot/services/http_server_service.py

@@ -1,15 +1,51 @@
-from aiohttp.web import Application, AppRunner, Response, TCPSite
+from aiohttp.web import Application, AppRunner, Response, RouteTableDef, TCPSite
+from dependency_injector.containers import DeclarativeContainer
 from dependency_injector.resources import AsyncResource
-from pydantic import ConfigDict
-from typing import Self
+from itertools import chain
+from loguru import logger
+from pydantic import ConfigDict, PrivateAttr
+from typing import Self, Sequence
 
+from ctp_slack_bot.controllers.base import ControllerBase, Route
 from ctp_slack_bot.core import ApplicationComponentBase, Settings
 
 
-class HTTPServerService(ApplicationComponentBase):
+class HTTPServer(ApplicationComponentBase):
     model_config = ConfigDict(frozen=True)
 
     settings: Settings
+    routes: Sequence[Route]
+    _app_runner: AppRunner = PrivateAttr()
+    _site: TCPSite = PrivateAttr()
 
-    async def listen(self: Self) -> None:
-        pass
+    async def initialize(self: Self) -> None:
+        app = Application()
+        for route in self.routes:
+            app.router.add_route(route.method, route.path, route.handler)
+            logger.info("Registered HTTP route: {} {}", route.method, route.path)
+        self._app_runner = AppRunner(app)
+        await self._app_runner.setup()
+        self._site = TCPSite(self._app_runner, self.settings.http_host, self.settings.http_port)
+
+    async def start(self: Self) -> None:
+        await self._site.start()
+
+    async def stop(self: Self) -> None:
+        await self._app_runner.cleanup()
+
+    @property
+    def name(self: Self) -> str:
+        return "http_server"
+
+
+class HTTPServerResource(AsyncResource):
+    async def init(self: Self, settings: Settings, controllers: Sequence[ControllerBase]) -> HTTPServer:
+        routes = tuple(chain.from_iterable(controller.get_routes()
+                                           for controller
+                                           in controllers))
+        http_server = HTTPServer(settings=settings, routes=routes)
+        await http_server.initialize()
+        return http_server
+
+    async def shutdown(self: Self, http_server: HTTPServer) -> None:
+        await http_server.stop()

src/ctp_slack_bot/services/question_dispatch_service.py

@@ -5,9 +5,9 @@ from typing import Any, Self
 from ctp_slack_bot.core import ApplicationComponentBase, Settings
 from ctp_slack_bot.enums import EventType
 from ctp_slack_bot.models import Chunk, SlackMessage
-from ctp_slack_bot.services.answer_retrieval_service import AnswerRetrievalService
-from ctp_slack_bot.services.context_retrieval_service import ContextRetrievalService
-from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
+from .answer_retrieval_service import AnswerRetrievalService
+from .context_retrieval_service import ContextRetrievalService
+from .event_brokerage_service import EventBrokerageService
 
 
 class QuestionDispatchService(ApplicationComponentBase):

src/ctp_slack_bot/services/slack_service.py

@@ -12,7 +12,7 @@ from typing import Any, ClassVar, Mapping, MutableMapping, Optional, Self, Set
 from ctp_slack_bot.core import HealthReportingApplicationComponentBase
 from ctp_slack_bot.enums import EventType
 from ctp_slack_bot.models import SlackMessage, SlackResponse
-from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
+from .event_brokerage_service import EventBrokerageService
 
 
 class SlackService(HealthReportingApplicationComponentBase):

src/ctp_slack_bot/services/vectorization_service.py

@@ -4,7 +4,7 @@ from typing import Self, Sequence
 
 from ctp_slack_bot.core import ApplicationComponentBase, Settings
 from ctp_slack_bot.models import Chunk, VectorizedChunk
-from ctp_slack_bot.services.embeddings_model_service import EmbeddingsModelService
+from .embeddings_model_service import EmbeddingsModelService
 
 
 class VectorizationService(ApplicationComponentBase):

src/ctp_slack_bot/utils/__init__.py

@@ -1,2 +1,2 @@
-from ctp_slack_bot.utils.immutable import to_deep_immutable
-from ctp_slack_bot.utils.secret_stripper import sanitize_mongo_db_uri
+from .immutable import to_deep_immutable
+from .secret_stripper import sanitize_mongo_db_uri