from __future__ import annotations
import logging
from asyncio import Event, Task, create_task, gather
from typing import Any, Callable, Protocol, TypeVar
from anyio import Semaphore
from reactpy.core._thread_local import ThreadLocal
from reactpy.core.types import ComponentType, Context, ContextProviderType
T = TypeVar("T")
[docs]class EffectFunc(Protocol):
async def __call__(self, stop: Event) -> None: ...
logger = logging.getLogger(__name__)
_HOOK_STATE: ThreadLocal[list[LifeCycleHook]] = ThreadLocal(list)
[docs]def current_hook() -> LifeCycleHook:
"""Get the current :class:`LifeCycleHook`"""
hook_stack = _HOOK_STATE.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]
[docs]class LifeCycleHook:
"""An object which manages the "life cycle" of a layout component.
The "life cycle" of a component is the set of events which occur from the time
a component is first rendered until it is removed from the layout. The life cycle
is ultimately driven by the layout itself, but components can "hook" into those
events to perform actions. Components gain access to their own life cycle hook
by calling :func:`current_hook`. They can then perform actions such as:
1. Adding state via :meth:`use_state`
2. Adding effects via :meth:`add_effect`
3. Setting or getting context providers via
:meth:`LifeCycleHook.set_context_provider` and
:meth:`get_context_provider` respectively.
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._life_cycle_hook import LifeCycleHook
from reactpy.core.hooks import current_hook
# this function will come from a layout implementation
schedule_render = lambda: ...
# --- start life cycle ---
hook = LifeCycleHook(schedule_render)
# --- start render cycle ---
component = ...
await hook.affect_component_will_render(component)
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: ...)
async def my_effect(stop_event):
...
current_hook().add_effect(my_effect)
finally:
await hook.affect_component_did_render()
# This should only be called after the full set of changes associated with a
# given render have been completed.
await 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",
"_effect_funcs",
"_effect_stops",
"_effect_tasks",
"_render_access",
"_rendered_atleast_once",
"_schedule_render_callback",
"_scheduled_render",
"_state",
"component",
)
component: ComponentType
def __init__(
self,
schedule_render: Callable[[], None],
) -> None:
self._context_providers: dict[Context[Any], ContextProviderType[Any]] = {}
self._schedule_render_callback = schedule_render
self._scheduled_render = False
self._rendered_atleast_once = False
self._current_state_index = 0
self._state: tuple[Any, ...] = ()
self._effect_funcs: list[EffectFunc] = []
self._effect_tasks: list[Task[None]] = []
self._effect_stops: list[Event] = []
self._render_access = Semaphore(1) # ensure only one render at a time
def schedule_render(self) -> None:
if self._scheduled_render:
return None
try:
self._schedule_render_callback()
except Exception:
msg = f"Failed to schedule render via {self._schedule_render_callback}"
logger.exception(msg)
else:
self._scheduled_render = True
[docs] def use_state(self, function: Callable[[], T]) -> T:
"""Add state to this hook
If this hook has not yet rendered, the state is appended to the state tuple.
Otherwise, the state is retrieved from the tuple. This allows state to be
preserved across renders.
"""
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_func: EffectFunc) -> None:
"""Add an effect to this hook
A task to run the effect is created when the component is done rendering.
When the component will be unmounted, the event passed to the effect is
triggered and the task is awaited. The effect should eventually halt after
the event is triggered.
"""
self._effect_funcs.append(effect_func)
[docs] def set_context_provider(self, provider: ContextProviderType[Any]) -> None:
"""Set a context provider for this hook
The context provider will be used to provide state to any child components
of this hook's component which request a context provider of the same type.
"""
self._context_providers[provider.type] = provider
[docs] def get_context_provider(
self, context: Context[T]
) -> ContextProviderType[T] | None:
"""Get a context provider for this hook of the given type
The context provider will have been set by a parent component. If no provider
is found, ``None`` is returned.
"""
return self._context_providers.get(context)
[docs] async def affect_component_will_render(self, component: ComponentType) -> None:
"""The component is about to render"""
await self._render_access.acquire()
self._scheduled_render = False
self.component = component
self.set_current()
[docs] async def affect_component_did_render(self) -> None:
"""The component completed a render"""
self.unset_current()
self._rendered_atleast_once = True
self._current_state_index = 0
self._render_access.release()
del self.component
[docs] async def affect_layout_did_render(self) -> None:
"""The layout completed a render"""
stop = Event()
self._effect_stops.append(stop)
self._effect_tasks.extend(create_task(e(stop)) for e in self._effect_funcs)
self._effect_funcs.clear()
[docs] async def affect_component_will_unmount(self) -> None:
"""The component is about to be removed from the layout"""
for stop in self._effect_stops:
stop.set()
self._effect_stops.clear()
try:
await gather(*self._effect_tasks)
except Exception:
logger.exception("Error in effect")
finally:
self._effect_tasks.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_STATE.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_STATE.get().pop() is not self:
raise RuntimeError("Hook stack is in an invalid state") # nocov