feat: pubsub triggers

Author: Salakar

src/firebase_functions/pubsub.py

@@ -0,0 +1,81 @@
+"""
+Cloud functions to handle events from Google Cloud Pub/Sub.
+"""
+
+import dataclasses as _dataclasses
+import functools as _functools
+import typing as _typing
+import json as _json
+import base64 as _base64
+
+import firebase_functions.options as _options
+import firebase_functions.private.util as _util
+from firebase_functions.core import CloudEvent, T
+
+
+@_dataclasses.dataclass(frozen=True)
+class Message(_typing.Generic[T]):
+    """
+    Interface representing a Google Cloud Pub/Sub message.
+    """
+
+    message_id: str
+    """
+    Autogenerated ID that uniquely identifies this message.
+    """
+
+    publish_time: str
+    """
+    Time the message was published.
+    """
+
+    attributes: dict[str, str]
+    """
+    User-defined attributes published with the message, if any.
+    """
+
+    data: str
+    """
+    The data payload of this message object as a base64-encoded string.
+    """
+
+    ordering_key: str
+    """
+    User-defined key used to ensure ordering amongst messages with the same key.
+    """
+
+    @property
+    def json(self) -> _typing.Optional[_core.T]:
+        try:
+            if self.data is not None:
+                return _json.loads(_base64.b64decode(self.data).decode("utf-8"))
+            else:
+                return None
+        except Exception as error:
+            raise Exception(
+                f"Unable to parse Pub/Sub message data as JSON: {error}"
+            ) from error
+
+
+@_dataclasses.dataclass(frozen=True)
+class MessagePublishedData(_typing.Generic[_core.T]):
+    """
+    The interface published in a Pub/Sub publish subscription.
+
+    'T' Type representing `Message.data`'s JSON format.
+    """
+    message: Message[_core.T]
+    """
+    Google Cloud Pub/Sub message.
+    """
+
+    subscription: str
+    """
+    A subscription resource.
+    """
+
+
+_E1 = DatabaseEvent[Change[_typing.Any | None]]
+_E2 = DatabaseEvent[_typing.Any | None]
+_C1 = _typing.Callable[[_E1], None]
+_C2 = _typing.Callable[[_E2], None]