def __init__(self):
if CanvasManager.SINGLETON is not None:
raise Exception("Can only create one canvas manager.")
CanvasManager.SINGLETON = self
QtCore.QObject.__init__(self)
self.canvases = weakref.WeakValueDictionary()
python类WeakValueDictionary()的实例源码
def prune(self):
"""prune unreferenced, non-dirty states."""
ref_count = len(self)
dirty = [s.obj() for s in self.all_states() if s.modified]
# work around http://bugs.python.org/issue6149
keepers = weakref.WeakValueDictionary()
keepers.update(self)
self._dict.clear()
self._dict.update(keepers)
self.modified = bool(dirty)
return ref_count - len(self)
def __init__(self):
self.__slots = WeakValueDictionary()
def __new__(typ, *args, **kw):
#check the class has an __instances__ dict, if not,
#create it and initialize __instance_id.
try:
typ.__instances__
except AttributeError:
typ.__instance_id = 0
typ.__instances__ = weakref.WeakValueDictionary()
obj = object.__new__(typ, *args, **kw)
obj.id = typ.__instance_id
typ.__instances__[typ.__instance_id] = obj
typ.__instance_id += 1
return obj
def __init__(self, function, capacity=50):
self.cache = weakref.WeakValueDictionary()
self.next = self.previous = self
self.function = function
self.capacity = capacity
def observable__getattribute__(f):
""" decoratore per Observable.__getattribute__"""
# https://bugs.python.org/issue3445
# from functools import wraps
# @wraps(f)
def wrapper(*args, **kwargs):
instance, attribute = args[:2]
result = f(*args, **kwargs)
if attribute == '_Observable__observers' and result is None:
result = weakref.WeakValueDictionary()
setattr(instance, '_Observable__observers', result)
return result
return wrapper
def global_cache(srctype, ffi, funcname, *args, **kwds):
key = kwds.pop('key', (funcname, args))
assert not kwds
try:
return ffi._backend.__typecache[key]
except KeyError:
pass
except AttributeError:
# initialize the __typecache attribute, either at the module level
# if ffi._backend is a module, or at the class level if ffi._backend
# is some instance.
if isinstance(ffi._backend, types.ModuleType):
ffi._backend.__typecache = weakref.WeakValueDictionary()
else:
type(ffi._backend).__typecache = weakref.WeakValueDictionary()
try:
res = getattr(ffi._backend, funcname)(*args)
except NotImplementedError as e:
raise NotImplementedError("%s: %r: %s" % (funcname, srctype, e))
# note that setdefault() on WeakValueDictionary is not atomic
# and contains a rare bug (http://bugs.python.org/issue19542);
# we have to use a lock and do it ourselves
cache = ffi._backend.__typecache
with global_lock:
res1 = cache.get(key)
if res1 is None:
cache[key] = res
return res
else:
return res1
def getReference(self, obj):
return weakref.WeakValueDictionary(obj)
def __init__(self, root_name, int_starts_with):
# dict of {'id': obj } for items to be evaluated eventually.
self.object_to_eval = weakref.WeakValueDictionary()
self.evaluated_items = {} # dict of {'path': evaluated object}
self.root_name = root_name
self.int_regex = re.compile('^{}([\d]+)$'.format(int_starts_with))
def _load_wrapper(self):
self._threadLock.acquire()
paths_to_eval = tuple(
set(i._item_key for i in self._registry.object_to_eval.values()))
self._registry.object_to_eval = weakref.WeakValueDictionary()
new_items = self.load(paths_to_eval)
if isinstance(new_items, MutableMapping):
self._registry.evaluated_items.update(new_items)
self._threadLock.release()
else:
self._threadLock.release()
raise Exception(
"load method needs to return a dictionary of {path: value}")
def __init__(self, rowClasses):
"""
Initialize me against a database.
@param rowClasses: a list of row class objects that describe the
database schema.
"""
self.rowCache = weakref.WeakValueDictionary() # does not hold references to cached rows.
self.rowClasses = rowClasses
self.schema = {}
self._populate()
def __setstate__(self, state):
self.__dict__ = state
self.rowCache = weakref.WeakValueDictionary()
self._populate()
def __init__(self):
self.__memo = weakref.WeakValueDictionary()
def cache(func=..., *, key='received'):
"""Memoize the function using weak references.
Once the decorated function has been called with a given signature,
it will return the same object for all subsequent calls with that
signature, as long as another non-weak reference to the object still
exists.
>>> class WeakReferenceableList(list):
... pass # Built-in lists can't be weak-referenced
>>> @cache
... def say(word, also='lmao'):
... return WeakReferenceableList([word, also])
>>> say('ayy') is say('ayy')
True
>>> say('ayy') is say('ayy', 'lmao')
True
>>> say('ayy') is say('ayy', also='lmao')
True
>>> say('ayy') is say(also='lmao', word='ayy')
True
"""
return memoize(func, cache=weakref.WeakValueDictionary(), key=key)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.__cache = weakref.WeakValueDictionary()
def __init__(self, context=None, **kwds):
super(DjangoTranslator, self).__init__(**kwds)
if context is not None:
self.context = context
self._django_hasher_cache = weakref.WeakKeyDictionary()
self._passlib_hasher_cache = weakref.WeakValueDictionary()
def __init__(self, name, aliases=(), defaults=None):
self.name = name
self.aliases = aliases
#: Maps (src, dst) -> transformation function
self.funcs = {}
#: Maps defaults variable names to values
self.defaults = defaults or {}
#: Maps (src, dst) -> self
#: Used as a convenience dictionary to be composed by ContextChain
self.relation_to_context = weakref.WeakValueDictionary()
def __init__(self, identity_map=None):
if identity_map is not None:
self.identity_map = identity_map
else:
self.identity_map = weakref.WeakValueDictionary()
self.attributes = global_attributes
self.new = util.HashSet(ordered = True)
self.dirty = util.HashSet()
self.modified_lists = util.HashSet()
self.deleted = util.HashSet()
def __init__(self, echo = False, use_threadlocal = True):
self._threadconns = weakref.WeakValueDictionary()
self._use_threadlocal = use_threadlocal
self._echo = echo
def global_cache(srctype, ffi, funcname, *args, **kwds):
key = kwds.pop('key', (funcname, args))
assert not kwds
try:
return ffi._backend.__typecache[key]
except KeyError:
pass
except AttributeError:
# initialize the __typecache attribute, either at the module level
# if ffi._backend is a module, or at the class level if ffi._backend
# is some instance.
if isinstance(ffi._backend, types.ModuleType):
ffi._backend.__typecache = weakref.WeakValueDictionary()
else:
type(ffi._backend).__typecache = weakref.WeakValueDictionary()
try:
res = getattr(ffi._backend, funcname)(*args)
except NotImplementedError as e:
raise NotImplementedError("%s: %r: %s" % (funcname, srctype, e))
# note that setdefault() on WeakValueDictionary is not atomic
# and contains a rare bug (http://bugs.python.org/issue19542);
# we have to use a lock and do it ourselves
cache = ffi._backend.__typecache
with global_lock:
res1 = cache.get(key)
if res1 is None:
cache[key] = res
return res
else:
return res1
