Skip to content

Aws iot core event listener driver

AwsIotCoreEventListenerDriver

Bases: BaseEventListenerDriver

Source code in griptape/drivers/event_listener/aws_iot_core_event_listener_driver.py 
15
16
17
18
19
20
21
22
23
@define
class AwsIotCoreEventListenerDriver(BaseEventListenerDriver):
    iot_endpoint: str = field(kw_only=True)
    topic: str = field(kw_only=True)
    session: boto3.Session = field(default=Factory(lambda: import_optional_dependency("boto3").Session()), kw_only=True)
    iotdata_client: Any = field(default=Factory(lambda self: self.session.client("iot-data"), takes_self=True))

    def try_publish_event_payload(self, event_payload: dict) -> None:
        self.iotdata_client.publish(topic=self.topic, payload=json.dumps(event_payload))

iot_endpoint: str = field(kw_only=True) class-attribute instance-attribute

iotdata_client: Any = field(default=Factory(lambda self: self.session.client('iot-data'), takes_self=True)) class-attribute instance-attribute

session: boto3.Session = field(default=Factory(lambda: import_optional_dependency('boto3').Session()), kw_only=True) class-attribute instance-attribute

topic: str = field(kw_only=True) class-attribute instance-attribute

try_publish_event_payload(event_payload)

Source code in griptape/drivers/event_listener/aws_iot_core_event_listener_driver.py 
22
23
def try_publish_event_payload(self, event_payload: dict) -> None:
    self.iotdata_client.publish(topic=self.topic, payload=json.dumps(event_payload))