Allow returning multiple messages in a queue binding

Fixes: #192

Author: elprans

azure/functions_worker/bindings/meta.py

@@ -1,10 +1,12 @@
 import abc
+import collections.abc
 import datetime
 import enum
 import json
 import typing
 
 from .. import protos
+from .. import typing_inspect
 
 
 class TypedDataKind(enum.Enum):
@@ -211,6 +213,28 @@ def to_proto(cls, obj: typing.Any, *,
         pass
 
 
+def is_iterable_type_annotation(annotation: type, pytype: type) -> bool:
+    is_iterable_anno = (
+        typing_inspect.is_generic_type(annotation) and
+        issubclass(typing_inspect.get_origin(annotation),
+                   collections.abc.Iterable)
+    )
+
+    if not is_iterable_anno:
+        return False
+
+    args = typing_inspect.get_args(annotation)
+    if not args:
+        return False
+
+    if isinstance(pytype, tuple):
+        return any(isinstance(t, type) and issubclass(t, arg)
+                   for t in pytype for arg in args)
+    else:
+        return any(isinstance(pytype, type) and issubclass(pytype, arg)
+                   for arg in args)
+
+
 def is_binding(bind_name: str) -> bool:
     return bind_name in _ConverterMeta._binding_types
 

azure/functions_worker/bindings/queue.py

@@ -1,3 +1,4 @@
+import collections.abc
 import datetime
 import json
 import typing
@@ -99,7 +100,11 @@ class QueueMessageOutConverter(meta.OutConverter, binding='queue'):
 
     @classmethod
     def check_output_type_annotation(cls, pytype: type) -> bool:
-        return issubclass(pytype, (azf_abc.QueueMessage, str, bytes))
+        valid_types = (azf_abc.QueueMessage, str, bytes)
+        return (
+            issubclass(pytype, valid_types) or
+            meta.is_iterable_type_annotation(pytype, valid_types)
+        )
 
     @classmethod
     def to_proto(cls, obj: typing.Any, *,
@@ -118,6 +123,25 @@ def to_proto(cls, obj: typing.Any, *,
                 })
             )
 
+        elif isinstance(obj, collections.abc.Iterable):
+            msgs = []
+            for item in obj:
+                if isinstance(item, str):
+                    msgs.append(item)
+                elif isinstance(item, azf_queue.QueueMessage):
+                    msgs.append({
+                        'id': item.id,
+                        'body': item.get_body().decode('utf-8')
+                    })
+                else:
+                    raise NotImplementedError(
+                        'invalid data type in output '
+                        'queue message list: {}'.format(type(item)))
+
+            return protos.TypedData(
+                json=json.dumps(msgs)
+            )
+
         raise NotImplementedError
 
     @classmethod

azure/functions_worker/functions.py

@@ -1,4 +1,5 @@
 import inspect
+import operator
 import typing
 
 import azure.functions as azf
@@ -138,6 +139,15 @@ def add_function(self, function_id: str,
                         f'binding {param.name} has invalid Out annotation '
                         f'{param_anno!r}')
                 param_py_type = param_anno_args[0]
+
+                # typing_inspect.get_args() returns a flat list,
+                # so if the annotation was func.Out[typing.List[foo]],
+                # we need to reconstruct it.
+                if (isinstance(param_py_type, tuple) and
+                        typing_inspect.is_generic_type(param_py_type[0])):
+
+                    param_py_type = operator.getitem(
+                        param_py_type[0], *param_py_type[1:])
             else:
                 param_py_type = param_anno
 

tests/queue_functions/put_queue_return_multiple/function.json

@@ -0,0 +1,19 @@
+{
+  "scriptFile": "main.py",
+  "disabled": false,
+
+  "bindings": [
+    {
+      "type": "httpTrigger",
+      "direction": "in",
+      "name": "req"
+    },
+    {
+      "direction": "out",
+      "name": "msgs",
+      "queueName": "testqueue-return-multiple",
+      "connection": "AzureWebJobsStorage",
+      "type": "queue"
+    }
+  ]
+}

tests/queue_functions/put_queue_return_multiple/main.py

@@ -0,0 +1,6 @@
+import typing
+import azure.functions as azf
+
+
+def main(req: azf.HttpRequest, msgs: azf.Out[typing.List[str]]):
+    msgs.set(['one', 'two'])

tests/queue_functions/queue_trigger_return_multiple/function.json

@@ -0,0 +1,14 @@
+{
+  "scriptFile": "main.py",
+  "disabled": false,
+
+  "bindings": [
+    {
+      "type": "queueTrigger",
+      "direction": "in",
+      "name": "msg",
+      "queueName": "testqueue-return-multiple",
+      "connection": "AzureWebJobsStorage",
+    }
+  ]
+}

tests/queue_functions/queue_trigger_return_multiple/main.py

@@ -0,0 +1,9 @@
+import logging
+import azure.functions as azf
+
+
+logger = logging.getLogger(__name__)
+
+
+def main(msg: azf.QueueMessage) -> None:
+    logging.info('trigger on message: %s', msg.get_body().decode('utf-8'))

tests/test_queue_functions.py

@@ -53,3 +53,11 @@ def test_queue_message_object_return(self):
         r = self.webhost.request('GET', 'get_queue_blob_message_return')
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r.text, 'test-message-object-return')
+
+    def test_queue_return_multiple(self):
+        r = self.webhost.request('POST', 'put_queue_return_multiple',
+                                 data='foo')
+        self.assertTrue(200 <= r.status_code < 300)
+
+        # wait for queue_trigger to process the queue item
+        time.sleep(1)