Add Cloud Events support for #55

setup.py

@@ -52,6 +52,7 @@
         "click>=7.0,<8.0",
         "watchdog>=0.10.0",
         "gunicorn>=19.2.0,<21.0; platform_system!='Windows'",
+        "cloudevents<1.0",
     ],
     extras_require={"test": ["pytest", "tox"]},
     entry_points={

src/functions_framework/__init__.py

@@ -12,8 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import functools
+from enum import Enum
 import importlib.util
+import io
+import json
 import os.path
 import pathlib
 import sys
@@ -31,10 +33,20 @@
 )
 from google.cloud.functions.context import Context
 
+
+import cloudevents.sdk.event
+import cloudevents.sdk
+
 DEFAULT_SOURCE = os.path.realpath("./main.py")
 DEFAULT_SIGNATURE_TYPE = "http"
 
 
+class _EventType(Enum):
+    LEGACY = 1
+    CLOUD_EVENT_BINARY = 2
+    CLOUD_EVENT_TEXT = 3
+
+
 class _Event(object):
     """Event passed to background functions."""
 
@@ -67,38 +79,81 @@ def view_func(path):
     return view_func
 
 
-def _is_binary_cloud_event(request):
-    return (
+def _get_cloud_event_version():
+    return cloudevents.sdk.event.v1.Event()
+
+
+def _run_legacy_event(function, request):
+    event_data = request.get_json()
+    if not event_data:
+        flask.abort(400)
+    event_object = _Event(**event_data)
+    data = event_object.data
+    context = Context(**event_object.context)
+    function(data, context)
+
+
+def _run_binary_cloud_event(function, request, cloud_event_def):
+    data = io.BytesIO(request.get_data())
+    http_marshaller = cloudevents.sdk.marshaller.NewDefaultHTTPMarshaller()
+    event = http_marshaller.FromRequest(
+        cloud_event_def, request.headers, data, json.load
+    )
+
+    function(event)
+
+
+def _run_text_cloud_event(function, request, cloud_event_def):
+    data = io.StringIO(request.get_data(as_text=True))
+    m = cloudevents.sdk.marshaller.NewDefaultHTTPMarshaller()
+    event = m.FromRequest(cloud_event_def, request.headers, data, json.loads)
+    function(event)
+
+
+def _get_event_type(request):
+    if (
         request.headers.get("ce-type")
         and request.headers.get("ce-specversion")
         and request.headers.get("ce-source")
         and request.headers.get("ce-id")
-    )
+    ):
+        return _EventType.CLOUD_EVENT_BINARY
+    elif request.headers.get("Content-Type") == "application/cloudevents+json":
+        return _EventType.CLOUD_EVENT_TEXT
+    else:
+        return _EventType.LEGACY
 
 
 def _event_view_func_wrapper(function, request):
     def view_func(path):
-        if _is_binary_cloud_event(request):
-            # Support CloudEvents in binary content mode, with data being the
-            # whole request body and context attributes retrieved from request
-            # headers.
-            data = request.get_data()
-            context = Context(
-                eventId=request.headers.get("ce-eventId"),
-                timestamp=request.headers.get("ce-timestamp"),
-                eventType=request.headers.get("ce-eventType"),
-                resource=request.headers.get("ce-resource"),
+        if _get_event_type(request) == _EventType.LEGACY:
+            _run_legacy_event(function, request)
+        else:
+            # here for defensive backwards compatibility in case we make a mistake in rollout.
+            raise InvalidConfigurationException(
+                "The FUNCTION_SIGNATURE_TYPE for this function is set to event "
+                "but no legacy event was given. If you are using CloudEvents set "
+                "FUNCTION_SIGNATURE_TYPE=cloudevent"
             )
-            function(data, context)
+
+        return "OK"
+
+    return view_func
+
+
+def _cloudevent_view_func_wrapper(function, request):
+    def view_func(path):
+        cloud_event_def = _get_cloud_event_version()
+        event_type = _get_event_type(request)
+        if event_type == _EventType.CLOUD_EVENT_TEXT:
+            _run_text_cloud_event(function, request, cloud_event_def)
+        elif event_type == _EventType.CLOUD_EVENT_BINARY:
+            _run_binary_cloud_event(function, request, cloud_event_def)
         else:
-            # This is a regular CloudEvent
-            event_data = request.get_json()
-            if not event_data:
-                flask.abort(400)
-            event_object = _Event(**event_data)
-            data = event_object.data
-            context = Context(**event_object.context)
-            function(data, context)
+            raise InvalidConfigurationException(
+                "Function was defined with FUNCTION_SIGNATURE_TYPE=cloudevent "
+                " but it did not receive a cloudevent as a request."
+            )
 
         return "OK"
 
@@ -179,19 +234,27 @@ def create_app(target=None, source=None, signature_type=None):
         app.url_map.add(werkzeug.routing.Rule("/<path:path>", endpoint="run"))
         app.view_functions["run"] = _http_view_func_wrapper(function, flask.request)
         app.view_functions["error"] = lambda: flask.abort(404, description="Not Found")
-    elif signature_type == "event":
+    elif signature_type == "event" or signature_type == "cloudevent":
         app.url_map.add(
             werkzeug.routing.Rule(
-                "/", defaults={"path": ""}, endpoint="run", methods=["POST"]
+                "/", defaults={"path": ""}, endpoint=signature_type, methods=["POST"]
             )
         )
         app.url_map.add(
-            werkzeug.routing.Rule("/<path:path>", endpoint="run", methods=["POST"])
+            werkzeug.routing.Rule(
+                "/<path:path>", endpoint=signature_type, methods=["POST"]
+            )
         )
-        app.view_functions["run"] = _event_view_func_wrapper(function, flask.request)
+
         # Add a dummy endpoint for GET /
         app.url_map.add(werkzeug.routing.Rule("/", endpoint="get", methods=["GET"]))
         app.view_functions["get"] = lambda: ""
+
+        # Add the view functions
+        app.view_functions["event"] = _event_view_func_wrapper(function, flask.request)
+        app.view_functions["cloudevent"] = _cloudevent_view_func_wrapper(
+            function, flask.request
+        )
     else:
         raise FunctionsFrameworkException(
             "Invalid signature type: {signature_type}".format(

tests/test_cloudevent_functions.py

@@ -0,0 +1,108 @@
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import json
+import pathlib
+import pytest
+import cloudevents.sdk
+import cloudevents.sdk.event.v1
+import cloudevents.sdk.event.v03
+import cloudevents.sdk.marshaller
+
+from functions_framework import LazyWSGIApp, create_app, exceptions
+
+TEST_FUNCTIONS_DIR = pathlib.Path(__file__).resolve().parent / "test_functions"
+
+# Python 3.5: ModuleNotFoundError does not exist
+try:
+    _ModuleNotFoundError = ModuleNotFoundError
+except:
+    _ModuleNotFoundError = ImportError
+
+
+@pytest.fixture
+def event_1_10():
+    event = (
+        cloudevents.sdk.event.v1.Event()
+        .SetContentType("application/json")
+        .SetData('{"name":"john"}')
+        .SetEventID("my-id")
+        .SetSource("from-galaxy-far-far-away")
+        .SetEventTime("tomorrow")
+        .SetEventType("cloudevent.greet.you")
+    )
+    return event
+
+
+@pytest.fixture
+def event_0_3():
+    event = (
+        cloudevents.sdk.event.v03.Event()
+        .SetContentType("application/json")
+        .SetData('{"name":"john"}')
+        .SetEventID("my-id")
+        .SetSource("from-galaxy-far-far-away")
+        .SetEventTime("tomorrow")
+        .SetEventType("cloudevent.greet.you")
+    )
+    return event
+
+
+def test_event_1_0(event_1_10):
+    source = TEST_FUNCTIONS_DIR / "cloudevents" / "main.py"
+    target = "function"
+
+    client = create_app(target, source, "cloudevent").test_client()
+
+    m = cloudevents.sdk.marshaller.NewDefaultHTTPMarshaller()
+    structured_headers, structured_data = m.ToRequest(
+        event_1_10, cloudevents.sdk.converters.TypeStructured, json.dumps
+    )
+
+    resp = client.post("/", headers=structured_headers, data=structured_data.getvalue())
+    assert resp.status_code == 200
+    assert resp.data == b"OK"
+
+
+def test_binary_event_1_0(event_1_10):
+    source = TEST_FUNCTIONS_DIR / "cloudevents" / "main.py"
+    target = "function"
+
+    client = create_app(target, source, "cloudevent").test_client()
+
+    m = cloudevents.sdk.marshaller.NewDefaultHTTPMarshaller()
+
+    binary_headers, binary_data = m.ToRequest(
+        event_1_10, cloudevents.sdk.converters.TypeBinary, json.dumps
+    )
+
+    resp = client.post("/", headers=binary_headers, data=binary_data)
+
+    assert resp.status_code == 200
+    assert resp.data == b"OK"
+
+
+def test_event_0_3(event_0_3):
+    source = TEST_FUNCTIONS_DIR / "cloudevents" / "main.py"
+    target = "function"
+
+    client = create_app(target, source, "cloudevent").test_client()
+
+    m = cloudevents.sdk.marshaller.NewDefaultHTTPMarshaller()
+    structured_headers, structured_data = m.ToRequest(
+        event_0_3, cloudevents.sdk.converters.TypeStructured, json.dumps
+    )
+
+    resp = client.post("/", headers=structured_headers, data=structured_data.getvalue())
+    assert resp.status_code == 200
+    assert resp.data == b"OK"