Update the code based on feedback. set Default to enabled, add to documentation how to disable it. Add extra documentation

cookbook/connectors/connector.py

@@ -3,6 +3,7 @@ from abc import ABC, abstractmethod
 from cookbook.models import ShoppingListEntry, Space, ConnectorConfig
 
 
+# A Connector is 'destroyed' & recreated each time 'any' ConnectorConfig in a space changes.
 class Connector(ABC):
     @abstractmethod
     def __init__(self, config: ConnectorConfig):
@@ -12,6 +13,7 @@ class Connector(ABC):
     async def on_shopping_list_entry_created(self, space: Space, instance: ShoppingListEntry) -> None:
         pass
 
+    # This method might not trigger on 'direct' entry updates: https://stackoverflow.com/a/35238823
     @abstractmethod
     async def on_shopping_list_entry_updated(self, space: Space, instance: ShoppingListEntry) -> None:
         pass

cookbook/connectors/connector_manager.py

@@ -7,18 +7,18 @@ from dataclasses import dataclass
 from enum import Enum
 from multiprocessing import JoinableQueue
 from types import UnionType
-from typing import List, Any, Dict, Optional
+from typing import List, Any, Dict, Optional, Type
 
 from django_scopes import scope
+from django.conf import settings
 
 from cookbook.connectors.connector import Connector
 from cookbook.connectors.homeassistant import HomeAssistant
-from cookbook.models import ShoppingListEntry, Recipe, MealPlan, Space, ConnectorConfig
+from cookbook.models import ShoppingListEntry, Space, ConnectorConfig
 
 multiprocessing.set_start_method('fork')  # https://code.djangoproject.com/ticket/31169
 
-QUEUE_MAX_SIZE = 25
-REGISTERED_CLASSES: UnionType = ShoppingListEntry | Recipe | MealPlan
+REGISTERED_CLASSES: UnionType | Type = ShoppingListEntry
 
 
 class ActionType(Enum):
@@ -33,12 +33,20 @@ class Work:
     actionType: ActionType
 
 
+# The way ConnectionManager works is as follows:
+# 1. On init, it starts a worker & creates a queue for 'Work'
+# 2. Then any time its called, it verifies the type of action (create/update/delete) and if the item is of interest, pushes the Work (non blocking) to the queue.
+# 3. The worker consumes said work from the queue.
+# 3.1 If the work is of type ConnectorConfig, it flushes its cache of known connectors (per space.id)
+# 3.2 If work is of type REGISTERED_CLASSES, it asynchronously fires of all connectors and wait for them to finish (runtime should depend on the 'slowest' connector)
+# 4. Work is marked as consumed, and next entry of the queue is consumed.
+# Each 'Work' is processed in sequential by the worker, so the throughput is about [workers * the slowest connector]
 class ConnectorManager:
     _queue: JoinableQueue
     _listening_to_classes = REGISTERED_CLASSES | ConnectorConfig
 
     def __init__(self):
-        self._queue = multiprocessing.JoinableQueue(maxsize=QUEUE_MAX_SIZE)
+        self._queue = multiprocessing.JoinableQueue(maxsize=settings.EXTERNAL_CONNECTORS_QUEUE_SIZE)
         self._worker = multiprocessing.Process(target=self.worker, args=(self._queue,), daemon=True)
         self._worker.start()
 
@@ -75,7 +83,7 @@ class ConnectorManager:
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
 
-        _connectors: Dict[str, List[Connector]] = dict()
+        _connectors: Dict[int, List[Connector]] = dict()
 
         while True:
             try:
@@ -90,7 +98,7 @@ class ConnectorManager:
             refresh_connector_cache = isinstance(item.instance, ConnectorConfig)
 
             space: Space = item.instance.space
-            connectors: Optional[List[Connector]] = _connectors.get(space.name)
+            connectors: Optional[List[Connector]] = _connectors.get(space.id)
 
             if connectors is None or refresh_connector_cache:
                 if connectors is not None:
@@ -111,7 +119,7 @@ class ConnectorManager:
 
                         connectors.append(connector)
 
-                    _connectors[space.name] = connectors
+                    _connectors[space.id] = connectors
 
             if len(connectors) == 0 or refresh_connector_cache:
                 worker_queue.task_done()
@@ -134,6 +142,9 @@ class ConnectorManager:
 async def close_connectors(connectors: List[Connector]):
     tasks: List[Task] = [asyncio.create_task(connector.close()) for connector in connectors]
 
+    if len(tasks) == 0:
+        return
+
     try:
         await asyncio.gather(*tasks, return_exceptions=False)
     except BaseException:
@@ -161,6 +172,7 @@ async def run_connectors(connectors: List[Connector], space: Space, instance: RE
         return
 
     try:
+        # Wait for all async tasks to finish, if one fails, the others still continue.
         await asyncio.gather(*tasks, return_exceptions=False)
     except BaseException:
         logging.exception("received an exception from one of the connectors")