Coverage for src/dataknobs_common/serialization.py: 36%

1"""Serialization protocols and utilities for dataknobs packages.

3This module provides standard interfaces for objects that can be serialized

4to and from dictionaries. This enables consistent serialization patterns

5across all dataknobs packages.

7The serialization framework supports:

8- Type-safe protocols for serializable objects

9- Utility functions for serialization/deserialization

10- Runtime type checking with isinstance()

11- Integration with dataclasses and custom classes

13Example:

14 ```python

15 from dataknobs_common.serialization import Serializable

16 from dataclasses import dataclass

18 @dataclass

19 class User:

20 name: str

21 email: str

23 def to_dict(self) -> dict:

24 return {"name": self.name, "email": self.email}

26 @classmethod

27 def from_dict(cls, data: dict) -> "User":

28 return cls(name=data["name"], email=data["email"])

30 # Type checking works

31 user = User("Alice", "alice@example.com")

32 assert isinstance(user, Serializable) # True

34 # Use utilities

35 from dataknobs_common.serialization import serialize, deserialize

37 data = serialize(user)

38 restored = deserialize(User, data)

39 ```

40"""

42from typing import Any, Dict, Protocol, Type, TypeVar, runtime_checkable

44from dataknobs_common.exceptions import SerializationError

46T = TypeVar("T")

49@runtime_checkable

50class Serializable(Protocol):

51 """Protocol for objects that can be serialized to/from dict.

53 Implement this protocol by providing to_dict() and from_dict() methods.

54 The @runtime_checkable decorator allows isinstance() checks at runtime.

56 Methods:

57 to_dict: Convert object to dictionary representation

58 from_dict: Create object from dictionary representation

60 Example:

61 >>> class MyClass:

62 ... def __init__(self, value: str):

63 ... self.value = value

64 ...

65 ... def to_dict(self) -> dict:

66 ... return {"value": self.value}

67 ...

68 ... @classmethod

69 ... def from_dict(cls, data: dict) -> "MyClass":

70 ... return cls(data["value"])

71 ...

72 >>> obj = MyClass("test")

73 >>> isinstance(obj, Serializable)

74 True

75 """

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

78 """Convert object to dictionary representation.

80 Returns:

81 Dictionary with serialized data

83 Raises:

84 SerializationError: If serialization fails

85 """

86 ...

88 @classmethod

89 def from_dict(cls: Type[T], data: Dict[str, Any]) -> T:

90 """Create object from dictionary representation.

92 Args:

93 data: Dictionary with serialized data

95 Returns:

96 Deserialized object instance

98 Raises:

99 SerializationError: If deserialization fails

100 """

101 ...

102

103

104def serialize(obj: Any) -> Dict[str, Any]:

105 """Serialize an object to dictionary.

106

107 Convenience function that calls to_dict() with error handling.

108

109 Args:

110 obj: Object to serialize (must have to_dict method)

111

112 Returns:

113 Serialized dictionary

114

115 Raises:

116 SerializationError: If object doesn't support serialization or serialization fails

117

118 Example:

119 >>> class Point:

120 ... def __init__(self, x: int, y: int):

121 ... self.x, self.y = x, y

122 ... def to_dict(self):

123 ... return {"x": self.x, "y": self.y}

124 ...

125 >>> point = Point(10, 20)

126 >>> data = serialize(point)

127 >>> data

128 {'x': 10, 'y': 20}

129 """

130 if not hasattr(obj, "to_dict"):

131 raise SerializationError(

132 f"Object of type {type(obj).__name__} is not serializable (missing to_dict method)",

133 context={"type": type(obj).__name__, "object": str(obj)},

134 )

135

136 try:

137 result = obj.to_dict()

138 if not isinstance(result, dict):

139 raise SerializationError(

140 f"to_dict() must return a dict, got {type(result).__name__}",

141 context={"type": type(obj).__name__, "result_type": type(result).__name__},

142 )

143 return result

144 except Exception as e:

145 if isinstance(e, SerializationError):

146 raise

147 raise SerializationError(

148 f"Failed to serialize {type(obj).__name__}: {e}",

149 context={"type": type(obj).__name__, "error": str(e)},

150 ) from e

151

152

153def deserialize(cls: Type[T], data: Dict[str, Any]) -> T:

154 """Deserialize dictionary into an object.

155

156 Convenience function that calls from_dict() with error handling.

157

158 Args:

159 cls: Class to deserialize into (must have from_dict classmethod)

160 data: Dictionary with serialized data

161

162 Returns:

163 Deserialized object instance

164

165 Raises:

166 SerializationError: If class doesn't support deserialization or deserialization fails

167

168 Example:

169 >>> class Point:

170 ... def __init__(self, x: int, y: int):

171 ... self.x, self.y = x, y

172 ... @classmethod

173 ... def from_dict(cls, data: dict):

174 ... return cls(data["x"], data["y"])

175 ...

176 >>> data = {"x": 10, "y": 20}

177 >>> point = deserialize(Point, data)

178 >>> point.x, point.y

179 (10, 20)

180 """

181 if not hasattr(cls, "from_dict"):

182 raise SerializationError(

183 f"Class {cls.__name__} is not deserializable (missing from_dict classmethod)",

184 context={"class": cls.__name__},

185 )

186

187 if not isinstance(data, dict):

188 raise SerializationError(

189 f"Data must be a dict, got {type(data).__name__}",

190 context={"class": cls.__name__, "data_type": type(data).__name__},

191 )

192

193 try:

194 return cls.from_dict(data)

195 except Exception as e:

196 if isinstance(e, SerializationError):

197 raise

198 raise SerializationError(

199 f"Failed to deserialize {cls.__name__}: {e}",

200 context={"class": cls.__name__, "error": str(e), "data": data},

201 ) from e

202

203

204def serialize_list(items: list[Any]) -> list[Dict[str, Any]]:

205 """Serialize a list of objects to list of dictionaries.

206

207 Args:

208 items: List of serializable objects

209

210 Returns:

211 List of serialized dictionaries

212

213 Raises:

214 SerializationError: If any item cannot be serialized

215

216 Example:

217 >>> items = [Point(1, 2), Point(3, 4)]

218 >>> data_list = serialize_list(items)

219 >>> len(data_list)

220 2

221 """

222 return [serialize(item) for item in items]

223

224

225def deserialize_list(cls: Type[T], data_list: list[Dict[str, Any]]) -> list[T]:

226 """Deserialize a list of dictionaries into objects.

227

228 Args:

229 cls: Class to deserialize into

230 data_list: List of serialized dictionaries

231

232 Returns:

233 List of deserialized objects

234

235 Raises:

236 SerializationError: If any item cannot be deserialized

237

238 Example:

239 >>> data_list = [{"x": 1, "y": 2}, {"x": 3, "y": 4}]

240 >>> points = deserialize_list(Point, data_list)

241 >>> len(points)

242 2

243 """

244 return [deserialize(cls, data) for data in data_list]

245

246

247def is_serializable(obj: Any) -> bool:

248 """Check if an object is serializable.

249

250 Args:

251 obj: Object to check

252

253 Returns:

254 True if object has to_dict method

255

256 Example:

257 >>> class Point:

258 ... def to_dict(self): return {}

259 ...

260 >>> is_serializable(Point())

261 True

262 >>> is_serializable("string")

263 False

264 """

265 return isinstance(obj, Serializable) or hasattr(obj, "to_dict")

266

267

268def is_deserializable(cls: Type) -> bool:

269 """Check if a class is deserializable.

270

271 Args:

272 cls: Class to check

273

274 Returns:

275 True if class has from_dict classmethod

276

277 Example:

278 >>> class Point:

279 ... @classmethod

280 ... def from_dict(cls, data): return cls()

281 ...

282 >>> is_deserializable(Point)

283 True

284 >>> is_deserializable(str)

285 False

286 """

287 return hasattr(cls, "from_dict")

288

289

290__all__ = [

291 "Serializable",

292 "serialize",

293 "deserialize",

294 "serialize_list",

295 "deserialize_list",

296 "is_serializable",

297 "is_deserializable",

298]