Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| import json | |
| import uuid | |
| from abc import ABC, abstractmethod | |
| from attr import define, field, Factory | |
| from marshmallow import class_registry | |
| from marshmallow.exceptions import RegistryError | |
| class BaseArtifact(ABC): | |
| id: str = field(default=Factory(lambda: uuid.uuid4().hex), kw_only=True) | |
| name: str = field(default=Factory(lambda self: self.id, takes_self=True), kw_only=True) | |
| value: any = field() | |
| type: str = field(default=Factory(lambda self: self.__class__.__name__, takes_self=True), kw_only=True) | |
| def value_to_bytes(cls, value: any) -> bytes: | |
| if isinstance(value, bytes): | |
| return value | |
| else: | |
| return str(value).encode() | |
| def value_to_dict(cls, value: any) -> dict: | |
| if isinstance(value, dict): | |
| dict_value = value | |
| else: | |
| dict_value = json.loads(value) | |
| return {k: v for k, v in dict_value.items()} | |
| def from_dict(cls, artifact_dict: dict) -> BaseArtifact: | |
| from griptape.schemas import ( | |
| TextArtifactSchema, | |
| InfoArtifactSchema, | |
| ErrorArtifactSchema, | |
| BlobArtifactSchema, | |
| CsvRowArtifactSchema, | |
| ListArtifactSchema | |
| ) | |
| class_registry.register("TextArtifact", TextArtifactSchema) | |
| class_registry.register("InfoArtifact", InfoArtifactSchema) | |
| class_registry.register("ErrorArtifact", ErrorArtifactSchema) | |
| class_registry.register("BlobArtifact", BlobArtifactSchema) | |
| class_registry.register("CsvRowArtifact", CsvRowArtifactSchema) | |
| class_registry.register("ListArtifact", ListArtifactSchema) | |
| try: | |
| return class_registry.get_class(artifact_dict["type"])().load(artifact_dict) | |
| except RegistryError: | |
| raise ValueError("Unsupported artifact type") | |
| def from_json(cls, artifact_str: str) -> BaseArtifact: | |
| return cls.from_dict(json.loads(artifact_str)) | |
| def __str__(self): | |
| return json.dumps(self.to_dict()) | |
| def to_json(self) -> str: | |
| return json.dumps(self.to_dict()) | |
| def to_text(self) -> str: | |
| ... | |
| def to_dict(self) -> dict: | |
| ... | |
| def __add__(self, other: BaseArtifact) -> BaseArtifact: | |
| ... |