src/ctp_slack_bot/containers.py

from dependency_injector.containers import DeclarativeContainer
from dependency_injector.providers import Callable, Resource, Singleton
from importlib import import_module
from pathlib import Path
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from slack_bolt.async_app import AsyncApp

from ctp_slack_bot.core.config import Settings
from ctp_slack_bot.db.mongo_db import MongoDBResource
from ctp_slack_bot.db.repositories.mongo_db_vectorized_chunk_repository import MongoVectorizedChunkRepository
from ctp_slack_bot.mime_type_handlers.base import MimeTypeHandlerMeta
from ctp_slack_bot.services.answer_retrieval_service import AnswerRetrievalService
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.schedule_service import ScheduleServiceResource
from ctp_slack_bot.services.slack_service import SlackServiceResource
from ctp_slack_bot.services.vector_database_service import VectorDatabaseService
from ctp_slack_bot.services.vectorization_service import VectorizationService


def __load_plugins(plugin_dir) -> None:
    for path in Path(plugin_dir).glob("*.py"):
        if path.stem == "__init__":
            continue  # Skip __init__.py files
        module_name = f"{plugin_dir.replace('/', '.')}.{path.stem}"
        import_module(module_name)


__load_plugins("ctp_slack_bot/mime_type_handlers")


class Container(DeclarativeContainer):
    settings = Singleton(Settings)
    event_brokerage_service = Singleton(EventBrokerageService)
    schedule_service = Resource(ScheduleServiceResource, settings=settings)
    mongo_db = Resource(MongoDBResource, settings=settings) # TODO: generalize to any database.
    vectorized_chunk_repository = Singleton(MongoVectorizedChunkRepository, mongo_db=mongo_db)
    vector_database_service = Singleton(VectorDatabaseService, settings=settings, mongo_db=mongo_db)
    embeddings_model_service = Singleton(EmbeddingsModelService, settings=settings)
    vectorization_service = Singleton(VectorizationService, settings=settings, embeddings_model_service=embeddings_model_service)
    content_ingestion_service = Singleton(ContentIngestionService, settings=settings, event_brokerage_service=event_brokerage_service, vector_database_service=vector_database_service, vectorization_service=vectorization_service)
    context_retrieval_service = Singleton(ContextRetrievalService, settings=settings, vectorization_service=vectorization_service, vector_database_service=vector_database_service)
    language_model_service = Singleton(LanguageModelService, settings=settings)
    answer_retrieval_service = Singleton(AnswerRetrievalService, settings=settings, event_brokerage_service=event_brokerage_service, language_model_service=language_model_service)
    question_dispatch_service = Singleton(QuestionDispatchService, settings=settings, event_brokerage_service=event_brokerage_service, content_ingestion_service=content_ingestion_service, context_retrieval_service=context_retrieval_service, answer_retrieval_service=answer_retrieval_service)
    slack_bolt_app = Singleton(AsyncApp, token=settings.provided.SLACK_BOT_TOKEN().get_secret_value())
    slack_service = Resource(SlackServiceResource, event_brokerage_service=event_brokerage_service, slack_bolt_app=slack_bolt_app)
    socket_mode_handler = Singleton(lambda _, app, app_token: AsyncSocketModeHandler(app, app_token), slack_service, slack_bolt_app, settings.provided.SLACK_APP_TOKEN().get_secret_value())
    mime_type_handler_factory = Callable(lambda mime_type: MimeTypeHandlerMeta._registry[mime_type]())
    google_drive_service = Singleton(GoogleDriveService, settings=settings)
    # file_monitor_service = Singleton(FileMonitorService, settings=settings, google_drive_service=google_drive_service, mime_type_handler_factory=mime_type_handler_factory)