Merge pull request #11 from octue/fix/endpoint-rountrip-test

Add missing parameter to example event handler

Author: thclark

.github/workflows/ci.yml

@@ -42,7 +42,7 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        python: ["3.8", "3.9", "3.10"]
+        python: ["3.9", "3.10"]
         os: [ubuntu-latest] # [ubuntu-latest, windows-latest, macos-latest] for full coverage but this gets expensive quickly
     runs-on: ${{ matrix.os }}
 

django_gcp/events/utils.py

@@ -1,4 +1,8 @@
+import base64
+import json
 import logging
+from datetime import timezone
+from dateutil.parser import isoparse
 from django.conf import settings
 from django.urls import reverse
 from django.utils.http import urlencode
@@ -7,6 +11,15 @@
 logger = logging.getLogger(__name__)
 
 
+def _make_naive_utc(value):
+    """Converts a timezone-aware datetime.datetime to UTC then makes it naive.
+    Used for strictly formatting the UTC-in-nanoseconds publish time of pub/sub messages
+    """
+    if value.tzinfo is not None:
+        value = value.astimezone(timezone.utc).replace(tzinfo=None)
+    return value
+
+
 def get_event_url(event_kind, event_reference, event_parameters=None, url_namespace="gcp-events", base_url=None):
     """Returns a fully constructed url for the events endpoint, suitable for receipt and processing of events
     :param str event_kind: The kind of the event (must be url-safe)
@@ -33,3 +46,74 @@ def get_event_url(event_kind, event_reference, event_parameters=None, url_namesp
     logger.debug("Generated webhook endpoitn url %s", url)
 
     return url
+
+
+def make_pubsub_message(
+    data,
+    attributes=None,
+    message_id=None,
+    ordering_key=None,
+    publish_time=None,
+):
+    """Make a json-encodable message replicating the GCP Pub/Sub v1 format
+
+    For more details see: https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage
+
+    :param Union[dict, list] data: JSON-serialisable data to form the body of the message
+    :param Union[dict, None] attributes: Dict of attributes to attach to the message. Contents must be flat, containing only string keys with string values.
+    :param Union[str, None] message_id: An optional id for the message.
+    :param Union[str, None] ordering_key: A string used to order messages.
+    :param Union[datetime, None] publish_time: If sending a message to PubSub, this will be set by the server on receipt so generally should be left as `None`. However, for the purposes of mocking messages for testing, supply a python datetime specifying the publish time of the message, which will be converted to a string timestamp with nanosecond accuracy.
+    :return dict: A dict containing a fully composed PubSub message
+
+    """
+    out = dict()
+
+    out["data"] = base64.b64encode(json.dumps(data).encode()).decode()
+
+    if publish_time is not None:
+        publish_time_utc_naive = _make_naive_utc(publish_time)
+        iso_us = publish_time_utc_naive.isoformat()
+        iso_ns = f"{iso_us}000Z"
+        out["publishTime"] = iso_ns
+
+    if attributes is not None:
+        # Check all attributes are k-v pairs of strings
+        for k, v in attributes.items():
+            if k.__class__ != str:
+                raise ValueError("All attribute keys must be strings")
+            if v.__class__ != str:
+                raise ValueError("All attribute values must be strings")
+        out["attributes"] = attributes
+
+    if message_id is not None:
+        if message_id.__class__ != str:
+            raise ValueError("The message_id, if given, must be a string")
+        out["messageId"] = message_id
+
+    if ordering_key is not None:
+        if ordering_key.__class__ != str:
+            raise ValueError("The ordering_key, if given, must be a string")
+        out["orderingKey"] = ordering_key
+
+    return out
+
+
+def decode_pubsub_message(message):
+    """Decode data within a pubsub message
+    :parameter dict message: The Pub/Sub message, which should already be decoded from a raw JSON string to a dict.
+    :return: None
+    """
+
+    decoded = {
+        "data": json.loads(base64.b64decode(message["data"])),
+        "attributes": message.get("attributes", None),
+        "message_id": message.get("messageId", None),
+        "ordering_key": message.get("orderingKey", None),
+        "publish_time": message.get("publishTime", None),
+    }
+
+    if decoded["publish_time"] is not None:
+        decoded["publish_time"] = isoparse(decoded["publish_time"])
+
+    return decoded

django_gcp/events/views.py

@@ -1,5 +1,6 @@
 import json
 import logging
+from django.conf import settings
 from django.http import HttpResponse
 from django.utils.decorators import method_decorator
 from django.views.decorators.csrf import csrf_exempt
@@ -34,9 +35,12 @@ def post(self, request, event_kind, event_reference):
             return self._prepare_response(status=201, payload={})
 
         except Exception as e:  # pylint: disable=broad-except
-            msg = f"Unable to handle event of kind {event_kind} with reference {event_reference}"
-            logger.warning("%s. Exception: %s", msg, str(e))
-            return self._prepare_response(status=400, payload={"error": msg})
+            if getattr(settings, "DEBUG", False):
+                raise e
+            else:
+                msg = f"Unable to handle event of kind {event_kind} with reference {event_reference}"
+                logger.warning("%s. Exception: %s", msg, str(e))
+                return self._prepare_response(status=400, payload={"error": msg})
 
     def _prepare_response(self, status, payload):
         return HttpResponse(status=status, content=json.dumps(payload), content_type="application/json")

docs/source/events.rst

@@ -58,13 +58,14 @@ This is how you attach your handler. In ``your-app/signals.py`` file, do:
    import logging
    from django.dispatch import receiver
    from django_gcp.events.signals import event_received
+   from django_gcp.events.utils import decode_pubsub_message
 
 
    logger = logging.getLogger(__name__)
 
 
    @receiver(event_received)
-   def receive_event(sender, event_kind, event_reference, event_payload):
+   def receive_event(sender, event_kind, event_reference, event_payload, event_parameters):
        """Handle question updates received via pubsub
        :param event_kind (str): A kind/variety allowing you to determine the handler to use (eg "something-update"). Required.
        :param event_reference (str): A reference value provided by the client allowing events to be sorted/filtered. Required.
@@ -78,7 +79,13 @@ This is how you attach your handler. In ``your-app/signals.py`` file, do:
        if event_kind is "something-important":
            # Here is where you handle the event using whatever logic you want
            # CAREFUL: See the tip above about authentication (verifying the payload is not malicious)
-           print("DO SOMETHING IMPORTANT")
+           print("DO SOMETHING IMPORTANT WITH THE PAYLOAD:", event_payload)
+           #
+           # Your payload can be from any arbitrary source, and is in the form of decoded json.
+           # However, if the source is Eventarc or Pub/Sub, the payload contains a formatted message
+           # with base64 encoded data; we provide a utility to further decode this into something sensible:
+           message = decode_pubsub_message(event_payload)
+           print("DECODED PUBSUB MESSAGE:" message)
 
 .. tip::
 
@@ -131,6 +138,35 @@ It generates absolute URLs by default, because integration with external systems
       )
 
 
+Generating and Consuming Pub/Sub Messages
+-----------------------------------------
+
+When hooked up to GCP Pub/Sub or eventarc, the event payload is in the form of a Pub/Sub message.
+
+These messages have a specific format (see https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage).
+
+To allow you to interact directly with Pub/Sub (i.e. publish messages to a topic), or for the purposes of testing your signals,
+``django-gcp`` includes a `make_pubsub_message` utility that provides an easy and pythonic way of constructing a Pub/Sub message.
+
+For example, to test the signal receiver above with a replica of a real pubsub message payload, you might do:
+
+.. code-block:: python
+
+    from django_gcp.events.utils import make_pubsub_message
+    from datetime import datetime
+
+    class YourTests(TestCase):
+        def test_your_code_handles_a_payload_from_pubsub(self):
+            payload = make_pubsub_message({"my": "data"}, publish_time=datetime.now())
+
+            response = self.client.post(
+                reverse("gcp-events", args=["the-event-kind", "the-event-reference"]),
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+
+        self.assertEqual(response.status_code, 201)
+
 
 Exception Handling
 ------------------