Call transform function

We get provided a namespace to the function from the transform
configuration passed back from stagecraft. We use this namespace to find
a function to call, passing through the data and options to get back
transformed data.

backdrop/transformers/dispatch.py

@@ -1,3 +1,4 @@
+import importlib
 import logging
 
 from worker import app, config
@@ -45,4 +46,13 @@ def run_transform(data_set_config, transform, earliest, latest):
 
     data = data_set.get(query_parameters=query_parameters)
 
-    logger.info(data)
+    function_namespace = transform['type']['function']
+    function_name = function_namespace.split('.')[-1]
+    module_namespace = '.'.join(function_namespace.split('.')[:-1])
+
+    transform_module = importlib.import_module(module_namespace)
+    transform_function = getattr(transform_module, function_name)
+
+    transformed_data = transform_function(data['data'], transform['options'])
+
+    logger.info(transformed_data)

backdrop/transformers/tasks/debug.py

@@ -0,0 +1,8 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def logging(data, options):
+    logger.info(data)
+    return []

tests/transformers/test_dispatch.py

@@ -31,17 +31,27 @@ def test_entrypoint(self, mock_app, mock_adminAPI):
             args=({"group": "foo", "type": "bar"}, {'type': 2}, 'earliest', 'latest'))
 
     @patch('backdrop.transformers.dispatch.DataSet')
-    def test_run_transform(self, mock_data_set):
+    @patch('backdrop.transformers.tasks.debug.logging')
+    def test_run_transform(self, mock_logging_task, mock_data_set):
         data_set_instance = MagicMock()
+        data_set_instance.get.return_value = {
+            'data': [
+                {'data': 'point'},
+            ],
+        }
         mock_data_set.from_group_and_type.return_value = data_set_instance
 
         run_transform({
             'data_group': 'group',
             'data_type': 'type',
         }, {
+            'type': {
+                'function': 'backdrop.transformers.tasks.debug.logging',
+            },
             'query-parameters': {
                 'period': 'day',
             },
+            'options': {},
         }, 'earliest', 'latest')
 
         data_set_instance.get.assert_called_with(
@@ -52,3 +62,7 @@ def test_run_transform(self, mock_data_set):
                 'end_at': 'latest',
             },
         )
+        mock_logging_task.assert_called_with(
+            [{'data': 'point'}],
+            {}
+        )