Use full class names in fixtures for simplicity

First pass at global config setup

Finish running and just starting on tests

Add a config test

Make half-way progress through demo script

Cut some more stuff out of the config

Fix failing unit test, handle queue can not be found

Review comment to consolidate factory handling

Factories refactor

Adopt new patterns up to some tests passing

Unfinished start on settings serialization

Author: AlanCoding

README.md

@@ -49,8 +49,8 @@ There are 2 ways to run the dispatcher service:
 - A CLI entrypoint `dispatcher-standalone` for demo purposes
 
 ```python
-from dispatcher.main import DispatcherMain
-import asyncio
+from dispatcher.config import setup
+from dispatcher import run_service
 
 config = {
     "producers": {
@@ -63,13 +63,9 @@ config = {
     },
     "pool": {"max_workers": 4},
 }
-loop = asyncio.get_event_loop()
-dispatcher = DispatcherMain(config)
+setup(config)
 
-try:
-    loop.run_until_complete(dispatcher.main())
-finally:
-    loop.close()
+run_service()
 ```
 
 Configuration tells how to connect to postgres, and what channel(s) to listen to.

dispatcher.yml

@@ -1,20 +1,23 @@
 # Demo config
 ---
-pool:
-  max_workers: 3
-producers:
-  brokers:
-    # List of channels to listen on
+service:
+  max_workers: 4
+brokers:
+  pg_notify:
+    config:
+      conninfo: dbname=dispatch_db user=dispatch password=dispatching host=localhost port=55777
+    sync_connection_factory: dispatcher.brokers.pg_notify.connection_saver
     channels:
     - test_channel
     - test_channel2
     - test_channel3
-    pg_notify:
-      # Database connection details
-      conninfo: dbname=dispatch_db user=dispatch password=dispatching host=localhost
-        port=55777
-  scheduled:
-    'lambda: __import__("time").sleep(1)':
-      schedule: 3
-    'lambda: __import__("time").sleep(2)':
-      schedule: 3
+    default_publish_channel: test_channel
+producers:
+  ScheduledProducer:
+    task_schedule:
+      'lambda: __import__("time").sleep(1)':
+        schedule: 3
+      'lambda: __import__("time").sleep(2)':
+        schedule: 3
+publish:
+  default_broker: pg_notify

dispatcher/__init__.py

@@ -0,0 +1,21 @@
+import asyncio
+import logging
+
+from dispatcher.factories import from_settings
+
+logger = logging.getLogger(__name__)
+
+
+def run_service() -> None:
+    """
+    Runs dispatcher task service (runs tasks due to messages from brokers and other local producers)
+    Before calling this you need to configure by calling dispatcher.config.setup
+    """
+    loop = asyncio.get_event_loop()
+    dispatcher = from_settings()
+    try:
+        loop.run_until_complete(dispatcher.main())
+    except KeyboardInterrupt:
+        logger.info('Dispatcher stopped by KeyboardInterrupt')
+    finally:
+        loop.close()

dispatcher/brokers/__init__.py

@@ -0,0 +1,25 @@
+from abc import abstractmethod
+from typing import Optional
+
+
+class BaseBroker:
+    @abstractmethod
+    async def connect(self): ...
+
+    @abstractmethod
+    async def aprocess_notify(self, connected_callback=None): ...
+
+    @abstractmethod
+    async def apublish_message(self, channel: Optional[str] = None, message: str = '') -> None: ...
+
+    @abstractmethod
+    async def aclose(self) -> None: ...
+
+    @abstractmethod
+    def get_connection(self): ...
+
+    @abstractmethod
+    def publish_message(self, channel=None, message=None): ...
+
+    @abstractmethod
+    def close(self): ...

dispatcher/brokers/base.py

@@ -1,7 +1,11 @@
 import logging
+from typing import Callable, Iterable, Optional
 
 import psycopg
 
+from dispatcher.brokers.base import BaseBroker
+from dispatcher.utils import resolve_callable
+
 logger = logging.getLogger(__name__)
 
 
@@ -13,87 +17,190 @@
 """
 
 
-# TODO: get database data from settings
-# # As Django settings, may not use
-# DATABASES = {
-#     "default": {
-#         "ENGINE": "django.db.backends.postgresql",
-#         "HOST": os.getenv("DB_HOST", "127.0.0.1"),
-#         "PORT": os.getenv("DB_PORT", 55777),
-#         "USER": os.getenv("DB_USER", "dispatch"),
-#         "PASSWORD": os.getenv("DB_PASSWORD", "dispatching"),
-#         "NAME": os.getenv("DB_NAME", "dispatch_db"),
-#     }
-# }
-
-
-async def aget_connection(config):
-    return await psycopg.AsyncConnection.connect(**config, autocommit=True)
-
-
-def get_connection(config):
-    return psycopg.Connection.connect(**config, autocommit=True)
-
-
-async def aprocess_notify(connection, channels, connected_callback=None):
-    async with connection.cursor() as cur:
-        for channel in channels:
-            await cur.execute(f"LISTEN {channel};")
-            logger.info(f"Set up pg_notify listening on channel '{channel}'")
-
-        if connected_callback:
-            await connected_callback()
-
-        while True:
-            logger.debug('Starting listening for pg_notify notifications')
-            async for notify in connection.notifies():
-                yield notify.channel, notify.payload
-
-
-async def apublish_message(connection, channel, payload=None):
-    async with connection.cursor() as cur:
-        if not payload:
-            await cur.execute(f'NOTIFY {channel};')
+class PGNotifyBase(BaseBroker):
+
+    def __init__(
+        self,
+        config: Optional[dict] = None,
+        channels: Iterable[str] = ('dispatcher_default',),
+        default_publish_channel: Optional[str] = None,
+    ) -> None:
+        """
+        channels - listening channels for the service and used for control-and-reply
+        default_publish_channel - if not specified on task level or in the submission
+          by default messages will be sent to this channel.
+          this should be one of the listening channels for messages to be received.
+        """
+        if config:
+            self._config: dict = config.copy()
+            self._config['autocommit'] = True
+        else:
+            self._config = {}
+
+        self.channels = channels
+        self.default_publish_channel = default_publish_channel
+
+    def get_publish_channel(self, channel: Optional[str] = None):
+        "Handle default for the publishing channel for calls to publish_message, shared sync and async"
+        if channel is not None:
+            return channel
+        if self.default_publish_channel is None:
+            raise ValueError('Could not determine a channel to use publish to from settings or PGNotify config')
+        return self.default_publish_channel
+
+    def get_connection_method(self, factory_path: Optional[str] = None) -> Callable:
+        "Handles settings, returns a method (async or sync) for getting a new connection"
+        if factory_path:
+            factory = resolve_callable(factory_path)
+            if not factory:
+                raise RuntimeError(f'Could not import connection factory {factory_path}')
+            return factory
+        elif self._config:
+            return self.create_connection
         else:
-            await cur.execute(f"NOTIFY {channel}, '{payload}';")
-
-
-def get_django_connection():
-    try:
-        from django.conf import ImproperlyConfigured
-        from django.db import connection as pg_connection
-    except ImportError:
-        return None
-    else:
-        try:
-            if pg_connection.connection is None:
-                pg_connection.connect()
-            if pg_connection.connection is None:
-                raise RuntimeError('Unexpectedly could not connect to postgres for pg_notify actions')
-            return pg_connection.connection
-        except ImproperlyConfigured:
-            return None
-
-
-def publish_message(queue, message, config=None, connection=None, new_connection=False):
-    conn = None
-    if connection:
-        conn = connection
-
-    if (not conn) and (not new_connection):
-        conn = get_django_connection()
-
-    created_new_conn = False
-    if not conn:
-        if config is None:
-            raise RuntimeError('Could not use Django connection, and no postgres config supplied')
-        conn = get_connection(config)
-        created_new_conn = True
-
-    with conn.cursor() as cur:
-        cur.execute('SELECT pg_notify(%s, %s);', (queue, message))
-
-    logger.debug(f'Sent pg_notify message to {queue}')
-
-    if created_new_conn:
-        conn.close()
+            raise RuntimeError('Could not construct connection for lack of config or factory')
+
+    def create_connection(self): ...
+
+
+class AsyncBroker(PGNotifyBase):
+    def __init__(
+        self,
+        config: Optional[dict] = None,
+        async_connection_factory: Optional[str] = None,
+        sync_connection_factory: Optional[str] = None,  # noqa
+        connection: Optional[psycopg.AsyncConnection] = None,
+        **kwargs,
+    ) -> None:
+        if not (config or async_connection_factory or connection):
+            raise RuntimeError('Must specify either config or async_connection_factory')
+
+        self._async_connection_factory = async_connection_factory
+        self._connection = connection
+
+        super().__init__(config=config, **kwargs)
+
+    async def get_connection(self) -> psycopg.AsyncConnection:
+        if not self._connection:
+            factory = self.get_connection_method(factory_path=self._async_connection_factory)
+            connection = await factory(**self._config)
+            self._connection = connection
+            return connection  # slightly weird due to MyPY
+        return self._connection
+
+    @staticmethod
+    async def create_connection(**config) -> psycopg.AsyncConnection:
+        return await psycopg.AsyncConnection.connect(**config)
+
+    async def aprocess_notify(self, connected_callback=None):
+        connection = await self.get_connection()
+        async with connection.cursor() as cur:
+            for channel in self.channels:
+                await cur.execute(f"LISTEN {channel};")
+                logger.info(f"Set up pg_notify listening on channel '{channel}'")
+
+            if connected_callback:
+                await connected_callback()
+
+            while True:
+                logger.debug('Starting listening for pg_notify notifications')
+                async for notify in connection.notifies():
+                    yield notify.channel, notify.payload
+
+    async def apublish_message(self, channel: Optional[str] = None, message: str = '') -> None:
+        connection = await self.get_connection()
+        channel = self.get_publish_channel(channel)
+
+        async with connection.cursor() as cur:
+            if not message:
+                await cur.execute(f'NOTIFY {channel};')
+            else:
+                await cur.execute(f"NOTIFY {channel}, '{message}';")
+
+        logger.debug(f'Sent pg_notify message of {len(message)} chars to {channel}')
+
+    async def aclose(self) -> None:
+        if self._connection:
+            await self._connection.close()
+            self._connection = None
+
+
+class SyncBroker(PGNotifyBase):
+    def __init__(
+        self,
+        config: Optional[dict] = None,
+        async_connection_factory: Optional[str] = None,  # noqa
+        sync_connection_factory: Optional[str] = None,
+        connection: Optional[psycopg.Connection] = None,
+        **kwargs,
+    ) -> None:
+        if not (config or sync_connection_factory or connection):
+            raise RuntimeError('Must specify either config or async_connection_factory')
+
+        self._sync_connection_factory = sync_connection_factory
+        self._connection = connection
+        super().__init__(config=config, **kwargs)
+
+    def get_connection(self) -> psycopg.Connection:
+        if not self._connection:
+            factory = self.get_connection_method(factory_path=self._sync_connection_factory)
+            connection = factory(**self._config)
+            self._connection = connection
+            return connection
+        return self._connection
+
+    @staticmethod
+    def create_connection(**config) -> psycopg.Connection:
+        return psycopg.Connection.connect(**config)
+
+    def publish_message(self, channel: Optional[str] = None, message: str = '') -> None:
+        connection = self.get_connection()
+        channel = self.get_publish_channel(channel)
+
+        with connection.cursor() as cur:
+            if message:
+                cur.execute('SELECT pg_notify(%s, %s);', (channel, message))
+            else:
+                cur.execute(f'NOTIFY {channel};')
+
+        logger.debug(f'Sent pg_notify message of {len(message)} chars to {channel}')
+
+    def close(self) -> None:
+        if self._connection:
+            self._connection.close()
+            self._connection = None
+
+
+class ConnectionSaver:
+    def __init__(self) -> None:
+        self._connection: Optional[psycopg.Connection] = None
+        self._async_connection: Optional[psycopg.AsyncConnection] = None
+
+
+connection_save = ConnectionSaver()
+
+
+def connection_saver(**config) -> psycopg.Connection:
+    """
+    This mimics the behavior of Django for tests and demos
+    Philosophically, this is used by an application that uses an ORM,
+    or otherwise has its own connection management logic.
+    Dispatcher does not manage connections, so this a simulation of that.
+    """
+    if connection_save._connection is None:
+        config['autocommit'] = True
+        connection_save._connection = SyncBroker.create_connection(**config)
+    return connection_save._connection
+
+
+async def async_connection_saver(**config) -> psycopg.AsyncConnection:
+    """
+    This mimics the behavior of Django for tests and demos
+    Philosophically, this is used by an application that uses an ORM,
+    or otherwise has its own connection management logic.
+    Dispatcher does not manage connections, so this a simulation of that.
+    """
+    if connection_save._async_connection is None:
+        config['autocommit'] = True
+        connection_save._async_connection = await AsyncBroker.create_connection(**config)
+    return connection_save._async_connection