Repository: mercadona/rele

Issue:
Filter by message attribute
We would like to have our subscribers filter messages from a particular topic based on message attribute(s). For instance, we have one publisher publishing to topic `foobar`. The message is published with either attribute `location='backyard'` or `location='frontyard'`. 

If we have two subscribers listening to the same topic, one should only act on `location='backyard'`. 

I don't want to publish to two separate topics as they are publishing the same data type, only the location attribute changes from message to message. 

Additionally, we would like to globally define the filter value in the app. If we have two workers running the same subscriber, one should be able to filter the messages only for `backyard` messages and the other for `frontyard`. 

This is a proposal implementation:

```
# Scenario A (Local usage)

# subs.py
@sub('update_location', filter_by={'location': 'backyard'})
def update_location_backyard():
    pass

@sub('update_location', filter_by={'location': 'frontyard'})
def update_location_frontyard():
    pass
```


```
# Scenario B (Global Setting)

# settings.py
RELE_FILTER_BY={
    'update_location': {
        'location': 'frontyard'
    }
}

# subs.py
@sub('update_location', filter_by=settings.RELE_FILTER_BY)
def update_location():
    pass
```

Discussion:

Patch:
diff --git a/Makefile b/Makefile
index 9bb9c26..def8f97 100644
--- a/Makefile
+++ b/Makefile
@@ -25,6 +25,7 @@ test: ## run tests quickly with the default Python
 
 coverage: ## generates codecov report
 	coverage run --source rele runtests.py tests
+	coverage report -m
 
 release: clean install-deploy-requirements sdist ## package and upload a release
 	twine upload -u mercadonatech dist/*
diff --git a/docs/index.rst b/docs/index.rst
index 659092f..ccb45db 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -32,9 +32,11 @@ Quickstart
 __________
 
 .. toctree::
-   :maxdepth: 2
+    :maxdepth: 2
+
+    quickstart/rele_and_emulator
+    quickstart/filters
 
-   quickstart/rele_and_emulator
 
 Modules
 _______
diff --git a/docs/quickstart/filters.rst b/docs/quickstart/filters.rst
new file mode 100644
index 0000000..20340d8
--- /dev/null
+++ b/docs/quickstart/filters.rst
@@ -0,0 +1,55 @@
+Filters
+=======
+
+Filter can be used to execute a subscription with specific parameters.
+There's two types of filters, global or passing a filter_by parameter in the
+subscription.
+
+
+Filters by `filter_by` parameter
+________________________________
+
+This filter is a function that is supposed to return a boolean and this function
+is passed as parameter `filter_by` in the subscription.
+
+.. code:: python
+
+    def landscape_filter(kwargs):
+        return kwargs.get('type') == 'landscape'
+
+
+    # This subscription is going to be called if in the `kwargs`
+    # has a key type with value landscape
+
+    @sub(topic='photo-updated', filter_by=landscape_filter)
+    def sub_process_landscape_photos(data, **kwargs):
+        print(f'Received a photo of type {kwargs.get("type")}')
+
+
+
+Global Filter
+_____________
+
+This filter is specified in the settings with the key `FILTER_SUBS_BY`
+that has a function as value.
+In case a subscription has a filter already it's going to use it's own filter.
+
+.. code:: python
+
+    import os
+
+    def landscape_filter(kwargs):
+        return kwargs.get('type') == 'landscape'
+
+    settings = {
+        'APP_NAME': 'test-rele',
+        'GC_PROJECT_ID': os.environ.get('GC_PROJECT_ID')   ,
+        'GC_CREDENTIALS': os.environ.get('GC_CREDENTIALS'),
+        'SUB_PREFIX': 'rele',
+        'MIDDLEWARE': [],
+        'FILTER_SUBS_BY': landscape_filter,
+    }
+
+
+
+
diff --git a/rele/config.py b/rele/config.py
index 8bac174..01fc176 100644
--- a/rele/config.py
+++ b/rele/config.py
@@ -22,7 +22,9 @@ def setup(setting):
     return config
 
 
-def load_subscriptions_from_paths(sub_module_paths, sub_prefix=None):
+def load_subscriptions_from_paths(
+        sub_module_paths, sub_prefix=None, filter_by=None):
+
     subscriptions = []
     for sub_module_path in sub_module_paths:
         sub_module = importlib.import_module(sub_module_path)
@@ -31,5 +33,9 @@ def load_subscriptions_from_paths(sub_module_paths, sub_prefix=None):
             if isinstance(attribute, Subscription):
                 if sub_prefix and not attribute.prefix:
                     attribute.set_prefix(sub_prefix)
+
+                if filter_by and not attribute.filter_by:
+                    attribute.set_filter_by(filter_by)
+
                 subscriptions.append(attribute)
     return subscriptions
diff --git a/rele/management/commands/runrele.py b/rele/management/commands/runrele.py
index f7f19d0..19ced37 100644
--- a/rele/management/commands/runrele.py
+++ b/rele/management/commands/runrele.py
@@ -17,7 +17,7 @@ class Command(BaseCommand):
     help = 'Start subscriber threads to consume Relé topics.'
 
     def handle(self, *args, **options):
-        subs = self._autodiscover_subs(settings.RELE['SUB_PREFIX'])
+        subs = self._autodiscover_subs()
         self.stdout.write(f'Configuring worker with {len(subs)} '
                           f'subscription(s)...')
         for sub in subs:
@@ -45,9 +45,11 @@ class Command(BaseCommand):
                 self.stdout.write(" * Discovered subs module: %r" % module)
         return subs_modules
 
-    def _autodiscover_subs(self, sub_prefix):
+    def _autodiscover_subs(self):
         return rele.config.load_subscriptions_from_paths(
-            self._discover_subs_modules(), sub_prefix)
+            self._discover_subs_modules(),
+            settings.RELE['SUB_PREFIX'],
+            settings.RELE.get('FILTER_SUBS_BY'))
 
     def _wait_forever(self):
         self.stdout.write('Consuming subscriptions...')
diff --git a/rele/subscription.py b/rele/subscription.py
index 2eb5868..67666c1 100644
--- a/rele/subscription.py
+++ b/rele/subscription.py
@@ -16,7 +16,7 @@ class Subscription:
         self.topic = topic
         self._prefix = prefix
         self._suffix = suffix
-        self._filter_by = filter_by or (lambda **_: True)
+        self._filter_by = filter_by
 
     @property
     def name(self):
@@ -30,18 +30,28 @@ class Subscription:
     def set_prefix(self, prefix):
         self._prefix = prefix
 
+    @property
+    def filter_by(self):
+        return self._filter_by
+
+    def set_filter_by(self, filter_by):
+        self._filter_by = filter_by
+
     def __call__(self, data, **kwargs):
         if 'published_at' in kwargs:
             kwargs['published_at'] = float(kwargs['published_at'])
 
-        if self._filter_by(**kwargs):
-            return self._func(data, **kwargs)
+        if self._filter_returns_false(kwargs):
+            return
 
-        return None
+        return self._func(data, **kwargs)
 
     def __str__(self):
         return f'{self.name} - {self._func.__name__}'
 
+    def _filter_returns_false(self, kwargs):
+        return self._filter_by and not self._filter_by(kwargs)
+
 
 class Callback:
 


Test patch:
diff --git a/tests/test_config.py b/tests/test_config.py
new file mode 100644
index 0000000..f232058
--- /dev/null
+++ b/tests/test_config.py
@@ -0,0 +1,30 @@
+import pytest
+from rele.config import load_subscriptions_from_paths
+from rele import sub
+
+
+@sub(topic='test-topic', prefix='rele')
+def sub_stub(data, **_kwargs):
+    return data
+
+
+class TestLoadSubscriptions:
+    @pytest.fixture
+    def subscriptions(self):
+        return load_subscriptions_from_paths(
+            ['tests.test_config'],
+            sub_prefix='test',
+            filter_by=lambda attrs: attrs.get('lang') == 'en')
+
+    def test_load_subscriptions_in_a_module(self, subscriptions):
+        assert len(subscriptions) == 1
+
+    def test_filter_by_applied_to_subscription_returns_true(
+            self, subscriptions):
+
+        assert subscriptions[-1].filter_by({'lang': 'en'}) is True
+
+    def test_filter_by_applied_to_subscription_returns_false(
+            self, subscriptions):
+
+        assert subscriptions[0].filter_by({'lang': 'es'}) is False
diff --git a/tests/test_subscription.py b/tests/test_subscription.py
index c8ea9ea..6ea8077 100644
--- a/tests/test_subscription.py
+++ b/tests/test_subscription.py
@@ -30,7 +30,7 @@ def sub_published_time_type(data, **kwargs):
     logger.info(f'{type(kwargs["published_at"])}')
 
 
-def landscape_filter(**kwargs):
+def landscape_filter(kwargs):
     return kwargs.get('type') == 'landscape'