"""Weak reference support for Python.

This module is an implementation of PEP 205:

https://peps.python.org/pep-0205/

"""

# Naming convention: Variables named "wr" are weak reference objects;

# they are called this instead of "ref" to avoid name collisions with

# the module-global ref() function imported from _weakref.

from _weakref import (

getweakrefcount,

getweakrefs,

ref,

proxy,

CallableProxyType,

ProxyType,

ReferenceType,

_remove_dead_weakref)

from _weakrefset import WeakSet

import _collections_abc # Import after _weakref to avoid circular import.

import copyreg

import sys

import itertools

ProxyTypes = (ProxyType, CallableProxyType)

__all__ = ["ref", "proxy", "getweakrefcount", "getweakrefs",

"WeakKeyDictionary", "ReferenceType", "ProxyType",

"CallableProxyType", "ProxyTypes", "WeakValueDictionary",

"WeakSet", "WeakMethod", "finalize"]

_collections_abc.MutableSet.register(WeakSet)

# Pickle reconstructs a reference to a dead object as this singleton -- all

# dead references are interchangeable. Born dead: nothing else holds the set.

_dead_ref = ref(set())

class WeakMethod(ref):

"""

A custom `weakref.ref` subclass which simulates a weak reference to

a bound method, working around the lifetime problem of bound methods.

"""

__slots__ = "_func_ref", "_meth_type", "_alive", "__weakref__"

def __new__(cls, meth, callback=None):

try:

obj = meth.__self__

func = meth.__func__

except AttributeError:

raise TypeError("argument should be a bound method, not {}"

.format(type(meth))) from None

def _cb(arg):

# The self-weakref trick is needed to avoid creating a reference

# cycle.

self = self_wr()

if self._alive:

self._alive = False

if callback is not None:

callback(self)

self = ref.__new__(cls, obj, _cb)

self._func_ref = ref(func, _cb)

self._meth_type = type(meth)

self._alive = True

self_wr = ref(self)

return self

def __call__(self):

obj = super().__call__()

func = self._func_ref()

if obj is None or func is None:

return None

return self._meth_type(func, obj)

def __eq__(self, other):

if isinstance(other, WeakMethod):

if not self._alive or not other._alive:

return self is other

return ref.__eq__(self, other) and self._func_ref == other._func_ref

return NotImplemented

def __ne__(self, other):

if isinstance(other, WeakMethod):

if not self._alive or not other._alive:

return self is not other

return ref.__ne__(self, other) or self._func_ref != other._func_ref

return NotImplemented

__hash__ = ref.__hash__

# Pickling support for weak methods. A WeakMethod is reduced to a weak

# reference to its instance; the bound method is rebuilt on unpickling by

# looking the function up on the instance. The callback is not preserved.

def _restore_weakmethod(obj_ref, name):

return WeakMethod(getattr(obj_ref(), name))

def _reduce_weakmethod(self):

meth = self()

if meth is None:

raise TypeError(

"cannot pickle a weak reference to a dead object")

return (_restore_weakmethod, (ref(meth.__self__), meth.__func__.__name__))

copyreg.pickle(WeakMethod, _reduce_weakmethod)

class WeakValueDictionary(_collections_abc.MutableMapping):

"""Mapping class that references values weakly.

Entries in the dictionary will be discarded when no strong

reference to the value exists anymore

"""

# Slots give __getstate__() a uniform (dict, slots) shape for __reduce__.

# __dict__ carries user attributes; the removal callback needs __weakref__.

__slots__ = '__dict__', '__weakref__', 'data', '_remove'

# We inherit the constructor without worrying about the input

# dictionary; since it uses our .update() method, we get the right

# checks (if the other dictionary is a WeakValueDictionary,

# objects are unwrapped on the way out, and we always wrap on the

# way in).

def __init__(self, other=(), /, **kw):

def remove(wr, selfref=ref(self), _atomic_removal=_remove_dead_weakref):

self = selfref()

if self is not None:

# Atomic removal is necessary since this function

# can be called asynchronously by the GC

_atomic_removal(self.data, wr.key)

self._remove = remove

self.data = {}

self.update(other, **kw)

def __getitem__(self, key):

o = self.data[key]()

if o is None:

raise KeyError(key)

else:

return o

def __delitem__(self, key):

del self.data[key]

def __len__(self):

return len(self.data)

def __contains__(self, key):

try:

o = self.data[key]()

except KeyError:

return False

return o is not None

def __repr__(self):

return "<%s at %#x>" % (self.__class__.__name__, id(self))

def __setitem__(self, key, value):

self.data[key] = KeyedRef(value, self._remove, key)

def copy(self):

new = WeakValueDictionary()

for key, wr in self.data.copy().items():

o = wr()

if o is not None:

new[key] = o

return new

__copy__ = copy

def __deepcopy__(self, memo):

from copy import deepcopy

new = self.__class__()

for key, wr in self.data.copy().items():

o = wr()

if o is not None:

new[deepcopy(key, memo)] = o

return new

def __reduce__(self):

# Pickle the values as weak references; data and _remove are rebuilt

# by the reconstructor. Copying uses __copy__/__deepcopy__ instead.

dict, slots = self.__getstate__()

for name in ('data', '_remove'):

slots.pop(name, None)

state = (dict, slots) if dict or slots else None

items = [(key, ref(o)) for key, o in self.items()]

return (_restore_weakvaluedict, (self.__class__, items), state)

def get(self, key, default=None):

try:

wr = self.data[key]

except KeyError:

return default

else:

o = wr()

if o is None:

# This should only happen

return default

else:

return o

def items(self):

for k, wr in self.data.copy().items():

v = wr()

if v is not None:

yield k, v

def keys(self):

for k, wr in self.data.copy().items():

if wr() is not None:

yield k

__iter__ = keys

def itervaluerefs(self):

"""Return an iterator that yields the weak references to the values.

The references are not guaranteed to be 'live' at the time

they are used, so the result of calling the references needs

to be checked before being used. This can be used to avoid

creating references that will cause the garbage collector to