from collections.abc import Mapping from typing import Any from configs import dify_config from .exc import VariableError from .segments import ( ArrayAnySegment, FloatSegment, IntegerSegment, NoneSegment, ObjectSegment, Segment, StringSegment, ) from .types import SegmentType from .variables import ( ArrayNumberVariable, ArrayObjectVariable, ArrayStringVariable, FloatVariable, IntegerVariable, ObjectVariable, SecretVariable, StringVariable, Variable, ) def build_variable_from_mapping(mapping: Mapping[str, Any], /) -> Variable: if (value_type := mapping.get('value_type')) is None: raise VariableError('missing value type') if not mapping.get('name'): raise VariableError('missing name') if (value := mapping.get('value')) is None: raise VariableError('missing value') match value_type: case SegmentType.STRING: result = StringVariable.model_validate(mapping) case SegmentType.SECRET: result = SecretVariable.model_validate(mapping) case SegmentType.NUMBER if isinstance(value, int): result = IntegerVariable.model_validate(mapping) case SegmentType.NUMBER if isinstance(value, float): result = FloatVariable.model_validate(mapping) case SegmentType.NUMBER if not isinstance(value, float | int): raise VariableError(f'invalid number value {value}') case SegmentType.OBJECT if isinstance(value, dict): result = ObjectVariable.model_validate(mapping) case SegmentType.ARRAY_STRING if isinstance(value, list): result = ArrayStringVariable.model_validate(mapping) case SegmentType.ARRAY_NUMBER if isinstance(value, list): result = ArrayNumberVariable.model_validate(mapping) case SegmentType.ARRAY_OBJECT if isinstance(value, list): result = ArrayObjectVariable.model_validate(mapping) case _: raise VariableError(f'not supported value type {value_type}') if result.size > dify_config.MAX_VARIABLE_SIZE: raise VariableError(f'variable size {result.size} exceeds limit {dify_config.MAX_VARIABLE_SIZE}') return result def build_segment(value: Any, /) -> Segment: if value is None: return NoneSegment() if isinstance(value, str): return StringSegment(value=value) if isinstance(value, int): return IntegerSegment(value=value) if isinstance(value, float): return FloatSegment(value=value) if isinstance(value, dict): return ObjectSegment(value=value) if isinstance(value, list): return ArrayAnySegment(value=value) raise ValueError(f'not supported value {value}')