Cut some more stuff out of the config

Author: AlanCoding

dispatcher/__init__.py

@@ -12,7 +12,7 @@ def run_service() -> None:
     Before calling this you need to configure by calling dispatcher.config.setup
     """
     loop = asyncio.get_event_loop()
-    dispatcher = DispatcherMain.from_config()
+    dispatcher = DispatcherMain.from_settings()
     try:
         loop.run_until_complete(dispatcher.main())
     except KeyboardInterrupt:

dispatcher/brokers/__init__.py

@@ -2,6 +2,7 @@
 from types import ModuleType
 
 from dispatcher.brokers.base import BaseBroker
+from dispatcher.config import settings
 
 
 def get_broker_module(broker_name) -> ModuleType:
@@ -10,11 +11,19 @@ def get_broker_module(broker_name) -> ModuleType:
 
 def get_async_broker(broker_name, broker_config) -> BaseBroker:
     broker_module = get_broker_module(broker_name)
-    broker_cls = getattr(broker_module, 'AsyncBroker')
-    return broker_cls(**broker_config)
+    return broker_module.AsyncBroker(**broker_config)
 
 
 def get_sync_broker(broker_name, broker_config) -> BaseBroker:
     broker_module = get_broker_module(broker_name)
-    broker_cls = getattr(broker_module, 'SyncBroker')
-    return broker_cls(**broker_config)
+    return broker_module.SyncBroker(**broker_config)
+
+
+def get_sync_publisher_from_settings() -> BaseBroker:
+    publish_broker = settings.publish['default_broker']
+    return get_sync_broker(publish_broker, settings.brokers[publish_broker])
+
+
+def get_async_publisher_from_settings() -> BaseBroker:
+    publish_broker = settings.publish['default_broker']
+    return get_async_broker(publish_broker, settings.brokers[publish_broker])

dispatcher/brokers/base.py

@@ -18,7 +18,7 @@ async def aclose(self) -> None: ...
     def get_connection(self): ...
 
     @abstractmethod
-    def publish_message(self, queue, message): ...
+    def publish_message(self, channel=None, message=None): ...
 
     @abstractmethod
     def close(self): ...

dispatcher/config.py

@@ -4,26 +4,16 @@
 
 import yaml
 
-from dispatcher.publish import DispatcherDecorator
-from dispatcher.registry import DispatcherMethodRegistry, registry
-from dispatcher.utils import resolve_callable
-
 
 class DispatcherSettings:
-    def __init__(self, config: dict, task_registry: DispatcherMethodRegistry = registry) -> None:
+    def __init__(self, config: dict) -> None:
         self.brokers: list = config.get('brokers', [])
         self.producers: list = config.get('producers', [])
         self.service: dict = config.get('service', {'max_workers': 3})
-        self.callbacks: dict = config.get('callbacks', {})
-        self.tasks: dict = config.get('tasks', {})
-        self.settings: dict = config.get('settings', {})
-
-        for task_name, options in self.tasks.items():
-            method = resolve_callable(task_name)
-            if method:
-                DispatcherDecorator(task_registry, **options)(method)
-            else:
-                raise RuntimeError(f'Could not locate task given in config: {task_name}')
+        self.publish: dict = config.get('publish', {})
+        # TODO: firmly planned sections of config for later
+        # self.callbacks: dict = config.get('callbacks', {})
+        # self.options: dict = config.get('options', {})
 
 
 def settings_from_file(path: str) -> DispatcherSettings:

dispatcher/main.py

@@ -96,12 +96,12 @@ def __init__(self, service_config: dict, producers: Iterable[producer_module.Bas
         self.events: DispatcherEvents = DispatcherEvents()
 
     @classmethod
-    def from_config(cls) -> 'DispatcherMain':
-        producers = DispatcherMain.get_producers()
+    def from_settings(cls) -> 'DispatcherMain':
+        producers = DispatcherMain.producers_from_settings()
         return cls(settings.service, producers)
 
     @classmethod
-    def get_producers(cls) -> Iterable[producer_module.BaseProducer]:
+    def producers_from_settings(cls) -> Iterable[producer_module.BaseProducer]:
         producers = []
         for broker_name, broker_kwargs in settings.brokers.items():
             broker = get_async_broker(broker_name, broker_kwargs)

dispatcher/registry.py

@@ -6,7 +6,7 @@
 from typing import Callable, Optional, Set, Tuple
 from uuid import uuid4
 
-from dispatcher.brokers import get_sync_broker
+from dispatcher.brokers import get_sync_publisher_from_settings
 from dispatcher.utils import MODULE_METHOD_DELIMITER, DispatcherCallable, resolve_callable
 
 logger = logging.getLogger(__name__)
@@ -82,25 +82,17 @@ def get_async_body(
 
     def apply_async(self, args=None, kwargs=None, queue=None, uuid=None, **kw) -> Tuple[dict, str]:
         queue = queue or self.submission_defaults.get('queue')
-        if not queue:
-            msg = f'{self.fn}: Queue value required and may not be None'
-            logger.error(msg)
-            raise ValueError(msg)
 
         if callable(queue):
             queue = queue()
 
         obj = self.get_async_body(args=args, kwargs=kwargs, uuid=uuid, **kw)
 
-        from dispatcher.conf import settings
+        broker = get_sync_publisher_from_settings()
 
-        publish_broker = settings.publish['default_broker']
-        broker = get_sync_broker(publish_broker, settings.brokers[publish_broker])
+        # TODO: exit if a setting is applied to disable publishing
 
-        # TODO: before sending, consult an app-specific callback if configured
-
-        # NOTE: the kw will communicate things in the database connection data
-        broker.publish_message(queue, json.dumps(obj))
+        broker.publish_message(channel=queue, message=json.dumps(obj))
         return (obj, queue)