Skip to content

from reactivity.helpers import *

Memoized

Bases: Subscribable, BaseComputation[T]

Source code in reactivity/helpers.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
 
class Memoized[T](Subscribable, BaseComputation[T]):
    def __init__(self, fn: Callable[[], T], *, context: Context | None = None):
        super().__init__(context=context)
        self.fn = fn
        self.is_stale = True
        self.cached_value: T

    def recompute(self):
        with self._enter():
            self.cached_value = self.fn()
            self.is_stale = False

    def trigger(self):
        self.invalidate()

    def __call__(self):
        self.track()
        if self.is_stale:
            self.recompute()
        return self.cached_value

    def invalidate(self):
        if not self.is_stale:
            del self.cached_value
            self.is_stale = True
            self.notify()

fn instance-attribute 

fn = fn

is_stale instance-attribute 

is_stale = True

cached_value instance-attribute 

cached_value: T

__init__ 

__init__(
    fn: Callable[[], T], *, context: Context | None = None
)
Source code in reactivity/helpers.py 
 9
10
11
12
13
 
def __init__(self, fn: Callable[[], T], *, context: Context | None = None):
    super().__init__(context=context)
    self.fn = fn
    self.is_stale = True
    self.cached_value: T

recompute 

recompute()
Source code in reactivity/helpers.py 
15
16
17
18
 
def recompute(self):
    with self._enter():
        self.cached_value = self.fn()
        self.is_stale = False

trigger 

trigger()
Source code in reactivity/helpers.py 
20
21
 
def trigger(self):
    self.invalidate()

__call__ 

__call__()
Source code in reactivity/helpers.py 
23
24
25
26
27
 
def __call__(self):
    self.track()
    if self.is_stale:
        self.recompute()
    return self.cached_value

invalidate 

invalidate()
Source code in reactivity/helpers.py 
29
30
31
32
33
 
def invalidate(self):
    if not self.is_stale:
        del self.cached_value
        self.is_stale = True
        self.notify()

MemoizedProperty

Bases: DescriptorMixin[Memoized[T]]

Source code in reactivity/helpers.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
class MemoizedProperty[T, I](DescriptorMixin[Memoized[T]]):
    def __init__(self, method: Callable[[I], T], *, context: Context | None = None):
        super().__init__()
        self.method = method
        self.context = context

    def _new(self, instance):
        return Memoized(self.method.__get__(instance), context=self.context)

    @overload
    def __get__(self, instance: None, owner: type[I]) -> Self: ...
    @overload
    def __get__(self, instance: I, owner: type[I]) -> T: ...

    def __get__(self, instance: I | None, owner):
        if instance is None:
            return self
        return self.find(instance)()

    __delete__ = __set__ = _not_implemented

method instance-attribute 

method = method

context instance-attribute 

context = context

__delete__ class-attribute instance-attribute 

__delete__ = _not_implemented

__set__ class-attribute instance-attribute 

__set__ = _not_implemented

__init__ 

__init__(
    method: Callable[[I], T],
    *,
    context: Context | None = None,
)
Source code in reactivity/helpers.py 
41
42
43
44
 
def __init__(self, method: Callable[[I], T], *, context: Context | None = None):
    super().__init__()
    self.method = method
    self.context = context

_new 

_new(instance)
Source code in reactivity/helpers.py 
46
47
 
def _new(self, instance):
    return Memoized(self.method.__get__(instance), context=self.context)

__get__ 

__get__(instance: None, owner: type[I]) -> Self

__get__(instance: I, owner: type[I]) -> T
Source code in reactivity/helpers.py 
54
55
56
57
 
def __get__(self, instance: I | None, owner):
    if instance is None:
        return self
    return self.find(instance)()

MemoizedMethod

Bases: DescriptorMixin[Memoized[T]]

Source code in reactivity/helpers.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
class MemoizedMethod[T, I](DescriptorMixin[Memoized[T]]):
    def __init__(self, method: Callable[[I], T], *, context: Context | None = None):
        super().__init__()
        self.method = method
        self.context = context

    def _new(self, instance):
        return Memoized(self.method.__get__(instance), context=self.context)

    @overload
    def __get__(self, instance: None, owner: type[I]) -> Self: ...
    @overload
    def __get__(self, instance: I, owner: type[I]) -> Memoized[T]: ...

    def __get__(self, instance: I | None, owner):
        if instance is None:
            return self
        return self.find(instance)

    __delete__ = __set__ = _not_implemented

method instance-attribute 

method = method

context instance-attribute 

context = context

__delete__ class-attribute instance-attribute 

__delete__ = _not_implemented

__set__ class-attribute instance-attribute 

__set__ = _not_implemented

__init__ 

__init__(
    method: Callable[[I], T],
    *,
    context: Context | None = None,
)
Source code in reactivity/helpers.py 
63
64
65
66
 
def __init__(self, method: Callable[[I], T], *, context: Context | None = None):
    super().__init__()
    self.method = method
    self.context = context

_new 

_new(instance)
Source code in reactivity/helpers.py 
68
69
 
def _new(self, instance):
    return Memoized(self.method.__get__(instance), context=self.context)

__get__ 

__get__(instance: None, owner: type[I]) -> Self

__get__(instance: I, owner: type[I]) -> Memoized[T]
Source code in reactivity/helpers.py 
76
77
78
79
 
def __get__(self, instance: I | None, owner):
    if instance is None:
        return self
    return self.find(instance)

DerivedProperty

Bases: DescriptorMixin[Derived[T]]

Source code in reactivity/helpers.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
 
class DerivedProperty[T, I](DescriptorMixin[Derived[T]]):
    def __init__(self, method: Callable[[I], T], check_equality=True, *, context: Context | None = None):
        super().__init__()
        self.method = method
        self.check_equality = check_equality
        self.context = context

    def _new(self, instance):
        return Derived(self.method.__get__(instance), self.check_equality, context=self.context)

    @overload
    def __get__(self, instance: None, owner: type[I]) -> Self: ...
    @overload
    def __get__(self, instance: I, owner: type[I]) -> T: ...

    def __get__(self, instance: I | None, owner):
        if instance is None:
            return self
        return self.find(instance)()

    __delete__ = __set__ = _not_implemented

method instance-attribute 

method = method

check_equality instance-attribute 

check_equality = check_equality

context instance-attribute 

context = context

__delete__ class-attribute instance-attribute 

__delete__ = _not_implemented

__set__ class-attribute instance-attribute 

__set__ = _not_implemented

__init__ 

__init__(
    method: Callable[[I], T],
    check_equality=True,
    *,
    context: Context | None = None,
)
Source code in reactivity/helpers.py 
85
86
87
88
89
 
def __init__(self, method: Callable[[I], T], check_equality=True, *, context: Context | None = None):
    super().__init__()
    self.method = method
    self.check_equality = check_equality
    self.context = context

_new 

_new(instance)
Source code in reactivity/helpers.py 
91
92
 
def _new(self, instance):
    return Derived(self.method.__get__(instance), self.check_equality, context=self.context)

__get__ 

__get__(instance: None, owner: type[I]) -> Self

__get__(instance: I, owner: type[I]) -> T
Source code in reactivity/helpers.py 
 99
100
101
102
 
def __get__(self, instance: I | None, owner):
    if instance is None:
        return self
    return self.find(instance)()

DerivedMethod

Bases: DescriptorMixin[Derived[T]]

Source code in reactivity/helpers.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
class DerivedMethod[T, I](DescriptorMixin[Derived[T]]):
    def __init__(self, method: Callable[[I], T], check_equality=True, *, context: Context | None = None):
        super().__init__()
        self.method = method
        self.check_equality = check_equality
        self.context = context

    def _new(self, instance):
        return Derived(self.method.__get__(instance), self.check_equality, context=self.context)

    @overload
    def __get__(self, instance: None, owner: type[I]) -> Self: ...
    @overload
    def __get__(self, instance: I, owner: type[I]) -> Derived[T]: ...

    def __get__(self, instance: I | None, owner):
        if instance is None:
            return self
        return self.find(instance)

    __delete__ = __set__ = _not_implemented

method instance-attribute 

method = method

check_equality instance-attribute 

check_equality = check_equality

context instance-attribute 

context = context

__delete__ class-attribute instance-attribute 

__delete__ = _not_implemented

__set__ class-attribute instance-attribute 

__set__ = _not_implemented

__init__ 

__init__(
    method: Callable[[I], T],
    check_equality=True,
    *,
    context: Context | None = None,
)
Source code in reactivity/helpers.py 
108
109
110
111
112
 
def __init__(self, method: Callable[[I], T], check_equality=True, *, context: Context | None = None):
    super().__init__()
    self.method = method
    self.check_equality = check_equality
    self.context = context

_new 

_new(instance)
Source code in reactivity/helpers.py 
114
115
 
def _new(self, instance):
    return Derived(self.method.__get__(instance), self.check_equality, context=self.context)

__get__ 

__get__(instance: None, owner: type[I]) -> Self

__get__(instance: I, owner: type[I]) -> Derived[T]
Source code in reactivity/helpers.py 
122
123
124
125
 
def __get__(self, instance: I | None, owner):
    if instance is None:
        return self
    return self.find(instance)

_not_implemented 

_not_implemented(self, instance, *_)
Source code in reactivity/helpers.py 
36
37
 
def _not_implemented(self, instance, *_):
    raise NotImplementedError(f"{type(instance).__name__}.{self.name} is read-only")  # todo: support optimistic updates