Skip to content

from reactivity.helpers import *

Memoized

Bases: Subscribable, BaseComputation[T]

Source code in .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
def recompute(self):
    with self._enter():
        self.cached_value = self.fn()
        self.is_stale = False

trigger

trigger()
Source code in .venv/lib/python3.12/site-packages/reactivity/helpers.py
def trigger(self):
    self.invalidate()

__call__

__call__()
Source code in .venv/lib/python3.12/site-packages/reactivity/helpers.py
def __call__(self):
    self.track()
    if self.is_stale:
        self.recompute()
    return self.cached_value

invalidate

invalidate()
Source code in .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
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 .venv/lib/python3.12/site-packages/reactivity/helpers.py
def _not_implemented(self, instance, *_):
    raise NotImplementedError(f"{type(instance).__name__}.{self.name} is read-only")  # todo: support optimistic updates