feat: add support for google-cloud-pubsub closes #25

.mypy.ini

@@ -8,3 +8,6 @@ ignore_missing_imports = True
 
 [mypy-setuptools.*]
 ignore_missing_imports = True
+
+[mypy-google.cloud.*]
+ignore_missing_imports = True

README.md

@@ -214,6 +214,10 @@ Here is a list of built-in event handlers:
     * import from `fastapi_events.handlers.echo`
     * to forward events to stdout with `pprint`. Great for debugging purpose
 
+* `GoogleCloudSimplePubSubHandler`:
+    * import from `fastapi_events.handlers.gcp`
+    * to publish events to a single pubsub topic
+
 # Creating your own handler
 
 Creating your own handler is nothing more than inheriting from the `BaseEventHandler` class

fastapi_events/handlers/gcp.py

@@ -0,0 +1,51 @@
+import json
+from typing import Any, Callable, Dict, Iterable
+
+from google.cloud import pubsub_v1
+
+from fastapi_events.errors import ConfigurationError
+from fastapi_events.handlers.base import BaseEventHandler
+from fastapi_events.typing import Event
+
+
+def _json_serializer(event: Event) -> str:
+    return json.dumps(event, default=str)
+
+
+class GoogleCloudSimplePubSubHandler(BaseEventHandler):
+    def __init__(
+        self,
+        project_id: str,
+        topic_id: str,
+        max_batch_size: int = 1000,  # GCP Pubsub's maximum supported batch size
+        batch_settings_kwargs: Dict[str, Any] = None,
+        serializer: Callable[[Event], str] = None,
+    ) -> None:
+        """Google cloud simple PubSub handler. Publishes events to a single topic."""
+
+        if max_batch_size > 1000:
+            raise ConfigurationError("GCP Pubsub batch size limit is 1000.")
+
+        if serializer is not None and not callable(serializer):
+            raise ConfigurationError("serializer must be of type Callable")
+
+        self._max_batch_size = max_batch_size
+
+        # Publish messages as soon as there are max_messages
+        # or 1 second is passed
+        self._batch_settings = pubsub_v1.types.BatchSettings(
+            max_messages=self._max_batch_size, **batch_settings_kwargs
+        )
+        self._client = pubsub_v1.PublisherClient(self._batch_settings)
+        self._serializer = serializer or _json_serializer
+        self._topic_path = self._client.topic_path(project_id, topic_id)
+
+    async def handle_many(self, events: Iterable[Event]) -> None:
+        for event in events:
+            await self.handle(event)
+
+    async def handle(self, event: Event) -> None:
+        self._client.publish(self._topic_path, self.format_message(event))
+
+    def format_message(self, event: Event) -> bytes:
+        return self._serializer(event).encode("utf-8")

setup.py

@@ -4,11 +4,13 @@
 
 
 def get_version():
-    package_init = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'fastapi_events', '__init__.py')
+    package_init = os.path.join(
+        os.path.abspath(os.path.dirname(__file__)), "fastapi_events", "__init__.py"
+    )
     with open(package_init) as f:
         for line in f:
-            if line.startswith('__version__ ='):
-                return line.split('=')[1].strip().strip('"\'')
+            if line.startswith("__version__ ="):
+                return line.split("=")[1].strip().strip("\"'")
 
 
 def get_long_description():
@@ -40,11 +42,9 @@ def get_long_description():
         "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
     },
-    python_requires='>=3.7',
+    python_requires=">=3.7",
     keywords=["starlette", "fastapi", "pydantic"],
-    install_requires=[
-        "starlette"
-    ],
+    install_requires=["starlette"],
     extras_require={
         "test": [
             "requests",
@@ -56,10 +56,10 @@ def get_long_description():
             "pytest-mypy>=0.9.1",
             "moto[sqs]==2.2",
             "flake8>=3.9.2",
-            "pydantic>=1.5.0"
+            "pydantic>=1.5.0",
+            "google-cloud-pubsub>=2.13.6",
         ],
-        "aws": [
-            "boto3>=1.14"
-        ]
-    }
+        "aws": ["boto3>=1.14"],
+        "google": ["google-cloud-pubsub>=2.13.6"],
+    },
 )

tests/handlers/test_gcp_handler.py

@@ -0,0 +1,48 @@
+from unittest.mock import Mock, patch
+
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.testclient import TestClient
+
+from fastapi_events.dispatcher import dispatch
+from fastapi_events.handlers.gcp import GoogleCloudSimplePubSubHandler
+from fastapi_events.middleware import EventHandlerASGIMiddleware
+
+
+@patch("google.cloud.pubsub_v1.PublisherClient")
+def test_gcp_pubsub_handler(mock_publisher_client: Mock):
+
+    topic_id = "gcp-topic-id"
+    project_id = "gcp-project-id"
+
+    def setup_app():
+        app = Starlette(
+            middleware=[
+                Middleware(
+                    EventHandlerASGIMiddleware,
+                    handlers=[
+                        GoogleCloudSimplePubSubHandler(
+                            project_id=project_id,
+                            topic_id=topic_id,
+                            batch_settings_kwargs={"max_latency": 1},
+                        )
+                    ],
+                )
+            ]
+        )
+
+        @app.route("/")
+        async def root(request: Request) -> JSONResponse:
+            for idx in range(50):
+                dispatch(event_name="event", payload={"idx": idx + 1})
+            return JSONResponse([])
+
+        return app
+
+    app = setup_app()
+    client = TestClient(app)
+    client.get("/")
+
+    assert mock_publisher_client.return_value.publish.call_count == 50