Source code for reactpy.core.hooks

from __future__ import annotations

import asyncio
from collections.abc import Awaitable, Sequence
from logging import getLogger
from types import FunctionType
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Generic,
    NewType,
    Protocol,
    TypeVar,
    cast,
    overload,
)

from typing_extensions import TypeAlias

from reactpy.config import REACTPY_DEBUG_MODE
from reactpy.core._thread_local import ThreadLocal
from reactpy.core.types import ComponentType, Key, State, VdomDict
from reactpy.utils import Ref

if not TYPE_CHECKING:
    # make flake8 think that this variable exists
    ellipsis = type(...)


__all__ = [
    "use_state",
    "use_effect",
    "use_reducer",
    "use_callback",
    "use_ref",
    "use_memo",
]

logger = getLogger(__name__)

_Type = TypeVar("_Type")


@overload
def use_state(initial_value: Callable[[], _Type]) -> State[_Type]:
    ...


@overload
def use_state(initial_value: _Type) -> State[_Type]:
    ...


[docs]def use_state(initial_value: _Type | Callable[[], _Type]) -> State[_Type]: """See the full :ref:`Use State` docs for details Parameters: initial_value: Defines the initial value of the state. A callable (accepting no arguments) can be used as a constructor function to avoid re-creating the initial value on each render. Returns: A tuple containing the current state and a function to update it. """ current_state = _use_const(lambda: _CurrentState(initial_value)) return State(current_state.value, current_state.dispatch)
class _CurrentState(Generic[_Type]): __slots__ = "value", "dispatch" def __init__( self, initial_value: _Type | Callable[[], _Type], ) -> None: if callable(initial_value): self.value = initial_value() else: self.value = initial_value hook = current_hook() def dispatch(new: _Type | Callable[[_Type], _Type]) -> None: if callable(new): next_value = new(self.value) else: next_value = new if not strictly_equal(next_value, self.value): self.value = next_value hook.schedule_render() self.dispatch = dispatch _EffectCleanFunc: TypeAlias = "Callable[[], None]" _SyncEffectFunc: TypeAlias = "Callable[[], _EffectCleanFunc | None]" _AsyncEffectFunc: TypeAlias = "Callable[[], Awaitable[_EffectCleanFunc | None]]" _EffectApplyFunc: TypeAlias = "_SyncEffectFunc | _AsyncEffectFunc" @overload def use_effect( function: None = None, dependencies: Sequence[Any] | ellipsis | None = ..., ) -> Callable[[_EffectApplyFunc], None]: ... @overload def use_effect( function: _EffectApplyFunc, dependencies: Sequence[Any] | ellipsis | None = ..., ) -> None: ...
[docs]def use_effect( function: _EffectApplyFunc | None = None, dependencies: Sequence[Any] | ellipsis | None = ..., ) -> Callable[[_EffectApplyFunc], None] | None: """See the full :ref:`Use Effect` docs for details Parameters: function: Applies the effect and can return a clean-up function dependencies: Dependencies for the effect. The effect will only trigger if the identity of any value in the given sequence changes (i.e. their :func:`id` is different). By default these are inferred based on local variables that are referenced by the given function. Returns: If not function is provided, a decorator. Otherwise ``None``. """ hook = current_hook() dependencies = _try_to_infer_closure_values(function, dependencies) memoize = use_memo(dependencies=dependencies) last_clean_callback: Ref[_EffectCleanFunc | None] = use_ref(None) def add_effect(function: _EffectApplyFunc) -> None: if not asyncio.iscoroutinefunction(function): sync_function = cast(_SyncEffectFunc, function) else: async_function = cast(_AsyncEffectFunc, function) def sync_function() -> _EffectCleanFunc | None: future = asyncio.ensure_future(async_function()) def clean_future() -> None: if not future.cancel(): clean = future.result() if clean is not None: clean() return clean_future def effect() -> None: if last_clean_callback.current is not None: last_clean_callback.current() clean = last_clean_callback.current = sync_function() if clean is not None: hook.add_effect(COMPONENT_WILL_UNMOUNT_EFFECT, clean) return memoize(lambda: hook.add_effect(LAYOUT_DID_RENDER_EFFECT, effect)) if function is not None: add_effect(function) return None else: return add_effect
[docs]def use_debug_value( message: Any | Callable[[], Any], dependencies: Sequence[Any] | ellipsis | None = ..., ) -> None: """Log debug information when the given message changes. .. note:: This hook only logs if :data:`~reactpy.config.REACTPY_DEBUG_MODE` is active. Unlike other hooks, a message is considered to have changed if the old and new values are ``!=``. Because this comparison is performed on every render of the component, it may be worth considering the performance cost in some situations. Parameters: message: The value to log or a memoized function for generating the value. dependencies: Dependencies for the memoized function. The message will only be recomputed if the identity of any value in the given sequence changes (i.e. their :func:`id` is different). By default these are inferred based on local variables that are referenced by the given function. """ old: Ref[Any] = _use_const(lambda: Ref(object())) memo_func = message if callable(message) else lambda: message new = use_memo(memo_func, dependencies) if REACTPY_DEBUG_MODE.current and old.current != new: old.current = new logger.debug(f"{current_hook().component} {new}")
[docs]def create_context(default_value: _Type) -> Context[_Type]: """Return a new context type for use in :func:`use_context`""" def context( *children: Any, value: _Type = default_value, key: Key | None = None, ) -> ContextProvider[_Type]: return ContextProvider( *children, value=value, key=key, type=context, ) context.__qualname__ = "context" return context
[docs]class Context(Protocol[_Type]): """Returns a :class:`ContextProvider` component""" def __call__( self, *children: Any, value: _Type = ..., key: Key | None = ..., ) -> ContextProvider[_Type]: ...
[docs]def use_context(context: Context[_Type]) -> _Type: """Get the current value for the given context type. See the full :ref:`Use Context` docs for more information. """ hook = current_hook() provider = hook.get_context_provider(context) if provider is None: # same assertions but with normal exceptions if not isinstance(context, FunctionType): raise TypeError(f"{context} is not a Context") # nocov if context.__kwdefaults__ is None: raise TypeError(f"{context} has no 'value' kwarg") # nocov if "value" not in context.__kwdefaults__: raise TypeError(f"{context} has no 'value' kwarg") # nocov return cast(_Type, context.__kwdefaults__["value"]) return provider._value
[docs]class ContextProvider(Generic[_Type]): def __init__( self, *children: Any, value: _Type, key: Key | None, type: Context[_Type], ) -> None: self.children = children self.key = key self.type = type self._value = value def render(self) -> VdomDict: current_hook().set_context_provider(self) return {"tagName": "", "children": self.children} def __repr__(self) -> str: return f"{type(self).__name__}({self.type})"
_ActionType = TypeVar("_ActionType")
[docs]def use_reducer( reducer: Callable[[_Type, _ActionType], _Type], initial_value: _Type, ) -> tuple[_Type, Callable[[_ActionType], None]]: """See the full :ref:`Use Reducer` docs for details Parameters: reducer: A function which applies an action to the current state in order to produce the next state. initial_value: The initial state value (same as for :func:`use_state`) Returns: A tuple containing the current state and a function to change it with an action """ state, set_state = use_state(initial_value) return state, _use_const(lambda: _create_dispatcher(reducer, set_state))
def _create_dispatcher( reducer: Callable[[_Type, _ActionType], _Type], set_state: Callable[[Callable[[_Type], _Type]], None], ) -> Callable[[_ActionType], None]: def dispatch(action: _ActionType) -> None: set_state(lambda last_state: reducer(last_state, action)) return dispatch _CallbackFunc = TypeVar("_CallbackFunc", bound=Callable[..., Any]) @overload def use_callback( function: None = None, dependencies: Sequence[Any] | ellipsis | None = ..., ) -> Callable[[_CallbackFunc], _CallbackFunc]: ... @overload def use_callback( function: _CallbackFunc, dependencies: Sequence[Any] | ellipsis | None = ..., ) -> _CallbackFunc: ...
[docs]def use_callback( function: _CallbackFunc | None = None, dependencies: Sequence[Any] | ellipsis | None = ..., ) -> _CallbackFunc | Callable[[_CallbackFunc], _CallbackFunc]: """See the full :ref:`Use Callback` docs for details Parameters: function: The function whose identity will be preserved dependencies: Dependencies of the callback. The identity the ``function`` will be updated if the identity of any value in the given sequence changes (i.e. their :func:`id` is different). By default these are inferred based on local variables that are referenced by the given function. Returns: The current function """ dependencies = _try_to_infer_closure_values(function, dependencies) memoize = use_memo(dependencies=dependencies) def setup(function: _CallbackFunc) -> _CallbackFunc: return memoize(lambda: function) if function is not None: return setup(function) else: return setup
class _LambdaCaller(Protocol): """MyPy doesn't know how to deal with TypeVars only used in function return""" def __call__(self, func: Callable[[], _Type]) -> _Type: ... @overload def use_memo( function: None = None, dependencies: Sequence[Any] | ellipsis | None = ..., ) -> _LambdaCaller: ... @overload def use_memo( function: Callable[[], _Type], dependencies: Sequence[Any] | ellipsis | None = ..., ) -> _Type: ...
[docs]def use_memo( function: Callable[[], _Type] | None = None, dependencies: Sequence[Any] | ellipsis | None = ..., ) -> _Type | Callable[[Callable[[], _Type]], _Type]: """See the full :ref:`Use Memo` docs for details Parameters: function: The function to be memoized. dependencies: Dependencies for the memoized function. The memo will only be recomputed if the identity of any value in the given sequence changes (i.e. their :func:`id` is different). By default these are inferred based on local variables that are referenced by the given function. Returns: The current state """ dependencies = _try_to_infer_closure_values(function, dependencies) memo: _Memo[_Type] = _use_const(_Memo) if memo.empty(): # we need to initialize on the first run changed = True memo.deps = () if dependencies is None else dependencies elif dependencies is None: changed = True memo.deps = () elif ( len(memo.deps) != len(dependencies) # if deps are same length check identity for each item or not all( strictly_equal(current, new) for current, new in zip(memo.deps, dependencies) ) ): memo.deps = dependencies changed = True else: changed = False setup: Callable[[Callable[[], _Type]], _Type] if changed: def setup(function: Callable[[], _Type]) -> _Type: current_value = memo.value = function() return current_value else: def setup(function: Callable[[], _Type]) -> _Type: return memo.value if function is not None: return setup(function) else: return setup
class _Memo(Generic[_Type]): """Simple object for storing memoization data""" __slots__ = "value", "deps" value: _Type deps: Sequence[Any] def empty(self) -> bool: try: self.value # noqa: B018 except AttributeError: return True else: return False
[docs]def use_ref(initial_value: _Type) -> Ref[_Type]: """See the full :ref:`Use State` docs for details Parameters: initial_value: The value initially assigned to the reference. Returns: A :class:`Ref` object. """ return _use_const(lambda: Ref(initial_value))
def _use_const(function: Callable[[], _Type]) -> _Type: return current_hook().use_state(function) def _try_to_infer_closure_values( func: Callable[..., Any] | None, values: Sequence[Any] | ellipsis | None, ) -> Sequence[Any] | None: if values is ...: if isinstance(func, FunctionType): return ( [cell.cell_contents for cell in func.__closure__] if func.__closure__ else [] ) else: return None else: return values
[docs]def current_hook() -> LifeCycleHook: """Get the current :class:`LifeCycleHook`""" hook_stack = _hook_stack.get() if not hook_stack: msg = "No life cycle hook is active. Are you rendering in a layout?" raise RuntimeError(msg) return hook_stack[-1]
_hook_stack: ThreadLocal[list[LifeCycleHook]] = ThreadLocal(list) EffectType = NewType("EffectType", str) """Used in :meth:`LifeCycleHook.add_effect` to indicate what effect should be saved""" COMPONENT_DID_RENDER_EFFECT = EffectType("COMPONENT_DID_RENDER") """An effect that will be triggered each time a component renders""" LAYOUT_DID_RENDER_EFFECT = EffectType("LAYOUT_DID_RENDER") """An effect that will be triggered each time a layout renders""" COMPONENT_WILL_UNMOUNT_EFFECT = EffectType("COMPONENT_WILL_UNMOUNT") """An effect that will be triggered just before the component is unmounted"""
[docs]class LifeCycleHook: """Defines the life cycle of a layout component. Components can request access to their own life cycle events and state through hooks while :class:`~reactpy.core.proto.LayoutType` objects drive drive the life cycle forward by triggering events and rendering view changes. Example: If removed from the complexities of a layout, a very simplified full life cycle for a single component with no child components would look a bit like this: .. testcode:: from reactpy.core.hooks import ( current_hook, LifeCycleHook, COMPONENT_DID_RENDER_EFFECT, ) # this function will come from a layout implementation schedule_render = lambda: ... # --- start life cycle --- hook = LifeCycleHook(schedule_render) # --- start render cycle --- hook.affect_component_will_render(...) hook.set_current() try: # render the component ... # the component may access the current hook assert current_hook() is hook # and save state or add effects current_hook().use_state(lambda: ...) current_hook().add_effect(COMPONENT_DID_RENDER_EFFECT, lambda: ...) finally: hook.unset_current() hook.affect_component_did_render() # This should only be called after the full set of changes associated with a # given render have been completed. hook.affect_layout_did_render() # Typically an event occurs and a new render is scheduled, thus beginning # the render cycle anew. hook.schedule_render() # --- end render cycle --- hook.affect_component_will_unmount() del hook # --- end render cycle --- """ __slots__ = ( "__weakref__", "_context_providers", "_current_state_index", "_event_effects", "_is_rendering", "_rendered_atleast_once", "_schedule_render_callback", "_schedule_render_later", "_state", "component", ) component: ComponentType def __init__( self, schedule_render: Callable[[], None], ) -> None: self._context_providers: dict[Context[Any], ContextProvider[Any]] = {} self._schedule_render_callback = schedule_render self._schedule_render_later = False self._is_rendering = False self._rendered_atleast_once = False self._current_state_index = 0 self._state: tuple[Any, ...] = () self._event_effects: dict[EffectType, list[Callable[[], None]]] = { COMPONENT_DID_RENDER_EFFECT: [], LAYOUT_DID_RENDER_EFFECT: [], COMPONENT_WILL_UNMOUNT_EFFECT: [], } def schedule_render(self) -> None: if self._is_rendering: self._schedule_render_later = True else: self._schedule_render() def use_state(self, function: Callable[[], _Type]) -> _Type: if not self._rendered_atleast_once: # since we're not initialized yet we're just appending state result = function() self._state += (result,) else: # once finalized we iterate over each succesively used piece of state result = self._state[self._current_state_index] self._current_state_index += 1 return result
[docs] def add_effect(self, effect_type: EffectType, function: Callable[[], None]) -> None: """Trigger a function on the occurrence of the given effect type""" self._event_effects[effect_type].append(function)
def set_context_provider(self, provider: ContextProvider[Any]) -> None: self._context_providers[provider.type] = provider def get_context_provider( self, context: Context[_Type] ) -> ContextProvider[_Type] | None: return self._context_providers.get(context)
[docs] def affect_component_will_render(self, component: ComponentType) -> None: """The component is about to render""" self.component = component self._is_rendering = True self._event_effects[COMPONENT_WILL_UNMOUNT_EFFECT].clear()
[docs] def affect_component_did_render(self) -> None: """The component completed a render""" del self.component component_did_render_effects = self._event_effects[COMPONENT_DID_RENDER_EFFECT] for effect in component_did_render_effects: try: effect() except Exception: logger.exception(f"Component post-render effect {effect} failed") component_did_render_effects.clear() self._is_rendering = False self._rendered_atleast_once = True self._current_state_index = 0
[docs] def affect_layout_did_render(self) -> None: """The layout completed a render""" layout_did_render_effects = self._event_effects[LAYOUT_DID_RENDER_EFFECT] for effect in layout_did_render_effects: try: effect() except Exception: logger.exception(f"Layout post-render effect {effect} failed") layout_did_render_effects.clear() if self._schedule_render_later: self._schedule_render() self._schedule_render_later = False
[docs] def affect_component_will_unmount(self) -> None: """The component is about to be removed from the layout""" will_unmount_effects = self._event_effects[COMPONENT_WILL_UNMOUNT_EFFECT] for effect in will_unmount_effects: try: effect() except Exception: logger.exception(f"Pre-unmount effect {effect} failed") will_unmount_effects.clear()
[docs] def set_current(self) -> None: """Set this hook as the active hook in this thread This method is called by a layout before entering the render method of this hook's associated component. """ hook_stack = _hook_stack.get() if hook_stack: parent = hook_stack[-1] self._context_providers.update(parent._context_providers) hook_stack.append(self)
[docs] def unset_current(self) -> None: """Unset this hook as the active hook in this thread""" if _hook_stack.get().pop() is not self: raise RuntimeError("Hook stack is in an invalid state") # nocov
def _schedule_render(self) -> None: try: self._schedule_render_callback() except Exception: logger.exception( f"Failed to schedule render via {self._schedule_render_callback}" )
[docs]def strictly_equal(x: Any, y: Any) -> bool: """Check if two values are identical or, for a limited set or types, equal. Only the following types are checked for equality rather than identity: - ``int`` - ``float`` - ``complex`` - ``str`` - ``bytes`` - ``bytearray`` - ``memoryview`` """ return x is y or (type(x) in _NUMERIC_TEXT_BINARY_TYPES and x == y)
_NUMERIC_TEXT_BINARY_TYPES = { # numeric int, float, complex, # text str, # binary types bytes, bytearray, memoryview, }