Исходный код 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()