Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class in_init(object):
# class to handle setting and unsetting self.__in_init bool."
def __init__(self, obj):
self.obj = obj
def __enter__(self):
self.obj._in_init = True
def __exit__(self, type, value, traceback):
self.obj._in_init = False
class ProxyDict(OrderedDict, CoW):
def __init__(self, d={}):
# Recursively proxify first
# import sys
# sys.setrecursionlimit(100000)
d = dict((item, proxify(d[item])) for item in d)
CoW.__init__(self)
with in_init(self):
self._hash_cache = None
# If we're already a Proxy Dict, just pass through
if type(d) in [ProxyDict, OrderedDict]:
return OrderedDict.__init__(self, d)
# Sorting this by default to reduce burden on hash
return self._hash_cache
def __copy__(self):
return ProxySet(self)
def __getattribute__(self, key):
# If we don't need to proxy this call, just do it.
if key not in proxy_list_inplace_set:
return super().__getattribute__(key)
# Proxy this call
return lambda *args, **kwargs: list_do_generic_call(
self, key, *args, **kwargs)
class ProxyStr(str, CoW):
def __init__(self, *args, **kwargs):
# str keeps it's setup all in __new__
CoW.__init__(self, *args, **kwargs)
self._hash_cache = None
def __hash__(self):
if self._hash_cache is None:
self._hash_cache = super().__hash__()
return self._hash_cache
def __copy__(self):
return ProxyStr(self)
class ProxyTuple(ProxyList, CoW):
def __init__(self, tup):
return lambda *args, **kwargs: list_do_generic_call(
self, key, *args, **kwargs)
def __getitem__(self, key):
# Proxy to call first, which will come back
# to our first subclass of list.
return CoW.__getitem__(self, key)
# Things list class does in-place that we need to watch out for
proxy_list_inplace_list = ["append", "clear", "extend",
"insert", "pop", "remove",
"reverse", "sort"]
class ProxyList(list, CoW):
def __init__(self, iterable):
# Recursively transform into CoW objects
proxified = [proxify(item) for item in iterable]
CoW.__init__(self, proxified)
list.__init__(self, proxified)
self._hash_cache = None
def __hash__(self):
if self._hash_cache is None:
self._hash_cache = hash(tuple(self))
return self._hash_cache
def __copy__(self):
return ProxyList(self)
def __getitem__(self, key):
# Proxy to call first, which will come back
# to our first subclass of list.
return CoW.__getitem__(self, key)
def __setitem__(self, *args, **kwargs):
if not self._in_init:
return CoW.__setitem__(self, *args, **kwargs)
else:
return super().__setitem__(*args, **kwargs)
def __iadd__(self, *args, **kwargs):
# Invalidate our cache
self._hash_cache = None
# Can't overload special methods through getattribute,
# so just proxying here.
return list_do_generic_call(self, "__iadd__", *args, **kwargs)
proxy_list_inplace_set = ["add", "clear", "difference_update", "discard",
"intersection_update", "pop", "remove",
"symmetric_difference_update", "update"]
class ProxySet(set, CoW):
def __init__(self, *args, **kwargs):
set.__init__(self, *args, **kwargs)
CoW.__init__(self, *args, **kwargs)
self._hash_cache = None
def __hash__(self):
if self._hash_cache is None:
self._hash_cache = hash(tuple(self))
return self._hash_cache
def __copy__(self):
return ProxySet(self)
def __getattribute__(self, key):
# If we don't need to proxy this call, just do it.
if key not in proxy_list_inplace_set:
args = [proxify(arg) for arg in args]
kwargs = dict((key, proxify(val)) for key, val in kwargs.items())
# Run this call in-place
ret = getattr(super(my_type, my_copy), method_name)(*args, **kwargs)
# Only bother calling update if we had to copy
if copy_required:
# Call our cb function
for func in self._flyweight_cb_func:
func(my_copy)
else:
# Update the cache with our new value
# Remove ref to this object since hash changed
del CoW._flyweight_cache[type(my_copy)][my_old_hash]
# Invalidate hash cache if one exists
if hasattr(my_copy, "_hash_cache"):
my_copy._hash_cache = None
# Add new ref
CoW._flyweight_cache[type(my_copy)][hash(my_copy)] = my_copy
# Return any value that might be returned
return ret
def __copy__(self):
# Perform fast copy of attribute pointers in self.
obj = type(self).__new__(type(self))
CoW.__init__(obj, self)
return obj
def __getattribute__(self, key):
# Grab the value
value = super().__getattribute__(key)
# Don't proxy our own stuff
# If this is a function, don't try to copy it
if key in flyweight_ignored_keys or type(value) in \
[types.BuiltinFunctionType,
types.MethodType,
types.FunctionType]:
return value
# Writing callback value so we can be notified
# if this object updates in place.
if issubclass(type(value), CoW):
self.__cow_add_cb_pointer(value)
return value