我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用weakref.ref()。
def __init__(self, *children): # Populate the list of children. self._children = [] parent = weakref.ref(self) previous = self for child in children: if isinstance(child, str): # bytes child = Raw(child, self.arch, self.os) elif not isinstance(child, Shellcode): raise TypeError( "Expected Shellcode, got %s instead" % type(child)) elif child.parent: msg = "Already had a parent: %r" % child.parent warnings.warn(msg, ShellcodeWarning) child._parent = parent self._children.append(child) previous._check_platform(child) previous = child # Dark magic to implement the metadata combination feature.
def test_WeakObject(self): """ Tests basic extended weak object support. """ obj = TestClass() ref = weakref.ref(obj) objref = weakobj.objectref(obj) objref2 = weakobj.objectref(obj) self.assertEquals(objref, objref2) self.assertEquals(objref.foo(), obj.foo()) # Delete what should be the only reference to the original object. del obj self.assertEqual(ref(), None) try: objref.foo() except weakref.ReferenceError: pass else: self.fail('Weak object should be invalidated')
def set_self(self, receiver, slf): # pragma: no cover """ Set the self instance on a receiver. .. deprecated:: 0.0.1 :param receiver: Receiver function. :param slf: Self instance """ with self.lock: lookup_key = _make_id(receiver) for key, _ in self.receivers: if lookup_key == key: ref = weakref.ref slf = ref(slf) self.self_refs[lookup_key] = slf return raise Exception('Receiver is not yet known! You registered too early!')
def _shutdown_executors(): # Copy the set. Stopping threads has the side effect of removing executors. executors = list(_EXECUTORS) # First signal all executors to close... for ref in executors: executor = ref() if executor: executor.close() # ...then try to join them. for ref in executors: executor = ref() if executor: executor.join(1) executor = None
def __init__(self, signal, delay=0.3, rateLimit=0, slot=None): """Initialization arguments: signal - a bound Signal or pyqtSignal instance delay - Time (in seconds) to wait for signals to stop before emitting (default 0.3s) slot - Optional function to connect sigDelayed to. rateLimit - (signals/sec) if greater than 0, this allows signals to stream out at a steady rate while they are being received. """ QtCore.QObject.__init__(self) signal.connect(self.signalReceived) self.signal = signal self.delay = delay self.rateLimit = rateLimit self.args = None self.timer = ThreadsafeTimer.ThreadsafeTimer() self.timer.timeout.connect(self.flush) self.block = False self.slot = weakref.ref(slot) self.lastFlushTime = None if slot is not None: self.sigDelayed.connect(slot)
def getViewWidget(self): """ Return the view widget for this item. If the scene has multiple views, only the first view is returned. The return value is cached; clear the cached value with forgetViewWidget(). If the view has been deleted by Qt, return None. """ if self._viewWidget is None: scene = self.scene() if scene is None: return None views = scene.views() if len(views) < 1: return None self._viewWidget = weakref.ref(self.scene().views()[0]) v = self._viewWidget() if v is not None and not isQObjectAlive(v): return None return v
def getViewBox(self): """ Return the first ViewBox or GraphicsView which bounds this item's visible space. If this item is not contained within a ViewBox, then the GraphicsView is returned. If the item is contained inside nested ViewBoxes, then the inner-most ViewBox is returned. The result is cached; clear the cache with forgetViewBox() """ if self._viewBox is None: p = self while True: try: p = p.parentItem() except RuntimeError: ## sometimes happens as items are being removed from a scene and collected. return None if p is None: vb = self.getViewWidget() if vb is None: return None else: self._viewBox = weakref.ref(vb) break if hasattr(p, 'implements') and p.implements('ViewBox'): self._viewBox = weakref.ref(p) break return self._viewBox() ## If we made it this far, _viewBox is definitely not None
def __init__(self, view, pos, color, movable=True, scale=10, pen='w'): self.movable = movable self.moving = False self.view = weakref.ref(view) self.scale = scale self.color = color self.pen = fn.mkPen(pen) self.hoverPen = fn.mkPen(255,255,0) self.currentPen = self.pen self.pg = QtGui.QPainterPath(QtCore.QPointF(0,0)) self.pg.lineTo(QtCore.QPointF(-scale/3**0.5, scale)) self.pg.lineTo(QtCore.QPointF(scale/3**0.5, scale)) self.pg.closeSubpath() QtGui.QGraphicsWidget.__init__(self) self.setPos(pos[0], pos[1]) if self.movable: self.setZValue(1) else: self.setZValue(0)
def test_getViewWidget_deleted(): view = pg.PlotWidget() item = pg.InfiniteLine() view.addItem(item) assert item.getViewWidget() is view # Arrange to have Qt automatically delete the view widget obj = pg.QtGui.QWidget() view.setParent(obj) del obj gc.collect() assert not pg.Qt.isQObjectAlive(view) assert item.getViewWidget() is None #if __name__ == '__main__': #view = pg.PlotItem() #vref = weakref.ref(view) #item = pg.InfiniteLine() #view.addItem(item) #del view #gc.collect()
def linkToView(self, view): """Link this axis to a ViewBox, causing its displayed range to match the visible range of the view.""" oldView = self.linkedView() self._linkedView = weakref.ref(view) if self.orientation in ['right', 'left']: if oldView is not None: oldView.sigYRangeChanged.disconnect(self.linkedViewChanged) view.sigYRangeChanged.connect(self.linkedViewChanged) else: if oldView is not None: oldView.sigXRangeChanged.disconnect(self.linkedViewChanged) view.sigXRangeChanged.connect(self.linkedViewChanged) if oldView is not None: oldView.sigResized.disconnect(self.linkedViewChanged) view.sigResized.connect(self.linkedViewChanged)
def getState(self, copy=True): """Return the current state of the ViewBox. Linked views are always converted to view names in the returned state.""" state = self.state.copy() views = [] for v in state['linkedViews']: if isinstance(v, weakref.ref): v = v() if v is None or isinstance(v, basestring): views.append(v) else: views.append(v.name) state['linkedViews'] = views if copy: return deepcopy(state) else: return state
def __init__(self, curve, index=0, pos=None, rotate=True): """Position can be set either as an index referring to the sample number or the position 0.0 - 1.0 If *rotate* is True, then the item rotates to match the tangent of the curve. """ GraphicsObject.__init__(self) #QObjectWorkaround.__init__(self) self._rotate = rotate self.curve = weakref.ref(curve) self.setParentItem(curve) self.setProperty('position', 0.0) self.setProperty('index', 0) if hasattr(self, 'ItemHasNoContents'): self.setFlags(self.flags() | self.ItemHasNoContents) if pos is not None: self.setPos(pos) else: self.setIndex(index)
def __init__(self): self._pid = os.getpid() self._loop_lock = Lock() self._started = False self._shutdown = False self._thread = None self._timers = TimerManager() try: dispatcher = self._loop_dispatch_class() dispatcher.validate() log.debug("Validated loop dispatch with %s", self._loop_dispatch_class) except Exception: log.exception("Failed validating loop dispatch with %s. Using busy wait execution instead.", self._loop_dispatch_class) dispatcher.close() dispatcher = _BusyWaitDispatcher() self._loop_dispatcher = dispatcher atexit.register(partial(_cleanup, weakref.ref(self)))
def custom_payload(self): """ The custom payload returned from the server, if any. This will only be set by Cassandra servers implementing a custom QueryHandler, and only for protocol_version 4+. Ensure the future is complete before trying to access this property (call :meth:`.result()`, or after callback is invoked). Otherwise it may throw if the response has not been received. :return: :ref:`custom_payload`. """ # TODO: When timers are introduced, just make this wait if not self._event.is_set(): raise DriverException("custom_payload cannot be retrieved before ResponseFuture is finalized") return self._custom_payload
def contextual_connect(self, **kw): if not hasattr(self._connections, 'conn'): connection = None else: connection = self._connections.conn() if connection is None or connection.closed: # guards against pool-level reapers, if desired. # or not connection.connection.is_valid: connection = self._tl_connection_cls( self, self._wrap_pool_connect( self.pool.connect, connection), **kw) self._connections.conn = weakref.ref(connection) return connection._increment_connect()
def _cleanup(self, ref): """Weakref callback cleanup. This callable cleans out the state when it is being garbage collected. this _cleanup **assumes** that there are no strong refs to us! Will not work otherwise! """ instance_dict = self._instance_dict() if instance_dict is not None: instance_dict._fast_discard(self) del self._instance_dict # we can't possibly be in instance_dict._modified # b.c. this is weakref cleanup only, that set # is strong referencing! # assert self not in instance_dict._modified self.session_id = self._strong_obj = None del self.obj
def _listen(cls, event_key, propagate=True, **kw): target, identifier, fn = \ event_key.dispatch_target, event_key.identifier, \ event_key._listen_fn def listen(target_cls, *arg): listen_cls = target() if propagate and issubclass(target_cls, listen_cls): return fn(target_cls, *arg) elif not propagate and target_cls is listen_cls: return fn(target_cls, *arg) def remove(ref): key = event.registry._EventKey( None, identifier, listen, instrumentation._instrumentation_factory) getattr(instrumentation._instrumentation_factory.dispatch, identifier).remove(key) target = weakref.ref(target.class_, remove) event_key.\ with_dispatch_target(instrumentation._instrumentation_factory).\ with_wrapper(listen).base_listen(**kw)
def before_flush(self, session, flush_context, instances): """Execute before flush process has started. :param session: The target :class:`.Session`. :param flush_context: Internal :class:`.UOWTransaction` object which handles the details of the flush. :param instances: Usually ``None``, this is the collection of objects which can be passed to the :meth:`.Session.flush` method (note this usage is deprecated). .. seealso:: :meth:`~.SessionEvents.after_flush` :meth:`~.SessionEvents.after_flush_postexec` :ref:`session_persistence_events` """
def after_flush(self, session, flush_context): """Execute after flush has completed, but before commit has been called. Note that the session's state is still in pre-flush, i.e. 'new', 'dirty', and 'deleted' lists still show pre-flush state as well as the history settings on instance attributes. :param session: The target :class:`.Session`. :param flush_context: Internal :class:`.UOWTransaction` object which handles the details of the flush. .. seealso:: :meth:`~.SessionEvents.before_flush` :meth:`~.SessionEvents.after_flush_postexec` :ref:`session_persistence_events` """
def after_flush_postexec(self, session, flush_context): """Execute after flush has completed, and after the post-exec state occurs. This will be when the 'new', 'dirty', and 'deleted' lists are in their final state. An actual commit() may or may not have occurred, depending on whether or not the flush started its own transaction or participated in a larger transaction. :param session: The target :class:`.Session`. :param flush_context: Internal :class:`.UOWTransaction` object which handles the details of the flush. .. seealso:: :meth:`~.SessionEvents.before_flush` :meth:`~.SessionEvents.after_flush` :ref:`session_persistence_events` """
def before_attach(self, session, instance): """Execute before an instance is attached to a session. This is called before an add, delete or merge causes the object to be part of the session. .. versionadded:: 0.8. Note that :meth:`~.SessionEvents.after_attach` now fires off after the item is part of the session. :meth:`.before_attach` is provided for those cases where the item should not yet be part of the session state. .. seealso:: :meth:`~.SessionEvents.after_attach` :ref:`session_lifecycle_events` """
def _stored_in_collection(event_key, owner): key = event_key._key dispatch_reg = _key_to_collection[key] owner_ref = owner.ref listen_ref = weakref.ref(event_key._listen_fn) if owner_ref in dispatch_reg: return False dispatch_reg[owner_ref] = listen_ref listener_to_key = _collection_to_key[owner_ref] listener_to_key[listen_ref] = key return True
def add_item(self, item): # protect against class registration race condition against # asynchronous garbage collection calling _remove_item, # [ticket:3208] modules = set([ cls.__module__ for cls in [ref() for ref in self.contents] if cls is not None]) if item.__module__ in modules: util.warn( "This declarative base already contains a class with the " "same class name and module name as %s.%s, and will " "be replaced in the string-lookup table." % ( item.__module__, item.__name__ ) ) self.contents.add(weakref.ref(item, self._remove_item))
def checkout(cls, pool): rec = pool._do_get() try: dbapi_connection = rec.get_connection() except: with util.safe_reraise(): rec.checkin() echo = pool._should_log_debug() fairy = _ConnectionFairy(dbapi_connection, rec, echo) rec.fairy_ref = weakref.ref( fairy, lambda ref: _finalize_fairy and _finalize_fairy( dbapi_connection, rec, pool, ref, echo) ) _refs.add(rec) if echo: pool.logger.debug("Connection %r checked out from pool", dbapi_connection) return fairy
def format(self, format='%(name)s size=%(size)d flat=%(flat)d', detail=-1, order_by='size', indent=''): '''Formats the size information of the object and of all sized referents as a string. *format='%(name)s...'* -- specifies the format string per instance, valid interpolation parameters are 'name', 'size' and 'flat' *detail=-1* -- detail level up to which referents are printed (-1 for unlimited) *order_by='size'* -- sort order of referents, valid choices are 'name', 'size' or 'flat' *indent=''* -- optional indentation ''' lines = [indent + (format % dict(size=self.size, flat=self.flat, name=self.name))] if detail and self.refs: refs = sorted(self.refs, key=lambda x: getattr(x, order_by), reverse=order_by in ('size', 'flat')) lines += [ref.format(format=format, detail=detail-1, order_by=order_by, indent=indent+' ') for ref in refs] return '\n'.join(lines)
def named_refs(obj, **opts): """Returns (a generator for) all named *referents* of an object (re-using functionality from **asizeof**). See function **basicsize** for a description of the options. Does not return un-named *referents*, e.g. objects in a list. """ t = _typedefof(obj, **opts) if t: r = t.refs if r and _iscallable(r): for nr in r(obj, True): try: yield nr.name, nr.ref except AttributeError: pass
def inc(self): # Copy these references so on_thread_died needn't close over self ident = self.ident _counters = self._counters tid = ident.get() _counters.setdefault(tid, 0) _counters[tid] += 1 if not ident.watching(): # Before the tid is possibly reused, remove it from _counters def on_thread_died(ref): ident.unwatch(tid) _counters.pop(tid, None) ident.watch(on_thread_died) return _counters[tid]
def test_task_refcounting(self): # On CPython, tasks and their arguments should be released immediately # without waiting for garbage collection. @gen.engine def f(): class Foo(object): pass arg = Foo() self.arg_ref = weakref.ref(arg) task = gen.Task(self.io_loop.add_callback, arg=arg) self.task_ref = weakref.ref(task) yield task self.stop() self.run_gen(f) self.assertIs(self.arg_ref(), None) self.assertIs(self.task_ref(), None)
def __setitem__( self, k, v, isinstance=isinstance ): if isinstance(v,_ParseResultsWithOffset): self.__tokdict[k] = self.__tokdict.get(k,list()) + [v] sub = v[0] elif isinstance(k,(int,slice)): self.__toklist[k] = v sub = v else: self.__tokdict[k] = self.__tokdict.get(k,list()) + [_ParseResultsWithOffset(v,0)] sub = v if isinstance(sub,ParseResults): sub.__parent = wkref(self)
def __iadd__( self, other ): if other.__tokdict: offset = len(self.__toklist) addoffset = lambda a: offset if a<0 else a+offset otheritems = other.__tokdict.items() otherdictitems = [(k, _ParseResultsWithOffset(v[0],addoffset(v[1])) ) for (k,vlist) in otheritems for v in vlist] for k,v in otherdictitems: self[k] = v if isinstance(v[0],ParseResults): v[0].__parent = wkref(self) self.__toklist += other.__toklist self.__accumNames.update( other.__accumNames ) return self
def __setstate__(self,state): self.__toklist = state[0] (self.__tokdict, par, inAccumNames, self.__name) = state[1] self.__accumNames = {} self.__accumNames.update(inAccumNames) if par is not None: self.__parent = wkref(par) else: self.__parent = None
def test_WeakObjectCallback(self): """ Test that the weak object callback occurs as expected. """ obj = TestClass() results = set() def callback(target): results.add(id(target)) ref = weakobj.objectref(obj, callback) expected = set([id(ref)]) del obj self.assertEqual(results, expected)
def _make_callback(obj, callback): # Wraps the given callback function for use with weakref.ref callbacks, # ensuring that a reference cycle is not created between the wrapper object # and the weak reference. if callback is not None: self_ref = weakref.ref(obj) def delref(ref): callback(self_ref()) return delref else: return None
def __init__(self, ref, callback=None): self.__ref__ = weakref.ref(ref, _make_callback(self, callback))
def __getref__(self): ref = self.__ref__() if ref is None: raise weakref.ReferenceError('weakly-referenced object no longer exists') return ref
