Coverage for src/dataknobs_data/fields.py: 58%

1import copy 

2from dataclasses import dataclass, field 

3from datetime import datetime 

4from enum import Enum 

5from typing import Any, Dict 

6 

7 

8class FieldType(Enum): 

9 """Enumeration of supported field types.""" 

10 

11 STRING = "string" 

12 INTEGER = "integer" 

13 FLOAT = "float" 

14 BOOLEAN = "boolean" 

15 DATETIME = "datetime" 

16 JSON = "json" 

17 BINARY = "binary" 

18 TEXT = "text" 

19 

20 

21@dataclass 

22class Field: 

23 """Represents a single field in a record.""" 

24 

25 name: str 

26 value: Any 

27 type: FieldType | None = None 

28 metadata: Dict[str, Any] = field(default_factory=dict) 

29 

30 def __post_init__(self): 

31 """Auto-detect type if not provided.""" 

32 if self.type is None: 

33 self.type = self._detect_type(self.value) 

34 

35 def _detect_type(self, value: Any) -> FieldType: 

36 """Detect the field type from the value.""" 

37 if value is None: 

38 return FieldType.STRING 

39 elif isinstance(value, bool): 

40 return FieldType.BOOLEAN 

41 elif isinstance(value, int): 

42 return FieldType.INTEGER 

43 elif isinstance(value, float): 

44 return FieldType.FLOAT 

45 elif isinstance(value, datetime): 

46 return FieldType.DATETIME 

47 elif isinstance(value, (dict, list)): 

48 return FieldType.JSON 

49 elif isinstance(value, bytes): 

50 return FieldType.BINARY 

51 elif isinstance(value, str): 

52 if len(value) > 1000: 

53 return FieldType.TEXT 

54 return FieldType.STRING 

55 else: 

56 return FieldType.JSON 

57 

58 def copy(self) -> "Field": 

59 """Create a deep copy of the field.""" 

60 return Field( 

61 name=self.name, 

62 value=copy.deepcopy(self.value), 

63 type=self.type, 

64 metadata=copy.deepcopy(self.metadata) 

65 ) 

66 

67 def validate(self) -> bool: 

68 """Validate that the value matches the field type.""" 

69 if self.value is None: 

70 return True 

71 

72 type_validators = { 

73 FieldType.STRING: lambda v: isinstance(v, str), 

74 FieldType.INTEGER: lambda v: isinstance(v, int) and not isinstance(v, bool), 

75 FieldType.FLOAT: lambda v: isinstance(v, (int, float)) and not isinstance(v, bool), 

76 FieldType.BOOLEAN: lambda v: isinstance(v, bool), 

77 FieldType.DATETIME: lambda v: isinstance(v, datetime), 

78 FieldType.JSON: lambda v: isinstance(v, (dict, list)), 

79 FieldType.BINARY: lambda v: isinstance(v, bytes), 

80 FieldType.TEXT: lambda v: isinstance(v, str), 

81 } 

82 

83 validator = type_validators.get(self.type) 

84 if validator: 

85 return validator(self.value) 

86 return True 

87 

88 def convert_to(self, target_type: FieldType) -> "Field": 

89 """Convert the field to a different type.""" 

90 if self.type == target_type: 

91 return self 

92 

93 converters = { 

94 (FieldType.INTEGER, FieldType.STRING): str, 

95 (FieldType.INTEGER, FieldType.FLOAT): float, 

96 (FieldType.FLOAT, FieldType.STRING): str, 

97 (FieldType.FLOAT, FieldType.INTEGER): int, 

98 (FieldType.BOOLEAN, FieldType.STRING): lambda v: "true" if v else "false", 

99 (FieldType.BOOLEAN, FieldType.INTEGER): int, 

100 (FieldType.STRING, FieldType.INTEGER): int, 

101 (FieldType.STRING, FieldType.FLOAT): float, 

102 (FieldType.STRING, FieldType.BOOLEAN): lambda v: v.lower() in ("true", "1", "yes"), 

103 (FieldType.STRING, FieldType.TEXT): lambda v: v, 

104 (FieldType.TEXT, FieldType.STRING): lambda v: v, 

105 } 

106 

107 converter_key = (self.type, target_type) 

108 if converter_key in converters: 

109 try: 

110 new_value = converters[converter_key](self.value) 

111 return Field( 

112 name=self.name, value=new_value, type=target_type, metadata=self.metadata.copy() 

113 ) 

114 except (ValueError, TypeError) as e: 

115 raise ValueError( 

116 f"Cannot convert {self.name} from {self.type} to {target_type}: {e}" 

117 ) 

118 else: 

119 raise ValueError(f"No converter available from {self.type} to {target_type}") 

120 

121 def to_dict(self) -> Dict[str, Any]: 

122 """Convert the field to a dictionary representation.""" 

123 return { 

124 "name": self.name, 

125 "value": self.value, 

126 "type": self.type.value if self.type else None, 

127 "metadata": self.metadata, 

128 } 

129 

130 @classmethod 

131 def from_dict(cls, data: Dict[str, Any]) -> "Field": 

132 """Create a field from a dictionary representation.""" 

133 field_type = None 

134 if data.get("type"): 

135 field_type = FieldType(data["type"]) 

136 

137 return cls( 

138 name=data["name"], 

139 value=data["value"], 

140 type=field_type, 

141 metadata=data.get("metadata", {}), 

142 )