Compare commits

...

2 Commits

Author SHA1 Message Date
Talley Lambert
4badc90425 add test for warning 2024-06-03 10:06:22 -04:00
Talley Lambert
0585197383 use weakmethod, add signature 2024-06-03 09:55:44 -04:00
2 changed files with 21 additions and 10 deletions

View File

@@ -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

View File

@@ -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()