Use decorator-based plugin registration

notebooks/google_drive_web_vtt_vectorizer_and_storer.ipynb

@@ -85,7 +85,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "web_vtt_parser = container.mime_type_handler_factory(MIME_TYPE)\n",
+    "web_vtt_parser = container.mime_type_handlers()[MIME_TYPE]\n",
     "display_html(f\"<p>{escape(str(type(web_vtt_parser)))}</p><code>{id(web_vtt_parser)}</code>\")"
    ]
   },
@@ -95,7 +95,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "display_html(f\"<code>{id(container.mime_type_handler_factory(MIME_TYPE))}</code>\")"
+    "display_html(f\"<code>{id(container.mime_type_handlers()[MIME_TYPE])}</code>\")"
    ]
   },
   {

src/ctp_slack_bot/app.py

@@ -18,7 +18,7 @@ async def main() -> None:
     container = Container()
     container.wire(packages=["ctp_slack_bot"])
 
-    # Kick off services which should be active from the start.
+    # Instantiate services which should be active from the beginning.
     application_health_service = await container.application_health_service()
     container.content_ingestion_service()
     container.question_dispatch_service()

src/ctp_slack_bot/containers.py

@@ -1,5 +1,5 @@
 from dependency_injector.containers import DeclarativeContainer
-from dependency_injector.providers import Callable, List, Resource, Singleton
+from dependency_injector.providers import Callable, Dict, List, Resource, Singleton
 from importlib import import_module
 from pkgutil import iter_modules
 from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
@@ -9,7 +9,7 @@ from types import ModuleType
 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 import MimeTypeHandler
+from ctp_slack_bot.mime_type_handlers.base 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
@@ -86,7 +86,9 @@ class Container(DeclarativeContainer): # TODO: audit for potential async-related
                                             slack_service,
                                             slack_bolt_app,
                                             settings)
-    mime_type_handler_factory   = Callable (MimeTypeHandler.for_mime_type)
+    mime_type_handlers          = Dict     ({mime_type: Singleton(handler)
+                                             for mime_type, handler
+                                             in MimeTypeHandlerRegistry.get_registry().items()})
     google_drive_service        = Singleton(GoogleDriveService,
                                             settings=settings)
     # file_monitor_service        = Singleton(FileMonitorService,

src/ctp_slack_bot/mime_type_handlers/__init__.py

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

src/ctp_slack_bot/mime_type_handlers/base.py

@@ -1,31 +1,39 @@
 from abc import ABC, abstractmethod
 from functools import lru_cache
+from importlib import import_module
+from types import MappingProxyType
 from typing import Any, ClassVar, Mapping, Optional
 
+from ctp_slack_bot.core import ApplicationComponentBase
 from ctp_slack_bot.models import Content
 
 
-class MimeTypeHandlerMeta(type):
+class MimeTypeHandler(ApplicationComponentBase):
 
-    _registry: ClassVar[dict[str, type["BaseMimeTypeHandler"]]] = {}
+    @abstractmethod
+    def from_bytes(cls, id: str, metadata: Mapping[str, Any], buffer: bytes) -> Content:
+        pass
 
-    def __init__(cls, name: str, bases: tuple[type, ...], dict: dict[str, Any]) -> None:
-        super().__init__(name, bases, dict)
-        if hasattr(cls, "MIME_TYPE"):
-            MimeTypeHandlerMeta._registry[cls.MIME_TYPE] = cls
 
+class MimeTypeHandlerRegistry:
 
-class MimeTypeHandlerABCMeta(MimeTypeHandlerMeta, type(ABC)):
-    pass
+    _registry: ClassVar[dict[str, MimeTypeHandler]] = {}
 
+    @classmethod
+    def get_registry(cls) -> Mapping[str, MimeTypeHandler]:
+        import_module(__package__)
+        return MappingProxyType(cls._registry)
 
-class MimeTypeHandler(metaclass=MimeTypeHandlerABCMeta):
+    @classmethod
+    def register(cls, mime_type: str):
+        def decorator(handler_cls: MimeTypeHandler):
+            if mime_type in cls._registry:
+                raise ValueError(f"The MIME type, {mime_type}, is already registered.")
+            cls._registry[mime_type] = handler_cls
+            return handler_cls
+        return decorator
 
     @classmethod
     @lru_cache
-    def for_mime_type(cls, mime_type: str) -> Optional[type["MimeTypeHandler"]]:
+    def get_handler(cls, mime_type: str) -> Optional[MimeTypeHandler]:
         return cls._registry.get(mime_type)()
-
-    @abstractmethod
-    def from_bytes(cls, id: str, metadata: Mapping[str, Any], buffer: bytes) -> Content:
-        pass

src/ctp_slack_bot/mime_type_handlers/text/vtt.py

@@ -1,27 +1,28 @@
 from datetime import datetime
 from io import BytesIO
-from re import compile as compile_re
+from pydantic import ConfigDict
+from re import compile as compile_re, Pattern
 from types import MappingProxyType
 from typing import Any, Mapping, Optional, Self
 from webvtt import WebVTT
 
-from ctp_slack_bot.mime_type_handlers import MimeTypeHandler
+from ctp_slack_bot.mime_type_handlers import MimeTypeHandler, MimeTypeHandlerRegistry
 from ctp_slack_bot.models import Content, WebVTTContent, WebVTTFrame
 
 
-ISO_DATE_TIME_PATTERN = compile_re(r"Start time: (\d{4}-\d{2}-\d{2}(?: \d{2}:\d{2}:\d{2}(?:Z|[+-]\d{2}:\d{2})?)?)")
-
-
+@MimeTypeHandlerRegistry.register("text/vtt")
 class WebVTTMimeTypeHandler(MimeTypeHandler):
 
-    MIME_TYPE = "text/vtt"
+    model_config = ConfigDict(frozen=True)
+
+    ISO_DATE_TIME_PATTERN: Pattern = compile_re(r"Start time: (\d{4}-\d{2}-\d{2}(?: \d{2}:\d{2}:\d{2}(?:Z|[+-]\d{2}:\d{2})?)?)")
 
     @classmethod
     def __get_start_time(cls, web_vtt: WebVTT) -> Optional[datetime]:
         try:
             return next(datetime.fromisoformat(result[0])
                         for result
-                        in map(ISO_DATE_TIME_PATTERN.findall, web_vtt.header_comments)
+                        in map(cls.ISO_DATE_TIME_PATTERN.findall, web_vtt.header_comments)
                         if result)
         except (StopIteration, ValueError):
             return None
@@ -32,3 +33,7 @@ class WebVTTMimeTypeHandler(MimeTypeHandler):
                        for index, caption
                        in enumerate(web_vtt.captions, 1))
         return WebVTTContent(id=id, metadata=MappingProxyType(metadata), start_time=WebVTTMimeTypeHandler.__get_start_time(web_vtt), frames=frames)
+
+    @property
+    def name(self: Self) -> str:
+        return "web_vtt_mime_type_handler"