Skip to content

Serializers

Built-in serializer implementations and helper to select serializer.

Bases: ABC

序列化器抽象基类

定义序列化和反序列化接口。

Source code in src/symphra_cache/serializers.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class BaseSerializer(ABC):
    """
    序列化器抽象基类

    定义序列化和反序列化接口。
    """

    @abstractmethod
    def serialize(self, value: CacheValue) -> bytes:
        """
        序列化值为字节

        Args:
            value: 要序列化的值

        Returns:
            序列化后的字节数据

        Raises:
            CacheSerializationError: 序列化失败
        """
        raise NotImplementedError

    @abstractmethod
    def deserialize(self, data: bytes) -> CacheValue:
        """
        反序列化字节为值

        Args:
            data: 要反序列化的字节数据

        Returns:
            反序列化后的值

        Raises:
            CacheSerializationError: 反序列化失败
        """
        raise NotImplementedError

deserialize(data) abstractmethod

反序列化字节为值

Parameters:

Name Type Description Default
data bytes

要反序列化的字节数据

 required

Returns:

Type Description
CacheValue

反序列化后的值

Raises:

Type Description
CacheSerializationError

反序列化失败
Source code in src/symphra_cache/serializers.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@abstractmethod
def deserialize(self, data: bytes) -> CacheValue:
    """
    反序列化字节为值

    Args:
        data: 要反序列化的字节数据

    Returns:
        反序列化后的值

    Raises:
        CacheSerializationError: 反序列化失败
    """
    raise NotImplementedError

serialize(value) abstractmethod

序列化值为字节

Parameters:

Name Type Description Default
value CacheValue

要序列化的值

 required

Returns:

Type Description
bytes

序列化后的字节数据

Raises:

Type Description
CacheSerializationError

序列化失败
Source code in src/symphra_cache/serializers.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@abstractmethod
def serialize(self, value: CacheValue) -> bytes:
    """
    序列化值为字节

    Args:
        value: 要序列化的值

    Returns:
        序列化后的字节数据

    Raises:
        CacheSerializationError: 序列化失败
    """
    raise NotImplementedError

Bases: BaseSerializer

JSON 序列化器

优点： - 可读性好 - 跨语言兼容 - 适合简单数据结构

缺点： - 不支持复杂 Python 对象（如 datetime、bytes） - 性能相对较低

示例： >>> serializer = JSONSerializer() >>> data = {"key": "value", "count": 123} >>> bytes_data = serializer.serialize(data) >>> original = serializer.deserialize(bytes_data)

Source code in src/symphra_cache/serializers.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class JSONSerializer(BaseSerializer):
    """
    JSON 序列化器

    优点：
    - 可读性好
    - 跨语言兼容
    - 适合简单数据结构

    缺点：
    - 不支持复杂 Python 对象（如 datetime、bytes）
    - 性能相对较低

    示例：
        >>> serializer = JSONSerializer()
        >>> data = {"key": "value", "count": 123}
        >>> bytes_data = serializer.serialize(data)
        >>> original = serializer.deserialize(bytes_data)
    """

    def serialize(self, value: CacheValue) -> bytes:
        """将值序列化为 JSON 字节"""
        try:
            # 使用 ensure_ascii=False 支持中文等 Unicode 字符
            json_str = json.dumps(value, ensure_ascii=False)
            return json_str.encode("utf-8")
        except (TypeError, ValueError) as e:
            msg = f"JSON 序列化失败: {e}"
            raise CacheSerializationError(msg) from e

    def deserialize(self, data: bytes) -> CacheValue:
        """从 JSON 字节反序列化值"""
        try:
            json_str = data.decode("utf-8")
            return json.loads(json_str)
        except (json.JSONDecodeError, UnicodeDecodeError) as e:
            msg = f"JSON 反序列化失败: {e}"
            raise CacheSerializationError(msg) from e

deserialize(data)

从 JSON 字节反序列化值

Source code in src/symphra_cache/serializers.py 
100
101
102
103
104
105
106
107
def deserialize(self, data: bytes) -> CacheValue:
    """从 JSON 字节反序列化值"""
    try:
        json_str = data.decode("utf-8")
        return json.loads(json_str)
    except (json.JSONDecodeError, UnicodeDecodeError) as e:
        msg = f"JSON 反序列化失败: {e}"
        raise CacheSerializationError(msg) from e

serialize(value)

将值序列化为 JSON 字节

Source code in src/symphra_cache/serializers.py 
90
91
92
93
94
95
96
97
98
def serialize(self, value: CacheValue) -> bytes:
    """将值序列化为 JSON 字节"""
    try:
        # 使用 ensure_ascii=False 支持中文等 Unicode 字符
        json_str = json.dumps(value, ensure_ascii=False)
        return json_str.encode("utf-8")
    except (TypeError, ValueError) as e:
        msg = f"JSON 序列化失败: {e}"
        raise CacheSerializationError(msg) from e

Bases: BaseSerializer

Pickle 序列化器

优点： - 支持几乎所有 Python 对象 - 性能较好 - Python 标准库内置

缺点： - 不跨语言 - 安全风险（不要反序列化不可信数据） - 二进制格式，不可读

警告： 仅反序列化可信来源的数据，避免代码注入风险

示例： >>> serializer = PickleSerializer() >>> import datetime >>> data = {"time": datetime.datetime.now(), "items": [1, 2, 3]} >>> bytes_data = serializer.serialize(data) >>> original = serializer.deserialize(bytes_data)

Source code in src/symphra_cache/serializers.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class PickleSerializer(BaseSerializer):
    """
    Pickle 序列化器

    优点：
    - 支持几乎所有 Python 对象
    - 性能较好
    - Python 标准库内置

    缺点：
    - 不跨语言
    - 安全风险（不要反序列化不可信数据）
    - 二进制格式，不可读

    警告：
        仅反序列化可信来源的数据，避免代码注入风险

    示例：
        >>> serializer = PickleSerializer()
        >>> import datetime
        >>> data = {"time": datetime.datetime.now(), "items": [1, 2, 3]}
        >>> bytes_data = serializer.serialize(data)
        >>> original = serializer.deserialize(bytes_data)
    """

    def serialize(self, value: CacheValue) -> bytes:
        """将值序列化为 Pickle 字节"""
        try:
            # 使用协议 5（Python 3.8+，性能最优）
            return pickle.dumps(value, protocol=pickle.HIGHEST_PROTOCOL)
        except (pickle.PicklingError, TypeError) as e:
            msg = f"Pickle 序列化失败: {e}"
            raise CacheSerializationError(msg) from e

    def deserialize(self, data: bytes) -> CacheValue:
        """从 Pickle 字节反序列化值"""
        try:
            return pickle.loads(data)  # noqa: S301
        except (pickle.UnpicklingError, AttributeError, EOFError) as e:
            msg = f"Pickle 反序列化失败: {e}"
            raise CacheSerializationError(msg) from e

deserialize(data)

从 Pickle 字节反序列化值

Source code in src/symphra_cache/serializers.py 
144
145
146
147
148
149
150
def deserialize(self, data: bytes) -> CacheValue:
    """从 Pickle 字节反序列化值"""
    try:
        return pickle.loads(data)  # noqa: S301
    except (pickle.UnpicklingError, AttributeError, EOFError) as e:
        msg = f"Pickle 反序列化失败: {e}"
        raise CacheSerializationError(msg) from e

serialize(value)

将值序列化为 Pickle 字节

Source code in src/symphra_cache/serializers.py 
135
136
137
138
139
140
141
142
def serialize(self, value: CacheValue) -> bytes:
    """将值序列化为 Pickle 字节"""
    try:
        # 使用协议 5（Python 3.8+，性能最优）
        return pickle.dumps(value, protocol=pickle.HIGHEST_PROTOCOL)
    except (pickle.PicklingError, TypeError) as e:
        msg = f"Pickle 序列化失败: {e}"
        raise CacheSerializationError(msg) from e

Bases: BaseSerializer

MessagePack 序列化器

优点： - 高性能（比 JSON 快 2-5 倍） - 紧凑的二进制格式 - 跨语言兼容

缺点： - 需要额外依赖 msgpack - 对复杂 Python 对象支持有限

示例： >>> serializer = MessagePackSerializer() >>> data = {"users": [{"id": 1}, {"id": 2}], "total": 2} >>> bytes_data = serializer.serialize(data) >>> original = serializer.deserialize(bytes_data)

Source code in src/symphra_cache/serializers.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class MessagePackSerializer(BaseSerializer):
    """
    MessagePack 序列化器

    优点：
    - 高性能（比 JSON 快 2-5 倍）
    - 紧凑的二进制格式
    - 跨语言兼容

    缺点：
    - 需要额外依赖 msgpack
    - 对复杂 Python 对象支持有限

    示例：
        >>> serializer = MessagePackSerializer()
        >>> data = {"users": [{"id": 1}, {"id": 2}], "total": 2}
        >>> bytes_data = serializer.serialize(data)
        >>> original = serializer.deserialize(bytes_data)
    """

    def __init__(self) -> None:
        """初始化 MessagePack 序列化器"""
        try:
            import msgpack

            self._msgpack = msgpack
        except ImportError as e:
            msg = "MessagePack 序列化需要安装 msgpack: pip install msgpack"
            raise ImportError(msg) from e

    def serialize(self, value: CacheValue) -> bytes:
        """将值序列化为 MessagePack 字节"""
        try:
            return self._msgpack.packb(value, use_bin_type=True)
        except (self._msgpack.PackException, TypeError) as e:
            msg = f"MessagePack 序列化失败: {e}"
            raise CacheSerializationError(msg) from e

    def deserialize(self, data: bytes) -> CacheValue:
        """从 MessagePack 字节反序列化值"""
        try:
            return self._msgpack.unpackb(data, raw=False)
        except (self._msgpack.UnpackException, ValueError) as e:
            msg = f"MessagePack 反序列化失败: {e}"
            raise CacheSerializationError(msg) from e

__init__()

初始化 MessagePack 序列化器

Source code in src/symphra_cache/serializers.py 
173
174
175
176
177
178
179
180
181
def __init__(self) -> None:
    """初始化 MessagePack 序列化器"""
    try:
        import msgpack

        self._msgpack = msgpack
    except ImportError as e:
        msg = "MessagePack 序列化需要安装 msgpack: pip install msgpack"
        raise ImportError(msg) from e

deserialize(data)

从 MessagePack 字节反序列化值

Source code in src/symphra_cache/serializers.py 
191
192
193
194
195
196
197
def deserialize(self, data: bytes) -> CacheValue:
    """从 MessagePack 字节反序列化值"""
    try:
        return self._msgpack.unpackb(data, raw=False)
    except (self._msgpack.UnpackException, ValueError) as e:
        msg = f"MessagePack 反序列化失败: {e}"
        raise CacheSerializationError(msg) from e

serialize(value)

将值序列化为 MessagePack 字节

Source code in src/symphra_cache/serializers.py 
183
184
185
186
187
188
189
def serialize(self, value: CacheValue) -> bytes:
    """将值序列化为 MessagePack 字节"""
    try:
        return self._msgpack.packb(value, use_bin_type=True)
    except (self._msgpack.PackException, TypeError) as e:
        msg = f"MessagePack 序列化失败: {e}"
        raise CacheSerializationError(msg) from e

获取指定模式的序列化器实例

Parameters:

Name Type Description Default
mode SerializationMode | str

序列化模式（SerializationMode 枚举或字符串）

 required

Returns:

Type Description
BaseSerializer

序列化器实例

Raises:

Type Description
ValueError

不支持的序列化模式
示例

serializer = get_serializer(SerializationMode.JSON)

或使用字符串

serializer = get_serializer("json")

Source code in src/symphra_cache/serializers.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def get_serializer(mode: SerializationMode | str) -> BaseSerializer:
    """
    获取指定模式的序列化器实例

    Args:
        mode: 序列化模式（SerializationMode 枚举或字符串）

    Returns:
        序列化器实例

    Raises:
        ValueError: 不支持的序列化模式

    示例:
        >>> serializer = get_serializer(SerializationMode.JSON)
        >>> # 或使用字符串
        >>> serializer = get_serializer("json")
    """
    # 支持字符串参数
    if isinstance(mode, str):
        try:
            mode = SerializationMode(mode)
        except ValueError as e:
            msg = f"不支持的序列化模式: {mode}"
            raise ValueError(msg) from e

    # 获取序列化器类
    serializer_cls = _SERIALIZERS.get(mode)
    if serializer_cls is None:
        msg = f"未注册的序列化模式: {mode}"
        raise ValueError(msg)

    # 返回实例
    return serializer_cls()