Repository: mercadona/rele (python)

Issue:
SUB_PREFIX is not added to subscriptions
When running Relé as a standalone worker (ie, no django, flask, etc.) I expect the subscribers to be prefixed with the SUB_PREFIX value. 

However, this does not happen since we are manually registering the subs in the worker. 

This can result in a conflicts when having multiple workers subscribed to the same topic, leading to race conditions when consuming a message.  

Pull request:
CLI
### :tophat: What?

Provide a CLI for Relé

### :thinking: Why?

- Easily run a worker.
- Auto Discovery of subs. 
- Auto Discovery of settings.
- Ability to add more functionality to the CLI (ie. graph...)

### :link: Related issue

Fixes #153 

Here is an example on my local test application. 

<img width="755" alt="Screen Shot 2020-05-25 at 9 25 29 AM" src="https://user-images.githubusercontent.com/14836934/82789209-3ae48c00-9e6a-11ea-88a0-32c9641eefed.png">
<img width="560" alt="Screen Shot 2020-05-25 at 9 24 56 AM" src="https://user-images.githubusercontent.com/14836934/82789266-4a63d500-9e6a-11ea-8f3e-b7b256d2b632.png">

Note: This doesnt include tests or documentation. If this feature is something we want, I will continue adding the necessary things to get it merged including tests, and documentation. 


Patch:
diff --git a/docs/guides/unrecoverable_middleware.rst b/docs/guides/unrecoverable_middleware.rst
index c62dfbf..640ee45 100644
--- a/docs/guides/unrecoverable_middleware.rst
+++ b/docs/guides/unrecoverable_middleware.rst
@@ -1,7 +1,7 @@
 .. _unrecoverable_middleware:
 
 Unrecoverable Middleware
-===========
+========================
 
 To acknowledge and ignore incompatible messages that your subscription is unable to handle, you can use the `UnrecoverableMiddleware`.
 
diff --git a/rele/__main__.py b/rele/__main__.py
new file mode 100644
index 0000000..8852d19
--- /dev/null
+++ b/rele/__main__.py
@@ -0,0 +1,39 @@
+import argparse
+import logging
+import os
+import sys
+
+from rele import config, discover, subscription
+
+from rele.worker import create_and_run
+
+logger = logging.getLogger(__name__)
+
+
+def main():
+    # modify path so we can import modules and packages
+    sys.path.insert(0, os.getcwd())
+
+    parser = argparse.ArgumentParser(
+        prog="Relé", description="Harness the power of Relé from the command line"
+    )
+
+    subparsers = parser.add_subparsers(help="Select a command", dest="command")
+
+    run_parser = subparsers.add_parser(
+        "run",
+        help="Run a Relé worker with auto-discovery of subs modules in the "
+        "current path. Auto-discovery will include all subs "
+        "and settings modules. If no settings module is discovered, "
+        "defaults will be used.",
+    )
+
+    args = parser.parse_args()
+
+    if args.command == "run":
+        settings, module_paths = discover.sub_modules()
+        configuration = config.setup(settings.RELE if settings else None)
+        subs = config.load_subscriptions_from_paths(
+            module_paths, configuration.sub_prefix, configuration.filter_by
+        )
+        create_and_run(subs, configuration)
diff --git a/rele/config.py b/rele/config.py
index f2d66bc..ae405a7 100644
--- a/rele/config.py
+++ b/rele/config.py
@@ -37,6 +37,7 @@ class Config:
         self._encoder_path = setting.get("ENCODER_PATH", DEFAULT_ENCODER_PATH)
         self.publisher_timeout = setting.get("PUBLISHER_TIMEOUT", 3.0)
         self.threads_per_subscription = setting.get("THREADS_PER_SUBSCRIPTION", 2)
+        self.filter_by = setting.get("FILTER_SUBS_BY")
 
     @property
     def encoder(self):
diff --git a/rele/discover.py b/rele/discover.py
new file mode 100644
index 0000000..2d43e20
--- /dev/null
+++ b/rele/discover.py
@@ -0,0 +1,46 @@
+import logging
+import pkgutil
+from importlib.util import find_spec as importlib_find
+
+logger = logging.getLogger(__name__)
+
+
+def module_has_submodule(package, module_name):
+    """
+    See if 'module' is in 'package'.
+    Taken from https://github.com/django/django/blob/master/django/utils/module_loading.py#L63
+    """
+    package = __import__(package)
+    package_name = package.__name__
+    package_path = package.__path__
+    full_module_name = package_name + "." + module_name
+
+    try:
+        return importlib_find(full_module_name, package_path) is not None
+    except (ModuleNotFoundError, AttributeError):
+        # When module_name is an invalid dotted path, Python raises
+        # ModuleNotFoundError.
+        return False
+
+
+def sub_modules():
+    """
+    In the current PYTHONPATH, we can traverse all modules and determine if they
+    have a settings.py or directory with a subs.py module. If either one of
+    those exists, we import it, and return the settings module, and
+    paths to the subs file.
+
+    If a settings module is not found, we return None.
+
+    :return: (settings module, List[string: subs module paths])
+    """
+    settings = None
+    module_paths = []
+    for f, package, is_package in pkgutil.iter_modules(path=["."]):
+        if package == "settings":
+            settings = __import__(package)
+        if is_package and module_has_submodule(package, "subs"):
+            module = package + ".subs"
+            module_paths.append(module)
+            logger.debug(" * Discovered subs module: %r" % module)
+    return settings, module_paths
diff --git a/rele/management/commands/runrele.py b/rele/management/commands/runrele.py
index 3c7e2e5..7757681 100644
--- a/rele/management/commands/runrele.py
+++ b/rele/management/commands/runrele.py
@@ -1,13 +1,10 @@
 import logging
-import signal
 
 from django.conf import settings
 from django.core.management import BaseCommand
 
-import rele
-from rele import Worker
-from rele.config import Config
-
+from rele.worker import create_and_run
+from rele import config
 from rele.management.discover import discover_subs_modules
 
 logger = logging.getLogger(__name__)
@@ -15,7 +12,7 @@ logger = logging.getLogger(__name__)
 
 class Command(BaseCommand):
     help = "Start subscriber threads to consume messages from Relé topics."
-    config = Config(settings.RELE)
+    config = config.Config(settings.RELE)
 
     def handle(self, *args, **options):
         if all(map(lambda x: x.get("CONN_MAX_AGE"), settings.DATABASES.values())):
@@ -26,27 +23,8 @@ class Command(BaseCommand):
                     "be exhausted."
                 )
             )
-        subs = self._autodiscover_subs()
-        self.stdout.write(f"Configuring worker with {len(subs)} " f"subscription(s)...")
-        for sub in subs:
-            self.stdout.write(f"  {sub}")
-        worker = Worker(
-            subs,
-            self.config.gc_project_id,
-            self.config.credentials,
-            self.config.ack_deadline,
-            self.config.threads_per_subscription,
-        )
-
-        signal.signal(signal.SIGINT, signal.SIG_IGN)
-        signal.signal(signal.SIGTERM, worker.stop)
-        signal.signal(signal.SIGTSTP, worker.stop)
-
-        worker.run_forever()
-
-    def _autodiscover_subs(self):
-        return rele.config.load_subscriptions_from_paths(
-            discover_subs_modules(),
-            self.config.sub_prefix,
-            settings.RELE.get("FILTER_SUBS_BY"),
+        subs = config.load_subscriptions_from_paths(
+            discover_subs_modules(), self.config.sub_prefix, self.config.filter_by
         )
+        self.stdout.write(f"Configuring worker with {len(subs)} " f"subscription(s)...")
+        create_and_run(subs, self.config)
diff --git a/rele/subscription.py b/rele/subscription.py
index c9e0391..4c752d5 100644
--- a/rele/subscription.py
+++ b/rele/subscription.py
@@ -2,6 +2,7 @@ import json
 import logging
 import time
 
+import rele
 from .middleware import run_middleware_hook
 
 logger = logging.getLogger(__name__)
diff --git a/rele/worker.py b/rele/worker.py
index 6b1e901..6ad6901 100644
--- a/rele/worker.py
+++ b/rele/worker.py
@@ -1,4 +1,5 @@
 import logging
+import signal
 import sys
 import time
 from concurrent import futures
@@ -108,3 +109,32 @@ class Worker:
         logger.info("Consuming subscriptions...")
         while True:
             time.sleep(sleep_interval)
+
+
+def create_and_run(subs, config):
+    """
+    Create and run a worker from a list of Subscription objects and a config
+    while waiting forever, until the process is stopped.
+
+    We stop a worker process on:
+    - SIGINT
+    - SIGTSTP
+
+    :param subs: List :class:`~rele.subscription.Subscription`
+    :param config: :class:`~rele.config.Config`
+    """
+    print(f"Configuring worker with {len(subs)} subscription(s)...")
+    for sub in subs:
+        print(f"  {sub}")
+    worker = Worker(
+        subs,
+        config.gc_project_id,
+        config.credentials,
+        config.ack_deadline,
+        config.threads_per_subscription,
+    )
+
+    signal.signal(signal.SIGINT, signal.SIG_IGN)
+    signal.signal(signal.SIGTERM, worker.stop)
+    signal.signal(signal.SIGTSTP, worker.stop)
+    worker.run_forever()
diff --git a/setup.py b/setup.py
index f9661d4..3f47ccc 100644
--- a/setup.py
+++ b/setup.py
@@ -58,4 +58,5 @@ setup(
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
     ],
+    entry_points={"console_scripts": ["rele-cli=rele.__main__:main"]},
 )


Test patch:
diff --git a/tests/commands/test_runrele.py b/tests/commands/test_runrele.py
index e10638e..5dfdeaf 100644
--- a/tests/commands/test_runrele.py
+++ b/tests/commands/test_runrele.py
@@ -14,7 +14,7 @@ class TestRunReleCommand:
 
     @pytest.fixture
     def mock_worker(self):
-        with patch("rele.management.commands.runrele.Worker", autospec=True) as p:
+        with patch("rele.worker.Worker", autospec=True) as p:
             yield p
 
     def test_calls_worker_start_and_setup_when_runrele(self, mock_worker):
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 395c356..8ea650a 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -7,6 +7,7 @@ from google.cloud.pubsub_v1.subscriber.scheduler import ThreadScheduler
 
 from rele import Subscriber, Worker, sub
 from rele.middleware import register_middleware
+from rele.worker import create_and_run
 
 
 @sub(topic="some-cool-topic", prefix="rele")
@@ -130,3 +131,24 @@ class TestWorker:
 
         assert worker._subscriber._ack_deadline == 60
         assert worker._subscriber._gc_project_id == "rele"
+
+
+class TestCreateAndRun:
+    @pytest.fixture(autouse=True)
+    def worker_wait_forever(self):
+        with patch.object(Worker, "_wait_forever", return_value=None) as p:
+            yield p
+
+    @pytest.fixture
+    def mock_worker(self):
+        with patch("rele.worker.Worker", autospec=True) as p:
+            yield p
+
+    def test_waits_forever_when_called_with_config_and_subs(self, config, mock_worker):
+        subscriptions = (sub_stub,)
+        create_and_run(subscriptions, config)
+
+        mock_worker.assert_called_with(
+            subscriptions, "test-project-id", "my-credentials", 60, 2
+        )
+        mock_worker.return_value.run_forever.assert_called_once_with()