Feature/pusher event driver (#821)

Co-authored-by: Collin Dutter <[email protected]>

.github/workflows/docs-integration-tests.yml

@@ -112,6 +112,10 @@ jobs:
       AWS_IOT_CORE_ENDPOINT: ${{ secrets.INTEG_AWS_IOT_CORE_ENDPOINT }}
       AWS_IOT_CORE_TOPIC: ${{ secrets.INTEG_AWS_IOT_CORE_TOPIC }}
       ELEVEN_LABS_API_KEY: ${{ secrets.INTEG_ELEVEN_LABS_API_KEY }}
+      PUSHER_APP_ID: ${{ secrets.INTEG_PUSHER_APP_ID }}
+      PUSHER_KEY: ${{ secrets.INTEG_PUSHER_KEY }}
+      PUSHER_SECRET: ${{ secrets.INTEG_PUSHER_SECRET }}
+      PUSHER_CLUSTER: ${{ secretes.INTEG_PUSHER_CLUSTER }}
     services:
       postgres:
         image: ankane/pgvector:v0.5.0

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `AudioTranscriptionTask` and `AudioTranscriptionClient` for transcribing audio content in Structures.
 - `OpenAiAudioTranscriptionDriver` for integration with OpenAI's speech-to-text models, including Whisper.
 - Parameter `env` to `BaseStructureRunDriver` to set environment variables for a Structure Run.
+- `PusherEventListenerDriver` to enable sending of framework events over a Pusher WebSocket. 
 
 ### Changed
 - **BREAKING**: Updated OpenAI-based image query drivers to remove Vision from the name.

docs/griptape-framework/drivers/event-listener-drivers.md

@@ -212,4 +212,39 @@ agent = Agent(
 
 agent.run("Analyze the pros and cons of remote work vs. office work")
 ```
+### Pusher Event Listener Driver
 
+!!! info
+    This driver requires the `drivers-event-listener-pusher` [extra](../index.md#extras).
+
+The [PusherEventListenerDriver](../../reference/griptape/drivers/event_listener/pusher_event_listener_driver.md) sends Events to [Pusher](https://pusher.com).
+
+```python
+import os
+from griptape.drivers import PusherEventListenerDriver
+from griptape.events import (
+    EventListener,
+    FinishStructureRunEvent
+)
+from griptape.structures import Agent
+
+agent = Agent(
+    event_listeners=[
+        EventListener(
+            event_types=[FinishStructureRunEvent],
+            driver=PusherEventListenerDriver(
+                batched=False,
+                app_id=os.environ["PUSHER_APP_ID"],
+                key=os.environ["PUSHER_KEY"],
+                secret=os.environ["PUSHER_SECRET"],
+                cluster=os.environ["PUSHER_CLUSTER"],
+                channel='my-channel',
+                event_name='my-event'
+            ),
+        ),
+    ],
+)
+
+agent.run("Analyze the pros and cons of remote work vs. office work")
+
+```

griptape/drivers/__init__.py

@@ -93,6 +93,7 @@
 from .event_listener.webhook_event_listener_driver import WebhookEventListenerDriver
 from .event_listener.aws_iot_core_event_listener_driver import AwsIotCoreEventListenerDriver
 from .event_listener.griptape_cloud_event_listener_driver import GriptapeCloudEventListenerDriver
+from .event_listener.pusher_event_listener_driver import PusherEventListenerDriver
 
 from .file_manager.base_file_manager_driver import BaseFileManagerDriver
 from .file_manager.local_file_manager_driver import LocalFileManagerDriver
@@ -193,6 +194,7 @@
     "WebhookEventListenerDriver",
     "AwsIotCoreEventListenerDriver",
     "GriptapeCloudEventListenerDriver",
+    "PusherEventListenerDriver",
     "BaseFileManagerDriver",
     "LocalFileManagerDriver",
     "AmazonS3FileManagerDriver",

griptape/drivers/event_listener/pusher_event_listener_driver.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from attrs import define, field, Factory
+from griptape.drivers import BaseEventListenerDriver
+from griptape.utils import import_optional_dependency
+
+if TYPE_CHECKING:
+    from pusher import Pusher
+
+
+@define
+class PusherEventListenerDriver(BaseEventListenerDriver):
+    app_id: str = field(kw_only=True)
+    key: str = field(kw_only=True)
+    secret: str = field(kw_only=True)
+    cluster: str = field(kw_only=True)
+    channel: str = field(kw_only=True)
+    event_name: str = field(kw_only=True)
+    pusher_client: Pusher = field(
+        default=Factory(
+            lambda self: import_optional_dependency("pusher").Pusher(
+                app_id=self.app_id, key=self.key, secret=self.secret, cluster=self.cluster, ssl=True
+            ),
+            takes_self=True,
+        ),
+        kw_only=True,
+    )
+
+    def try_publish_event_payload_batch(self, event_payload_batch: list[dict]) -> None:
+        data = [
+            {"channel": self.channel, "name": self.event_name, "data": event_payload}
+            for event_payload in event_payload_batch
+        ]
+
+        self.pusher_client.trigger_batch(data)
+
+    def try_publish_event_payload(self, event_payload: dict) -> None:
+        self.pusher_client.trigger(channels=self.channel, event_name=self.event_name, data=event_payload)

pyproject.toml

@@ -53,6 +53,7 @@ markdownify = {version = "^0.11.6", optional = true}
 voyageai = {version = "^0.2.1", optional = true}
 elevenlabs = {version = "^1.1.2", optional = true}
 torch = {version = "^2.3.0", optional = true}
+pusher = {version = "^3.3.2", optional = true}
 
 # loaders
 pandas = {version = "^1.3", optional = true}
@@ -96,6 +97,7 @@ drivers-web-scraper-markdownify = ["playwright", "beautifulsoup4", "markdownify"
 
 drivers-event-listener-amazon-sqs = ["boto3"]
 drivers-event-listener-amazon-iot = ["boto3"]
+drivers-event-listener-pusher = ["pusher"]
 
 loaders-dataframe = ["pandas"]
 loaders-pdf = ["pypdf"]
@@ -128,6 +130,7 @@ all = [
     "voyageai",
     "elevenlabs",
     "torch",
+    "pusher",
 
     # loaders
     "pandas",

tests/unit/drivers/event_listener/test_pusher_event_listener_driver.py

@@ -0,0 +1,42 @@
+from pytest import fixture
+from tests.mocks.mock_event import MockEvent
+from griptape.drivers import PusherEventListenerDriver
+from unittest.mock import Mock
+
+
+class TestPusherEventListenerDriver:
+    @fixture(autouse=True)
+    def mock_post(self, mocker):
+        mock_pusher_client = mocker.patch("pusher.Pusher")
+        mock_pusher_client.return_value.trigger.return_value = Mock()
+        mock_pusher_client.return_value.trigger_batch.return_value = Mock()
+
+        return mock_pusher_client
+
+    @fixture()
+    def driver(self):
+        return PusherEventListenerDriver(
+            app_id="test-app-id",
+            key="test-key",
+            secret="test-secret",
+            cluster="test-cluster",
+            channel="test-channel",
+            event_name="test-event",
+        )
+
+    def test_init(self, driver):
+        assert driver
+
+    def test_try_publish_event_payload(self, driver):
+        data = MockEvent().to_dict()
+        driver.try_publish_event_payload(data)
+
+        driver.pusher_client.trigger.assert_called_with(channels="test-channel", event_name="test-event", data=data)
+
+    def test_try_publish_event_payload_batch(self, driver):
+        data = [MockEvent().to_dict() for _ in range(3)]
+        driver.try_publish_event_payload_batch(data)
+
+        driver.pusher_client.trigger_batch.assert_called_with(
+            [{"channel": "test-channel", "name": "test-event", "data": data[i]} for i in range(3)]
+        )