added cloudevents to events-storage (#4537)

* added cloudevents to events-storage

Signed-off-by: Curtis Mason <[email protected]>

* modified README

Signed-off-by: Curtis Mason <[email protected]>

* nit lower case ce-subject fix

Signed-off-by: Curtis Mason <[email protected]>

Co-authored-by: Grant Timmerman <[email protected]>
Co-authored-by: Leah E. Cole <[email protected]>

Author: 3 people

run/events-storage/README.md

@@ -1,6 +1,6 @@
 # Events for Cloud Run – GCS tutorial
 
-This sample shows how to create a service that processes GCS events.
+This sample shows how to create a service that processes GCS using [the CloudEvents SDK](https://github.com/cloudevents/sdk-python).
 
 ## Setup
 
@@ -57,13 +57,13 @@ gcloud alpha events triggers create $MY_GCS_TRIGGER \
 
 ## Test
 
-Test your Cloud Run service by publishing a message to the topic: 
+Test your Cloud Run service by publishing a message to the topic:
 
 ```sh
 gsutil defstorageclass set STANDARD gs://$MY_GCS_BUCKET
 ```
 
-Observe the Cloud Run service printing upon receiving an event in 
+Observe the Cloud Run service printing upon receiving an event in
 Cloud Logging:
 
 ```sh

run/events-storage/main.py

@@ -14,6 +14,9 @@
 # [START run_events_gcs_handler]
 import os
 
+import cloudevents.exceptions as cloud_exceptions
+from cloudevents.http import from_http
+
 from flask import Flask, request
 
 
@@ -23,20 +26,29 @@
 
 @app.route('/', methods=['POST'])
 def index():
-    for field in required_fields:
-        if field not in request.headers:
-            errmsg = f'Bad Request: missing required header {field}'
-            print(errmsg)
-            return errmsg, 400
-
-    if 'Ce-Subject' not in request.headers:
-        errmsg = 'Bad Request: expected header Ce-Subject'
+    # Create CloudEvent from HTTP headers and body
+    try:
+        event = from_http(request.headers, request.get_data())
+
+    except cloud_exceptions.MissingRequiredFields as e:
+        print(f"cloudevents.exceptions.MissingRequiredFields: {e}")
+        return "Failed to find all required cloudevent fields. ", 400
+
+    except cloud_exceptions.InvalidStructuredJSON as e:
+        print(f"cloudevents.exceptions.InvalidStructuredJSON: {e}")
+        return "Could not deserialize the payload as JSON. ", 400
+
+    except cloud_exceptions.InvalidRequiredFields as e:
+        print(f"cloudevents.exceptions.InvalidRequiredFields: {e}")
+        return "Request contained invalid required cloudevent fields. ", 400
+
+    if 'subject' not in event:
+        errmsg = 'Bad Request: expected header ce-subject'
         print(errmsg)
         return errmsg, 400
 
-    ce_subject = request.headers.get('Ce-Subject')
-    print(f'GCS CloudEvent type: {ce_subject}')
-    return (f'GCS CloudEvent type: {ce_subject}', 200)
+    print(f"GCS CloudEvent type: {event['subject']}")
+    return (f"GCS CloudEvent type: {event['subject']}", 200)
 # [END run_events_gcs_handler]
 
 

run/events-storage/main_test.py

@@ -20,9 +20,12 @@
 import main
 
 
-required_fields = ['Ce-Id', 'Ce-Source', 'Ce-Type', 'Ce-Specversion']
-
-header_data = {field: str(uuid4()) for field in required_fields}
+binary_headers = {
+    "ce-id": str(uuid4),
+    "ce-type": "com.pytest.sample.event",
+    "ce-source": "<my-test-source>",
+    "ce-specversion": "1.0"
+}
 
 
 @pytest.fixture
@@ -32,7 +35,7 @@ def client():
 
 
 def test_endpoint(client, capsys):
-    test_headers = copy.copy(header_data)
+    test_headers = copy.copy(binary_headers)
     test_headers['Ce-Subject'] = 'test-subject'
 
     r = client.post('/', headers=test_headers)
@@ -43,20 +46,20 @@ def test_endpoint(client, capsys):
 
 
 def test_missing_subject(client, capsys):
-    r = client.post('/', headers=header_data)
+    r = client.post('/', headers=binary_headers)
     assert r.status_code == 400
 
     out, _ = capsys.readouterr()
-    assert 'Bad Request: expected header Ce-Subject' in out
+    assert 'Bad Request: expected header ce-subject' in out
 
 
 def test_missing_required_fields(client, capsys):
-    for field in required_fields:
-        test_headers = copy.copy(header_data)
+    for field in binary_headers:
+        test_headers = copy.copy(binary_headers)
         test_headers.pop(field)
 
         r = client.post('/', headers=test_headers)
         assert r.status_code == 400
 
         out, _ = capsys.readouterr()
-        assert f'Bad Request: missing required header {field}' in out
+        assert 'MissingRequiredFields' in out

run/events-storage/requirements-test.txt

@@ -1,2 +1 @@
 pytest==6.0.1
-

run/events-storage/requirements.txt

@@ -1,2 +1,3 @@
 Flask==1.1.2
 gunicorn==20.0.4
+cloudevents==1.2.0