Skip to content

griptape_cloud

P = ParamSpec('P') module-attribute

T = TypeVar('T') module-attribute

GriptapeCloudStructure

Utility for working with Griptape Cloud Structures.

Attributes:

Name Type Description
_event_listener Optional[EventListener]

Event Listener to use. Defaults to an EventListener with a GriptapeCloudEventListenerDriver.
_observability Optional[Observability]

Observability to use. Defaults to an Observability with a GriptapeCloudObservabilityDriver.
observe bool

Whether to enable observability. Enabling requires the drivers-observability-griptape-cloud extra.
Source code in griptape/utils/griptape_cloud.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@define()
class GriptapeCloudStructure:
    """Utility for working with Griptape Cloud Structures.

    Attributes:
        _event_listener: Event Listener to use. Defaults to an EventListener with a GriptapeCloudEventListenerDriver.
        _observability: Observability to use. Defaults to an Observability with a GriptapeCloudObservabilityDriver.
        observe: Whether to enable observability. Enabling requires the `drivers-observability-griptape-cloud` extra.
    """

    _event_listener: Optional[EventListener] = field(default=None, kw_only=True, alias="event_listener")
    _observability: Optional[Observability] = field(default=None, kw_only=True, alias="observability")
    observe: bool = field(default=False, kw_only=True)
    _output: Optional[BaseArtifact] = field(default=None, init=False)

    @lazy_property()
    def event_listener(self) -> EventListener:
        from griptape.drivers.event_listener.griptape_cloud import GriptapeCloudEventListenerDriver

        return EventListener(event_listener_driver=GriptapeCloudEventListenerDriver())

    @lazy_property()
    def observability(self) -> Observability:
        from griptape.drivers.observability.griptape_cloud import GriptapeCloudObservabilityDriver
        from griptape.observability.observability import Observability

        return Observability(observability_driver=GriptapeCloudObservabilityDriver())

    @property
    def output(self) -> Optional[BaseArtifact]:
        return self._output

    @output.setter
    def output(self, value: BaseArtifact | Any) -> None:
        if isinstance(value, BaseArtifact):
            self._output = value
        else:
            if isinstance(value, list):
                self._output = ListArtifact([self._to_artifact(item) for item in value])
            else:
                self._output = self._to_artifact(value)

    @property
    def structure_run_id(self) -> str:
        return os.environ["GT_CLOUD_STRUCTURE_RUN_ID"]

    @property
    def in_managed_environment(self) -> bool:
        return "GT_CLOUD_STRUCTURE_RUN_ID" in os.environ

    def __enter__(self) -> GriptapeCloudStructure:
        from griptape.observability.observability import Observability

        if self.in_managed_environment:
            EventBus.add_event_listener(self.event_listener)

            if self.observe:
                Observability.set_global_driver(self.observability.observability_driver)
                self.observability.observability_driver.__enter__()

        return self

    def __exit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_value: Optional[BaseException],
        exc_traceback: Optional[TracebackType],
    ) -> None:
        from griptape.observability.observability import Observability

        if self.in_managed_environment:
            if self.output is not None:
                EventBus.publish_event(FinishStructureRunEvent(output_task_output=self.output), flush=True)
            EventBus.remove_event_listener(self.event_listener)

            if self.observe:
                Observability.set_global_driver(None)
                self.observability.observability_driver.__exit__(exc_type, exc_value, exc_traceback)

    def _to_artifact(self, value: Any) -> BaseArtifact:
        if isinstance(value, str):
            return TextArtifact(value)
        elif isinstance(value, bool):
            return BooleanArtifact(value)
        elif isinstance(value, dict):
            return JsonArtifact(value)
        elif isinstance(value, bytes):
            return BlobArtifact(value)
        else:
            return GenericArtifact(value)

_event_listener = field(default=None, kw_only=True, alias='event_listener') class-attribute instance-attribute

_observability = field(default=None, kw_only=True, alias='observability') class-attribute instance-attribute

_output = field(default=None, init=False) class-attribute instance-attribute

in_managed_environment property

observe = field(default=False, kw_only=True) class-attribute instance-attribute

output property writable

structure_run_id property

__enter__()

Source code in griptape/utils/griptape_cloud.py 
80
81
82
83
84
85
86
87
88
89
90
def __enter__(self) -> GriptapeCloudStructure:
    from griptape.observability.observability import Observability

    if self.in_managed_environment:
        EventBus.add_event_listener(self.event_listener)

        if self.observe:
            Observability.set_global_driver(self.observability.observability_driver)
            self.observability.observability_driver.__enter__()

    return self

__exit__(exc_type, exc_value, exc_traceback)

Source code in griptape/utils/griptape_cloud.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __exit__(
    self,
    exc_type: Optional[type[BaseException]],
    exc_value: Optional[BaseException],
    exc_traceback: Optional[TracebackType],
) -> None:
    from griptape.observability.observability import Observability

    if self.in_managed_environment:
        if self.output is not None:
            EventBus.publish_event(FinishStructureRunEvent(output_task_output=self.output), flush=True)
        EventBus.remove_event_listener(self.event_listener)

        if self.observe:
            Observability.set_global_driver(None)
            self.observability.observability_driver.__exit__(exc_type, exc_value, exc_traceback)

_to_artifact(value)

Source code in griptape/utils/griptape_cloud.py 
109
110
111
112
113
114
115
116
117
118
119
def _to_artifact(self, value: Any) -> BaseArtifact:
    if isinstance(value, str):
        return TextArtifact(value)
    elif isinstance(value, bool):
        return BooleanArtifact(value)
    elif isinstance(value, dict):
        return JsonArtifact(value)
    elif isinstance(value, bytes):
        return BlobArtifact(value)
    else:
        return GenericArtifact(value)

event_listener()

Source code in griptape/utils/griptape_cloud.py 
45
46
47
48
49
@lazy_property()
def event_listener(self) -> EventListener:
    from griptape.drivers.event_listener.griptape_cloud import GriptapeCloudEventListenerDriver

    return EventListener(event_listener_driver=GriptapeCloudEventListenerDriver())

observability()

Source code in griptape/utils/griptape_cloud.py 
51
52
53
54
55
56
@lazy_property()
def observability(self) -> Observability:
    from griptape.drivers.observability.griptape_cloud import GriptapeCloudObservabilityDriver
    from griptape.observability.observability import Observability

    return Observability(observability_driver=GriptapeCloudObservabilityDriver())