Исходный код EJIO.utils.interface.prototype

## /utils/interface/prototype.py

from threading import Lock
from typing import Any, Protocol, Self
from copy import copy, deepcopy

__all__ = [
    'Prototype',
    'ThreadSafePrototypeMixin',
]


[документация] class Prototype(Protocol): """ Протокол, реализующий методы для копирования себя. """ __slots__ = ()
[документация] def clone(self) -> Self: """ Поверхностное копирование (вложенные объекты остаются общими). :return: Поверхностную копию объекта. """ return copy(self)
[документация] def deep_clone(self) -> Self: """ Глубокое копирование (вложенные объекты тоже дублируются). :return: Полную копию объекта. """ return deepcopy(self)
[документация] class ThreadSafePrototypeMixin(Prototype): """ Миксин-реализация для создания потокобезопасных прототипов. Использует Threading Lock для предотвращения состояния гонки при чтении/копировании. """ def __init__(self, *args: object, **kwargs: object) -> None: """ Конструктор класса, реализующего потокобезопасное копирование. Гарантирует наличие замка у класса. :param args: Аргументы конструктора. :param kwargs: Ключевые аргументы конструктора. """ self.__dict__['_clone_lock'] = Lock() super().__init__(*args, **kwargs)
[документация] def clone(self) -> Self: """ Потокобезопасное поверхностное копирование. :return: Поверхностную копию объекта. """ if '_clone_lock' not in self.__dict__: self.__dict__['_clone_lock'] = Lock() with self._clone_lock: # type: ignore return copy(self)
[документация] def deep_clone(self) -> Self: """ Потокобезопасное глубокое копирование. :return: Полную копию объекта. """ if '_clone_lock' not in self.__dict__: self.__dict__['_clone_lock'] = Lock() with self._clone_lock: # type: ignore return deepcopy(self)
def __getstate__(self) -> dict[str, Any]: """ Удаляет замок из состояния при копировании через deepcopy, так как объекты Lock не могут быть сериализованы/скопированы глубоким методом. :return: Словарь с атрибутами и методами класса с удаленным из него замком. """ state = self.__dict__.copy() if '_clone_lock' in state: del state['_clone_lock'] return state def __setstate__(self, state: dict[str, Any]) -> None: """ Восстанавливает состояние и создает новый замок для клонированного объекта. :param state: Словарь с атрибутами и методами класса. """ self.__dict__.update(state) self.__dict__['_clone_lock'] = Lock()