Coverage for autocrud/serializer.py: 77%

1"""序列化器模組""" 

2 

3import json 

4import pickle 

5from typing import Any 

6from abc import ABC, abstractmethod 

7 

8try: 

9 import msgpack 

10 

11 HAS_MSGPACK = True 

12except ImportError: 

13 HAS_MSGPACK = False 

14 

15 

16class Serializer(ABC): 

17 """序列化器抽象基類""" 

18 

19 @abstractmethod 

20 def serialize(self, data: Any) -> bytes: 

21 """序列化數據""" 

22 pass 

23 

24 @abstractmethod 

25 def deserialize(self, data: bytes) -> Any: 

26 """反序列化數據""" 

27 pass 

28 

29 

30class JsonSerializer(Serializer): 

31 """JSON 序列化器""" 

32 

33 def serialize(self, data: Any) -> bytes: 

34 return json.dumps(data, ensure_ascii=False).encode("utf-8") 

35 

36 def deserialize(self, data: bytes) -> Any: 

37 return json.loads(data.decode("utf-8")) 

38 

39 

40class PickleSerializer(Serializer): 

41 """Pickle 序列化器""" 

42 

43 def serialize(self, data: Any) -> bytes: 

44 return pickle.dumps(data) 

45 

46 def deserialize(self, data: bytes) -> Any: 

47 return pickle.loads(data) 

48 

49 

50class MsgPackSerializer(Serializer): 

51 """MsgPack 序列化器""" 

52 

53 def __init__(self): 

54 if not HAS_MSGPACK: 

55 raise ImportError("需要安裝 msgpack: pip install msgpack") 

56 

57 def serialize(self, data: Any) -> bytes: 

58 return msgpack.packb(data, use_bin_type=True) 

59 

60 def deserialize(self, data: bytes) -> Any: 

61 return msgpack.unpackb(data, raw=False) 

62 

63 

64class SerializerFactory: 

65 """序列化器工廠""" 

66 

67 _serializers = { 

68 "json": JsonSerializer, 

69 "pickle": PickleSerializer, 

70 "msgpack": MsgPackSerializer, 

71 } 

72 

73 @classmethod 

74 def create(cls, serializer_type: str) -> Serializer: 

75 """創建序列化器實例""" 

76 if serializer_type not in cls._serializers: 

77 available = ", ".join(cls._serializers.keys()) 

78 raise ValueError( 

79 f"不支援的序列化器類型: {serializer_type}. 可用類型: {available}" 

80 ) 

81 

82 serializer_class = cls._serializers[serializer_type] 

83 return serializer_class() 

84 

85 @classmethod 

86 def register(cls, name: str, serializer_class: type): 

87 """註冊新的序列化器""" 

88 cls._serializers[name] = serializer_class 

89 

90 @classmethod 

91 def available_types(cls) -> list: 

92 """獲取可用的序列化器類型""" 

93 return list(cls._serializers.keys()) 

94 

95 

96# 使用範例 

97if __name__ == "__main__": 

98 # 測試 JSON 序列化器 

99 json_serializer = SerializerFactory.create("json") 

100 

101 test_data = {"name": "Alice", "age": 30, "email": "alice@example.com"} 

102 

103 # 序列化 

104 serialized = json_serializer.serialize(test_data) 

105 print(f"序列化後: {serialized}") 

106 

107 # 反序列化 

108 deserialized = json_serializer.deserialize(serialized) 

109 print(f"反序列化後: {deserialized}") 

110 

111 # 顯示可用的序列化器 

112 print(f"可用的序列化器: {SerializerFactory.available_types()}")