CloudEvents/EG Events must recognize magic events

sdk/core/azure-core/azure/core/messaging.py

@@ -5,6 +5,7 @@
 # license information.
 # --------------------------------------------------------------------------
 import uuid
+import json
 from base64 import b64decode
 from datetime import datetime
 from .utils._utils import _convert_to_isoformat, TZ_UTC
@@ -21,8 +22,28 @@
 
 __all__ = ["CloudEvent"]
 
-
-class CloudEvent(object):  # pylint:disable=too-many-instance-attributes
+class _EventMixin(object):
+    """Event mixin to have methods that are common to different Event types
+    like CloudEvent, EventGridEvent etc.
+    """
+    @staticmethod
+    def _get_bytes(obj):
+        # type: (Any) -> Dict
+        try:
+            # storage queue
+            return json.loads(obj.content)
+        except AttributeError:
+            # eventhubs
+            try:
+                return json.loads(next(obj.body))[0]
+            except KeyError:
+                # servicebus
+                return json.loads(next(obj.body))
+            except:
+                return obj
+
+
+class CloudEvent(_EventMixin):  # pylint:disable=too-many-instance-attributes
     """Properties of the CloudEvent 1.0 Schema.
     All required parameters must be populated in order to send to Azure.
 
@@ -122,6 +143,7 @@ def from_dict(cls, event):
         :type event: dict
         :rtype: CloudEvent
         """
+        event = CloudEvent._get_bytes(event)
         kwargs = {}  # type: Dict[Any, Any]
         reserved_attr = [
             "data",

sdk/eventgrid/azure-eventgrid/azure/eventgrid/_models.py

@@ -3,16 +3,18 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 # pylint:disable=protected-access
+from _typeshed import Self
 from typing import Any
 import datetime as dt
 import uuid
+from azure.core.messaging import _EventMixin
 from msrest.serialization import UTC
 from ._generated.models import (
     EventGridEvent as InternalEventGridEvent,
 )
 
 
-class EventGridEvent(InternalEventGridEvent):
+class EventGridEvent(InternalEventGridEvent, _EventMixin):
     """Properties of an event published to an Event Grid topic using the EventGrid Schema.
 
     Variables are only populated by the server, and will be ignored when sending a request.
@@ -96,3 +98,8 @@ def __repr__(self):
         return "EventGridEvent(subject={}, event_type={}, id={}, event_time={})".format(
             self.subject, self.event_type, self.id, self.event_time
         )[:1024]
+
+    @classmethod
+    def from_dict(cls, event):
+        event = EventGridEvent._get_bytes(event)
+        super(EventGridEvent).from_dict(event)

sdk/eventgrid/azure-eventgrid/samples/consume_samples/consume_cloud_events_from_eventhub.py

@@ -24,11 +24,9 @@
 
 CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
 EVENTHUB_NAME = os.environ["EVENT_HUB_NAME"]
-
-
 def on_event(partition_context, event):
-    dict_event = CloudEvent.from_dict(json.loads(event)[0])
-    print("data: {}\n".format(deserialized_event.data))
+    dict_event = CloudEvent.from_dict(event)
+    print("data: {}\n".format(dict_event.data))
 
 consumer_client = EventHubConsumerClient.from_connection_string(
     conn_str=CONNECTION_STR,

sdk/eventgrid/azure-eventgrid/samples/consume_samples/consume_cloud_events_from_storage_queue.py

@@ -27,10 +27,10 @@
     payload =  qsc.get_queue_client(
         queue=queue_name,
         message_decode_policy=BinaryBase64DecodePolicy()
-        ).peek_messages()
+        ).peek_messages(max_messages=32)
 
     ## deserialize payload into a list of typed Events
-    events = [CloudEvent.from_dict(json.loads(msg.content)) for msg in payload]
+    events = [CloudEvent.from_dict(msg) for msg in payload]
 
     for event in events:
         print(type(event)) ## CloudEvent

sdk/eventgrid/azure-eventgrid/samples/consume_samples/consume_eventgrid_events_from_service_bus_queue.py

@@ -30,7 +30,7 @@
     payload =  sb_client.get_queue_receiver(queue_name).receive_messages()
 
     ## deserialize payload into a list of typed Events
-    events = [EventGridEvent.from_dict(json.loads(next(msg.body).decode('utf-8'))) for msg in payload]
+    events = [EventGridEvent.from_dict(msg) for msg in payload]
 
     for event in events:
         print(type(event)) ## EventGridEvent