Исходный код EJIO.utils.serialization.implementations.msgpack_serializer

## /utils/serialization/implementations/msgpack_serializer.py

from msgpack import packb, unpackb, pack, unpack
from os.path import exists
from typing import Any
from ..serializer import Serializer
from ..serializer_registry import register_serializer
from ..err import SerializationError

__all__ = [
    'MessagePackSerializer'
]




@register_serializer('msgpack')
class MessagePackSerializer(Serializer):
    """
    Реализация интерфейса Serializer для бинарного формата MessagePack.
    """
    __slots__ = ()



    @staticmethod
    def serialize[T](data: dict[str, T], /, **kwargs: Any) -> str:
        """
        Преобразовать словарь в строку, закодированную через MessagePack.

        :param data: Сериализуемый словарь.
        :param kwargs: Ключевые параметры для сереализатора.
        :return: Текстовое представление байтового потока.
        :raises SerializationError: При ошибках сериализации.
        """
        try:
            packed_bytes = packb(data, use_bin_type=True)
            return packed_bytes.decode('latin-1')
        except Exception as e:
            raise SerializationError("Serialization to msgpack failed") from e




    @staticmethod
    def deserialize[T](data: str, /, **kwargs: Any) -> dict[str, T]:
        """
        Преобразовать закодированную строку обратно в словарь.

        :param data: Строковое представление байтового потока.
        :param kwargs: Ключевые параметры для сереализатора.
        :return: Десериализованный словарь.
        :raises SerializationError: При ошибках десериализации.
        """
        try:
            packed_bytes = data.encode('latin-1')
            result = unpackb(packed_bytes, raw=False)
            if not isinstance(result, dict):
                raise SerializationError("MessagePack data must represent a dictionary")
            return result
        except Exception as e:
            raise SerializationError("Deserialization from msgpack failed") from e




    @staticmethod
    def save[T](data: dict[str, T], filename: str, /, **kwargs: Any) -> None:
        """
        Сохранить словарь в бинарный файл MessagePack.

        :param data: Сериализуемый словарь.
        :param filename: Путь к файлу для сохранения.
        :param kwargs: Ключевые параметры для сереализатора.
        :raises SerializationError: При ошибках сериализации.
        """
        try:
            with open(filename, 'wb') as f:
                pack(data, f, use_bin_type=True)
        except (IOError, OSError, Exception) as e:
            raise SerializationError("Failed to serialize data to file") from e




    @staticmethod
    def load[T](filename: str, /, **kwargs: Any) -> dict[str, T]:
        """
        Загрузить словарь из бинарного файла MessagePack.

        :param filename: Путь к файлу для загрузки.
        :param kwargs: Ключевые параметры для сереализатора.
        :return: Словарь, полученный из файла.
        :raises SerializationError: При ошибках десериализации.
        """
        try:
            if not exists(filename):
                raise FileNotFoundError(f"File {filename} not found")
            with open(filename, 'rb') as f:
                result = unpack(f, raw=False)
            if not isinstance(result, dict):
                raise SerializationError("MessagePack file data must represent a dictionary")
            return result
        except (IOError, OSError, FileNotFoundError, Exception) as e:
            raise SerializationError("Failed to deserialize data from file") from e