## /utils/serialization/operations.py
from pathlib import Path
from typing import Any, overload
from .storable import Storable
from .serializable import Serializable
from .serializer_registry import SerializerRegistry
from ..system import save_atomic
__all__ = [
'save',
'load',
'serialize',
'deserialize',
'register_serializer_for_extension',
]
# Краткая карта соответствий расширений и ключей реестра
mapping = {
'json': 'json',
'yaml': 'yaml',
'yml': 'yaml',
'toml': 'toml',
'msgpack': 'msgpack',
'mpk': 'msgpack',
'ini': 'ini',
'xml': 'xml',
}
[документация]
def register_serializer_for_extension(extension: str, serializer_name: str) -> None:
"""
Выставляет соответствие между расширением файла и сериализатором, ответственным за файлы с этим расширением.
:param extension: Расширение файла.
:param serializer_name: Имя сериализатора, под которым он зарегестрирован в реестре.
:raises KeyError: Если соответствие для расширения уже задано.
"""
if extension in mapping:
raise KeyError(f"Serializer for extension '{extension}' already registered.")
mapping[extension] = serializer_name
def _detect_serializer_key(filename: str, fallback_key: str | None) -> str:
"""
Внутренний хелпер: автоматически определяет ключ сериализатора
на основе расширения файла, если fallback_key равен None или дефолтному 'json'.
:param filename: Имя файла.
:param fallback_key: Ключ сериализатора.
:return: Ключ сериализатора для подходящего расширения файла, если соответствие задано,
или fallback_key если он был явно передан в функцию или json для всех остальных случаев.
"""
# Если пользователь явно передал кастомный ключ (не 'json'), используем его
if fallback_key is not None and fallback_key != 'json':
return fallback_key
# Извлекаем расширение файла (например, '.yaml' -> 'yaml')
suffix = Path(filename).suffix.lstrip('.').lower()
detected_key = mapping.get(suffix)
if detected_key and SerializerRegistry.has_key(detected_key): # type: ignore[attr-defined]
return detected_key
# Если расширение неизвестно, откатываемся на дефолтный ключ (или 'json')
return fallback_key or 'json'
@overload
def save[T: Storable](obj: type[T], filename: str, serializer_key: str | None = None, /, **kwargs: Any) -> None: ...
@overload
def save(obj: dict[str, Any], filename: str, serializer_key: str | None = None, /, **kwargs: Any) -> None: ...
[документация]
def save(obj: Any, filename: str, serializer_key: str | None = None, /, **kwargs: Any) -> None:
"""
Сохранить данные в файл, используя ключ зарегистрированного сериализатора.
Поддерживает как объекты Storable, так и простые словари.
Автоматически определяет формат по расширению файла, если serializer_key не задан.
:param obj: Объект, реализующий интерфейс Storable или словарь.
:param filename: Имя или путь к файлу.
:param serializer_key: Строковый ключ сериализатора в глобальном регистре.
:param kwargs: Ключевые параметры для сереализатора.
"""
actual_key = _detect_serializer_key(filename, serializer_key)
serializer_cls = SerializerRegistry.get(actual_key)
def execute_write(temp_file_path: Path) -> None:
if hasattr(obj, 'save') and callable(getattr(obj, 'save')):
obj.save(filename=str(temp_file_path), serializer=serializer_cls, **kwargs)
else:
data_to_save = obj.to_dict() if hasattr(obj, 'to_dict') else obj
serializer_cls.save(data=data_to_save, filename=str(temp_file_path), **kwargs)
save_atomic(filename, execute_write)
@overload
def load[T: Storable](target_cls: type[T], filename: str, serializer_key: str | None = None, /, **kwargs: Any) -> T: ...
@overload
def load(target_cls: None, filename: str, serializer_key: str | None = None, /, **kwargs: Any) -> dict[str, Any]: ...
[документация]
def load(target_cls: Any, filename: str, serializer_key: str | None = None, /, **kwargs: Any) -> Any:
"""
Загрузить и восстановить данные из файла, используя ключ зарегистрированного сериализатора.
Поддерживает как объекты Storable, так и простые словари.
Автоматически определяет формат по расширению файла, если serializer_key не задан.
:param target_cls: Класс восстанавливаемого объекта (должен быть наследником Storable) либо None.
:param filename: Имя или путь к файлу.
:param serializer_key: Строковый ключ сериализатора в глобальном регистре.
:param kwargs: Ключевые параметры для сереализатора.
:return: Восстановленный экземпляр класса target_cls, или словарь, если target_cls == None.
"""
actual_key = _detect_serializer_key(filename, serializer_key)
serializer_cls = SerializerRegistry.get(actual_key)
if target_cls is not None and hasattr(target_cls, 'load') and callable(getattr(target_cls, 'load')):
return target_cls.load(filename=filename, serializer=serializer_cls, **kwargs)
return serializer_cls.load(filename=filename, **kwargs)
@overload
def serialize[T: Serializable](obj: T, serializer_key: str = 'json', /, **kwargs: Any) -> str: ...
@overload
def serialize(obj: dict[str, Any], serializer_key: str = 'json', /, **kwargs: Any) -> str: ...
[документация]
def serialize(obj: Any, serializer_key: str = 'json', /, **kwargs: Any) -> str:
"""
Сериализовать данные в строку, используя ключ зарегистрированного сериализатора.
Поддерживает как объекты Serializable, так и простые словари.
:param obj: Объект, реализующий интерфейс Serializable или словарь.
:param serializer_key: Строковый ключ сериализатора в глобальном регистре.
:param kwargs: Ключевые параметры для сереализатора.
:return: Строка с сериализованными данными.
"""
serializer_cls = SerializerRegistry.get(serializer_key)
if hasattr(obj, 'serialize') and callable(getattr(obj, 'serialize')):
return obj.serialize(serializer=serializer_cls, **kwargs)
data_to_serialize = obj.to_dict() if hasattr(obj, 'to_dict') else obj
return serializer_cls.serialize(data=data_to_serialize, **kwargs)
@overload
def deserialize[T: Serializable](target_cls: type[T], data: str, serializer_key: str = 'json', /, **kwargs: Any) -> T: ...
@overload
def deserialize(target_cls: None, data: str, serializer_key: str = 'json', /, **kwargs: Any) -> dict[str, Any]: ...
[документация]
def deserialize(target_cls: Any, data: str, serializer_key: str = 'json', /, **kwargs: Any) -> Any:
"""
Десериализовать данные из строки, используя ключ зарегистрированного сериализатора.
Поддерживает как объекты Serializable, так и простые словари.
:param target_cls: Класс восстанавливаемого объекта (должен быть наследником Serializable) либо None.
:param data: Сериализованная строка данных.
:param serializer_key: Строковый ключ сериализатора в глобальном регистре.
:param kwargs: Ключевые параметры для сереализатора.
:return: Новое состояние экземпляра класса target_cls, или словарь, если target_cls == None.
"""
serializer_cls = SerializerRegistry.get(serializer_key)
if target_cls is not None and hasattr(target_cls, 'deserialize') and callable(getattr(target_cls, 'deserialize')):
return target_cls.deserialize(data=data, serializer=serializer_cls, **kwargs)
return serializer_cls.deserialize(data=data, **kwargs)