Merge branch 'dev'

Author: Hanzhang Zeng (Roger)

azure_functions_worker/bindings/datumdef.py

@@ -1,5 +1,8 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+
+from typing import Any
+import json
 from .. import protos
 
 
@@ -8,6 +11,29 @@ def __init__(self, value, type):
         self.value = value
         self.type = type
 
+    @property
+    def python_value(self) -> Any:
+        if self.value is None or self.type is None:
+            return None
+        elif self.type in ('bytes', 'string', 'int', 'double'):
+            return self.value
+        elif self.type == 'json':
+            return json.loads(self.value)
+        elif self.type == 'collection_string':
+            return [v for v in self.value.string]
+        elif self.type == 'collection_bytes':
+            return [v for v in self.value.bytes]
+        elif self.type == 'collection_double':
+            return [v for v in self.value.double]
+        elif self.type == 'collection_sint64':
+            return [v for v in self.value.sint64]
+        else:
+            return self.value
+
+    @property
+    def python_type(self) -> type:
+        return type(self.python_value)
+
     def __eq__(self, other):
         if not isinstance(other, type(self)):
             return False

azure_functions_worker/dispatcher.py

@@ -170,9 +170,9 @@ def on_logging(self, record: logging.LogRecord, formatted_msg: str) -> None:
             log_level = getattr(protos.RpcLog, 'None')
 
         if is_system_log_category(record.name):
-            log_category = protos.RpcLog.RpcLogCategory.System
+            log_category = protos.RpcLog.RpcLogCategory.Value('System')
         else:  # customers using logging will yield 'root' in record.name
-            log_category = protos.RpcLog.RpcLogCategory.User
+            log_category = protos.RpcLog.RpcLogCategory.Value('User')
 
         log = dict(
             level=log_level,

setup.py

@@ -257,7 +257,7 @@ def run(self):
 
 setup(
     name='azure-functions-worker',
-    version='1.1.3',
+    version='1.1.4',
     description='Python Language Worker for Azure Functions Host',
     long_description=long_description,
     long_description_content_type='text/markdown',
@@ -286,6 +286,8 @@ def run(self):
     extras_require={
         'dev': [
             'azure-functions==1.3.0',
+            'azure-eventhub~=5.1.0',
+            'python-dateutil~=2.8.1',
             'flake8~=3.7.9',
             'mypy',
             'pytest',

tests/endtoend/eventhub_batch_functions/eventhub_multiple/__init__.py

@@ -3,6 +3,8 @@
 import json
 
 
+# This is an actual EventHub trigger which handles Eventhub events in batches.
+# It serializes multiple event data into a json and store it into a blob.
 def main(events):
     table_entries = []
     for event in events:

tests/endtoend/eventhub_batch_functions/eventhub_output_batch/__init__.py

@@ -3,6 +3,7 @@
 import azure.functions as func
 
 
+# An HttpTrigger to generating EventHub event from EventHub Output Binding
 def main(req: func.HttpRequest) -> str:
     events = req.get_body().decode('utf-8')
     return events

tests/endtoend/eventhub_batch_functions/get_eventhub_batch_triggered/__init__.py

@@ -3,5 +3,6 @@
 import azure.functions as func
 
 
+# Retrieve the event data from storage blob and return it as Http response
 def main(req: func.HttpRequest, testEntities):
     return func.HttpResponse(status_code=200, body=testEntities)

tests/endtoend/eventhub_batch_functions/get_eventhub_batch_triggered/function.json

@@ -1,27 +1,27 @@
-{
-  "scriptFile": "__init__.py",
-  "bindings": [
-    {
-      "type": "httpTrigger",
-      "direction": "in",
-      "authLevel": "anonymous",
-      "methods": [
-        "get"
-      ],
-      "name": "req"
-    },
-    {
-      "direction": "in",
-      "type": "table",
-      "name": "testEntities",
-      "partitionKey": "WillBePopulated",
-      "tableName": "EventHubBatchTest",
-      "connection": "AzureWebJobsStorage"
-    },
-    {
-      "type": "http",
-      "direction": "out",
-      "name": "$return"
-    }
-  ]
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "type": "httpTrigger",
+      "direction": "in",
+      "authLevel": "anonymous",
+      "methods": [
+        "get"
+      ],
+      "name": "req"
+    },
+    {
+      "direction": "in",
+      "type": "table",
+      "name": "testEntities",
+      "partitionKey": "WillBePopulated",
+      "tableName": "EventHubBatchTest",
+      "connection": "AzureWebJobsStorage"
+    },
+    {
+      "type": "http",
+      "direction": "out",
+      "name": "$return"
+    }
+  ]
 }

tests/endtoend/eventhub_batch_functions/get_metadata_batch_triggered/__init__.py

@@ -0,0 +1,10 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import azure.functions as func
+
+
+# Retrieve the event data from storage blob and return it as Http response
+def main(req: func.HttpRequest, file: func.InputStream) -> str:
+    return func.HttpResponse(body=file.read().decode('utf-8'),
+                             status_code=200,
+                             mimetype='application/json')

tests/endtoend/eventhub_batch_functions/get_metadata_batch_triggered/function.json

@@ -0,0 +1,22 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "type": "httpTrigger",
+      "direction": "in",
+      "name": "req"
+    },
+    {
+      "type": "blob",
+      "direction": "in",
+      "name": "file",
+      "connection": "AzureWebJobsStorage",
+      "path": "python-worker-tests/test-metadata-batch-triggered.txt"
+    },
+    {
+      "type": "http",
+      "direction": "out",
+      "name": "$return"
+    }
+  ]
+}

tests/endtoend/eventhub_batch_functions/metadata_multiple/__init__.py

@@ -0,0 +1,24 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import typing
+import json
+import azure.functions as func
+
+
+# This is an actual EventHub trigger which handles Eventhub events in batches.
+# It serializes multiple event data into a json and store it into a blob.
+def main(events: typing.List[func.EventHubEvent]) -> bytes:
+    event_list = []
+    for event in events:
+        event_dict: typing.Mapping[str, typing.Any] = {
+            'body': event.get_body().decode('utf-8'),
+            'enqueued_time': event.enqueued_time.isoformat(),
+            'partition_key': event.partition_key,
+            'sequence_number': event.sequence_number,
+            'offset': event.offset,
+            'metadata': event.metadata
+        }
+        event_list.append(event_dict)
+
+    return json.dumps(event_list)