Underground-Digital/Workflow-Engine

1import json2import sys3from collections.abc import Mapping, Sequence4from typing import Any5 6from pydantic import BaseModel, ConfigDict, field_validator7 8from core.file import File9 10from .types import SegmentType11 12 13class Segment(BaseModel):14    model_config = ConfigDict(frozen=True)15 16    value_type: SegmentType17    value: Any18 19    @field_validator("value_type")20    @classmethod21    def validate_value_type(cls, value):22        """23        This validator checks if the provided value is equal to the default value of the 'value_type' field.24        If the value is different, a ValueError is raised.25        """26        if value != cls.model_fields["value_type"].default:27            raise ValueError("Cannot modify 'value_type'")28        return value29 30    @property31    def text(self) -> str:32        return str(self.value)33 34    @property35    def log(self) -> str:36        return str(self.value)37 38    @property39    def markdown(self) -> str:40        return str(self.value)41 42    @property43    def size(self) -> int:44        """45        Return the size of the value in bytes.46        """47        return sys.getsizeof(self.value)48 49    def to_object(self) -> Any:50        return self.value51 52 53class NoneSegment(Segment):54    value_type: SegmentType = SegmentType.NONE55    value: None = None56 57    @property58    def text(self) -> str:59        return ""60 61    @property62    def log(self) -> str:63        return ""64 65    @property66    def markdown(self) -> str:67        return ""68 69 70class StringSegment(Segment):71    value_type: SegmentType = SegmentType.STRING72    value: str73 74 75class FloatSegment(Segment):76    value_type: SegmentType = SegmentType.NUMBER77    value: float78 79 80class IntegerSegment(Segment):81    value_type: SegmentType = SegmentType.NUMBER82    value: int83 84 85class ObjectSegment(Segment):86    value_type: SegmentType = SegmentType.OBJECT87    value: Mapping[str, Any]88 89    @property90    def text(self) -> str:91        return json.dumps(self.model_dump()["value"], ensure_ascii=False)92 93    @property94    def log(self) -> str:95        return json.dumps(self.model_dump()["value"], ensure_ascii=False, indent=2)96 97    @property98    def markdown(self) -> str:99        return json.dumps(self.model_dump()["value"], ensure_ascii=False, indent=2)100 101 102class ArraySegment(Segment):103    @property104    def markdown(self) -> str:105        items = []106        for item in self.value:107            items.append(str(item))108        return "\n".join(items)109 110 111class FileSegment(Segment):112    value_type: SegmentType = SegmentType.FILE113    value: File114 115    @property116    def markdown(self) -> str:117        return self.value.markdown118 119    @property120    def log(self) -> str:121        return str(self.value)122 123    @property124    def text(self) -> str:125        return str(self.value)126 127 128class ArrayAnySegment(ArraySegment):129    value_type: SegmentType = SegmentType.ARRAY_ANY130    value: Sequence[Any]131 132 133class ArrayStringSegment(ArraySegment):134    value_type: SegmentType = SegmentType.ARRAY_STRING135    value: Sequence[str]136 137 138class ArrayNumberSegment(ArraySegment):139    value_type: SegmentType = SegmentType.ARRAY_NUMBER140    value: Sequence[float | int]141 142 143class ArrayObjectSegment(ArraySegment):144    value_type: SegmentType = SegmentType.ARRAY_OBJECT145    value: Sequence[Mapping[str, Any]]146 147 148class ArrayFileSegment(ArraySegment):149    value_type: SegmentType = SegmentType.ARRAY_FILE150    value: Sequence[File]151 152    @property153    def markdown(self) -> str:154        items = []155        for item in self.value:156            items.append(item.markdown)157        return "\n".join(items)158