mirror of
https://github.com/pyapp-kit/superqt.git
synced 2025-07-21 12:11:07 +02:00
Compare commits
2 Commits
15228eb989
...
4badc90425
Author | SHA1 | Date | |
---|---|---|---|
|
4badc90425 | ||
|
0585197383 |
@@ -29,13 +29,15 @@ SOFTWARE.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import weakref
|
||||
import warnings
|
||||
from concurrent.futures import Future
|
||||
from contextlib import suppress
|
||||
from enum import IntFlag, auto
|
||||
from functools import wraps
|
||||
from inspect import signature
|
||||
from types import MethodType
|
||||
from typing import TYPE_CHECKING, Any, Callable, Generic, TypeVar, overload
|
||||
from weakref import WeakKeyDictionary
|
||||
from weakref import WeakKeyDictionary, WeakMethod
|
||||
|
||||
from qtpy.QtCore import QObject, Qt, QTimer, Signal
|
||||
|
||||
@@ -214,18 +216,17 @@ class QSignalDebouncer(GenericSignalThrottler):
|
||||
def _weak_func(func: Callable[P, R]) -> Callable[P, R]:
|
||||
if isinstance(func, MethodType):
|
||||
# this is a bound method, we need to avoid strong references
|
||||
owner = func.__self__
|
||||
class_func = func.__func__
|
||||
try:
|
||||
instance_ref = weakref.ref(owner)
|
||||
weak_method = WeakMethod(func)
|
||||
except TypeError as e:
|
||||
raise TypeError(REF_ERROR) from e
|
||||
|
||||
def weak_func(*args, **kwargs):
|
||||
if (obj := instance_ref()) is None:
|
||||
raise RuntimeError("Method called on dead object")
|
||||
method = class_func.__get__(obj, type(obj))
|
||||
return method(*args, **kwargs)
|
||||
if method := weak_method():
|
||||
return method(*args, **kwargs)
|
||||
warnings.warn(
|
||||
"Method has been garbage collected", RuntimeWarning, stacklevel=2
|
||||
)
|
||||
|
||||
return weak_func
|
||||
|
||||
@@ -250,6 +251,9 @@ class ThrottledCallable(GenericSignalThrottler, Generic[P, R]):
|
||||
func = func.__func__
|
||||
|
||||
max_args = get_max_args(func)
|
||||
with suppress(TypeError, ValueError):
|
||||
self.__signature__ = signature(func)
|
||||
|
||||
self._func = _weak_func(func)
|
||||
self.__wrapped__ = self._func
|
||||
|
||||
|
@@ -213,7 +213,7 @@ def test_qthrottled_does_not_prevent_gc(qtbot):
|
||||
mock()
|
||||
|
||||
@qthrottled(timeout=1)
|
||||
def tmethod(self) -> None:
|
||||
def tmethod(self, x: int = 1) -> None:
|
||||
mock()
|
||||
|
||||
thing = Thing()
|
||||
@@ -225,6 +225,13 @@ def test_qthrottled_does_not_prevent_gc(qtbot):
|
||||
thing.tmethod()
|
||||
qtbot.waitUntil(thing.tmethod._future.done, timeout=2000)
|
||||
assert mock.call_count == 2
|
||||
|
||||
wm = thing.tmethod
|
||||
assert isinstance(wm, ThrottledCallable)
|
||||
del thing
|
||||
gc.collect()
|
||||
assert thing_ref() is None
|
||||
|
||||
with pytest.warns(RuntimeWarning, match="Method has been garbage collected"):
|
||||
wm()
|
||||
wm._set_future_result()
|
||||
|
Reference in New Issue
Block a user