Skip to content

from reactivity import *

__all__ module-attribute 

__all__ = [
    "async_derived",
    "async_effect",
    "batch",
    "derived",
    "derived_method",
    "derived_property",
    "effect",
    "memoized",
    "memoized_method",
    "memoized_property",
    "new_context",
    "reactive",
    "signal",
    "state",
]

async_derived 

async_derived[T](
    fn: Callable[[], Awaitable[T]],
    /,
    check_equality=True,
    *,
    context: Context | None = None,
    task_factory: TaskFactory | None = None,
) -> AsyncDerived[T]

async_derived[T](
    *,
    check_equality=True,
    context: Context | None = None,
    task_factory: TaskFactory | None = None,
) -> Callable[
    [Callable[[], Awaitable[T]]], AsyncDerived[T]
]
Source code in reactivity/_curried.py 
123
124
125
126
 
def async_derived[T](fn: Callable[[], Awaitable[T]] = __, /, check_equality=True, *, context: Context | None = None, task_factory: TaskFactory | None = None):  # type: ignore
    if fn is __:
        return lambda fn: AsyncDerived(fn, check_equality, context=context, task_factory=task_factory or default_task_factory)
    return AsyncDerived(fn, check_equality, context=context, task_factory=task_factory or default_task_factory)

async_effect 

async_effect[T](
    fn: Callable[[], Awaitable[T]],
    /,
    call_immediately=True,
    *,
    context: Context | None = None,
    task_factory: TaskFactory | None = None,
) -> AsyncEffect[T]

async_effect[T](
    *,
    call_immediately=True,
    context: Context | None = None,
    task_factory: TaskFactory | None = None,
) -> Callable[[Callable[[], Awaitable[T]]], AsyncEffect[T]]
Source code in reactivity/_curried.py 
111
112
113
114
 
def async_effect[T](fn: Callable[[], Awaitable[T]] = __, /, call_immediately=True, *, context: Context | None = None, task_factory: TaskFactory | None = None):  # type: ignore
    if fn is __:
        return lambda fn: AsyncEffect(fn, call_immediately, context=context, task_factory=task_factory or default_task_factory)
    return AsyncEffect(fn, call_immediately, context=context, task_factory=task_factory or default_task_factory)

batch 

batch(*, context: Context | None = None) -> Batch

batch[**P, T](
    func: Callable[P, T], /, context: Context | None = None
) -> Callable[P, T]
Source code in reactivity/_curried.py 
135
136
137
138
139
140
141
142
143
144
 
def batch[**P, T](func: Callable[P, T] = __, /, context: Context | None = None) -> Callable[P, T] | Batch:
    if func is __:
        return Batch(context=context)

    @wraps(func)
    def wrapped(*args, **kwargs):
        with Batch(context=context):
            return func(*args, **kwargs)

    return wrapped

derived 

derived[T](
    fn: Callable[[], T],
    /,
    check_equality=True,
    *,
    context: Context | None = None,
) -> Derived[T]

derived[T](
    *, check_equality=True, context: Context | None = None
) -> Callable[[Callable[[], T]], Derived[T]]
Source code in reactivity/_curried.py 
39
40
41
42
 
def derived[T](fn: Callable[[], T] = __, /, check_equality=True, *, context: Context | None = None):  # type: ignore
    if fn is __:
        return lambda fn: Derived(fn, check_equality, context=context)
    return Derived(fn, check_equality, context=context)

derived_method 

derived_method[T, I](
    method: Callable[[I], T],
    /,
    check_equality=True,
    *,
    context: Context | None = None,
) -> DerivedMethod[T, I]

derived_method[T, I](
    *, check_equality=True, context: Context | None = None
) -> Callable[[Callable[[I], T]], DerivedMethod[T, I]]
Source code in reactivity/_curried.py 
63
64
65
66
 
def derived_method[T, I](method: Callable[[I], T] = __, /, check_equality=True, *, context: Context | None = None):  # type: ignore
    if method is __:
        return lambda method: DerivedMethod(method, check_equality, context=context)
    return DerivedMethod(method, check_equality, context=context)

derived_property 

derived_property[T, I](
    method: Callable[[I], T],
    /,
    check_equality=True,
    *,
    context: Context | None = None,
) -> DerivedProperty[T, I]

derived_property[T, I](
    *, check_equality=True, context: Context | None = None
) -> Callable[[Callable[[I], T]], DerivedProperty[T, I]]
Source code in reactivity/_curried.py 
51
52
53
54
 
def derived_property[T, I](method: Callable[[I], T] = __, /, check_equality=True, *, context: Context | None = None):  # type: ignore
    if method is __:
        return lambda method: DerivedProperty(method, check_equality, context=context)
    return DerivedProperty(method, check_equality, context=context)

effect 

effect[T](
    fn: Callable[[], T],
    /,
    call_immediately=True,
    *,
    context: Context | None = None,
) -> Effect[T]

effect[T](
    *, call_immediately=True, context: Context | None = None
) -> Callable[[Callable[[], T]], Effect[T]]
Source code in reactivity/_curried.py 
27
28
29
30
 
def effect[T](fn: Callable[[], T] = __, /, call_immediately=True, *, context: Context | None = None):  # type: ignore
    if fn is __:
        return lambda fn: Effect(fn, call_immediately, context=context)
    return Effect(fn, call_immediately, context=context)

memoized 

memoized[T](
    fn: Callable[[], T],
    /,
    *,
    context: Context | None = None,
) -> Memoized[T]

memoized[T](
    *, context: Context | None = None
) -> Callable[[Callable[[], T]], Memoized[T]]
Source code in reactivity/_curried.py 
75
76
77
78
 
def memoized[T](fn: Callable[[], T] = __, /, *, context: Context | None = None):  # type: ignore
    if fn is __:
        return lambda fn: Memoized(fn, context=context)
    return Memoized(fn, context=context)

memoized_method 

memoized_method[T, I](
    method: Callable[[I], T],
    /,
    *,
    context: Context | None = None,
) -> MemoizedMethod[T, I]

memoized_method[T, I](
    *, context: Context | None = None
) -> Callable[[Callable[[I], T]], MemoizedMethod[T, I]]
Source code in reactivity/_curried.py 
 99
100
101
102
 
def memoized_method[T, I](method: Callable[[I], T] = __, /, *, context: Context | None = None):  # type: ignore
    if method is __:
        return lambda method: MemoizedMethod(method, context=context)
    return MemoizedMethod(method, context=context)

memoized_property 

memoized_property[T, I](
    method: Callable[[I], T],
    /,
    *,
    context: Context | None = None,
) -> MemoizedProperty[T, I]

memoized_property[T, I](
    *, context: Context | None = None
) -> Callable[[Callable[[I], T]], MemoizedProperty[T, I]]
Source code in reactivity/_curried.py 
87
88
89
90
 
def memoized_property[T, I](method: Callable[[I], T] = __, /, *, context: Context | None = None):  # type: ignore
    if method is __:
        return lambda method: MemoizedProperty(method, context=context)
    return MemoizedProperty(method, context=context)

signal 

signal[T](
    initial_value: T = None,
    /,
    check_equality=True,
    *,
    context: Context | None = None,
) -> Signal[T]
Source code in reactivity/_curried.py 
10
11
 
def signal[T](initial_value: T = None, /, check_equality=True, *, context: Context | None = None) -> Signal[T]:
    return Signal(initial_value, check_equality, context=context)

state 

state[T](
    initial_value: T = None,
    /,
    check_equality=True,
    *,
    context: Context | None = None,
) -> State[T]
Source code in reactivity/_curried.py 
14
15
 
def state[T](initial_value: T = None, /, check_equality=True, *, context: Context | None = None) -> State[T]:
    return State(initial_value, check_equality, context=context)

reactive 

reactive[K, V](
    value: MutableMapping[K, V],
    check_equality=True,
    *,
    context: Context | None = None,
) -> ReactiveMappingProxy[K, V]

reactive[K, V](
    value: Mapping[K, V],
    check_equality=True,
    *,
    context: Context | None = None,
) -> ReactiveMapping[K, V]

reactive[T](
    value: MutableSet[T],
    check_equality=True,
    *,
    context: Context | None = None,
) -> ReactiveSetProxy[T]

reactive[T](
    value: Set[T],
    check_equality=True,
    *,
    context: Context | None = None,
) -> ReactiveSet[T]

reactive[T](
    value: MutableSequence[T],
    check_equality=True,
    *,
    context: Context | None = None,
) -> ReactiveSequenceProxy[T]

reactive[T](
    value: Sequence[T],
    check_equality=True,
    *,
    context: Context | None = None,
) -> ReactiveSequence[T]

reactive[T](
    value: T,
    check_equality=True,
    *,
    context: Context | None = None,
) -> T
Source code in reactivity/collections.py 
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
 
def reactive(value: Mapping | Set | Sequence | Any, check_equality=True, *, context: Context | None = None):
    match value:
        case MutableMapping():
            return ReactiveMappingProxy(value, check_equality, context=context)
        case Mapping():
            return ReactiveMapping(value, check_equality, context=context)
        case MutableSet():
            return ReactiveSetProxy(value, check_equality, context=context)
        case Set():
            return ReactiveSet(value, check_equality, context=context)
        case MutableSequence():
            return ReactiveSequenceProxy(value, check_equality, context=context)
        case Sequence():
            return ReactiveSequence(value, check_equality, context=context)
        case _:
            return reactive_object_proxy(value, check_equality, context=context)

new_context 

new_context()
Source code in reactivity/context.py 
94
95
 
def new_context():
    return Context([], [], async_execution_context=ContextVar("current context", default=None))