eptm_dashboard/.venv/lib/python3.12/site-packages/reflex_base/components/component.py

2705 lines
90 KiB
Python

"""Base component definitions."""
from __future__ import annotations
import contextlib
import dataclasses
import enum
import functools
import inspect
import operator
import typing
from abc import ABC, ABCMeta, abstractmethod
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from dataclasses import _MISSING_TYPE, MISSING
from functools import wraps
from hashlib import md5
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, ClassVar, TypeVar, cast, get_args, get_origin
from rich.markup import escape
from typing_extensions import dataclass_transform
from reflex_base import constants
from reflex_base.breakpoints import Breakpoints
from reflex_base.components.dynamic import load_dynamic_serializer
from reflex_base.components.field import BaseField, FieldBasedMeta
from reflex_base.components.tags import Tag
from reflex_base.constants import Dirs, EventTriggers, Hooks, Imports, MemoizationMode
from reflex_base.constants.compiler import SpecialAttributes
from reflex_base.constants.state import CAMEL_CASE_MEMO_MARKER
from reflex_base.event import (
EventCallback,
EventChain,
EventHandler,
EventSpec,
args_specs_from_fields,
no_args_event_spec,
parse_args_spec,
pointer_event_spec,
run_script,
unwrap_var_annotation,
)
from reflex_base.style import Style, format_as_emotion
from reflex_base.utils import console, format, imports, types
from reflex_base.utils.imports import ImportDict, ImportVar, ParsedImportDict
from reflex_base.vars import VarData
from reflex_base.vars.base import (
CachedVarOperation,
LiteralNoneVar,
LiteralVar,
Var,
cached_property_no_lock,
)
from reflex_base.vars.function import (
ArgsFunctionOperation,
FunctionStringVar,
FunctionVar,
)
from reflex_base.vars.number import ternary_operation
from reflex_base.vars.object import ObjectVar
from reflex_base.vars.sequence import LiteralArrayVar, LiteralStringVar, StringVar
if TYPE_CHECKING:
import reflex.state
FIELD_TYPE = TypeVar("FIELD_TYPE")
class ComponentField(BaseField[FIELD_TYPE]):
"""A field for a component."""
def __init__(
self,
default: FIELD_TYPE | _MISSING_TYPE = MISSING,
default_factory: Callable[[], FIELD_TYPE] | None = None,
is_javascript: bool | None = None,
annotated_type: type[Any] | _MISSING_TYPE = MISSING,
doc: str | None = None,
) -> None:
"""Initialize the field.
Args:
default: The default value for the field.
default_factory: The default factory for the field.
is_javascript: Whether the field is a javascript property.
annotated_type: The annotated type for the field.
doc: Documentation string for the field.
"""
super().__init__(default, default_factory, annotated_type)
self.doc = doc
self.is_javascript = is_javascript
def __repr__(self) -> str:
"""Represent the field in a readable format.
Returns:
The string representation of the field.
"""
annotated_type_str = (
f", annotated_type={self.annotated_type!r}"
if self.annotated_type is not MISSING
else ""
)
if self.default is not MISSING:
return f"ComponentField(default={self.default!r}, is_javascript={self.is_javascript!r}{annotated_type_str})"
return f"ComponentField(default_factory={self.default_factory!r}, is_javascript={self.is_javascript!r}{annotated_type_str})"
def field(
default: FIELD_TYPE | _MISSING_TYPE = MISSING,
default_factory: Callable[[], FIELD_TYPE] | None = None,
is_javascript_property: bool | None = None,
doc: str | None = None,
) -> FIELD_TYPE:
"""Create a field for a component.
Args:
default: The default value for the field.
default_factory: The default factory for the field.
is_javascript_property: Whether the field is a javascript property.
doc: Documentation string for the field.
Returns:
The field for the component.
Raises:
ValueError: If both default and default_factory are specified.
"""
if default is not MISSING and default_factory is not None:
msg = "cannot specify both default and default_factory"
raise ValueError(msg)
return ComponentField( # pyright: ignore [reportReturnType]
default=default,
default_factory=default_factory,
is_javascript=is_javascript_property,
doc=doc,
)
@dataclass_transform(kw_only_default=True, field_specifiers=(field,))
class BaseComponentMeta(FieldBasedMeta, ABCMeta):
"""Meta class for BaseComponent."""
if TYPE_CHECKING:
_inherited_fields: Mapping[str, ComponentField]
_own_fields: Mapping[str, ComponentField]
_fields: Mapping[str, ComponentField]
_js_fields: Mapping[str, ComponentField]
@classmethod
def _process_annotated_fields(
cls,
namespace: dict[str, Any],
annotations: dict[str, Any],
inherited_fields: dict[str, ComponentField],
) -> dict[str, ComponentField]:
own_fields: dict[str, ComponentField] = {}
for key, annotation in annotations.items():
value = namespace.get(key, MISSING)
if types.is_classvar(annotation):
# If the annotation is a classvar, skip it.
continue
if value is MISSING:
value = ComponentField(
default=None,
is_javascript=(key[0] != "_"),
annotated_type=annotation,
)
elif not isinstance(value, ComponentField):
value = ComponentField(
default=value,
is_javascript=(
(key[0] != "_")
if (existing_field := inherited_fields.get(key)) is None
else existing_field.is_javascript
),
annotated_type=annotation,
)
else:
is_js = value.is_javascript
if is_js is None:
if (existing_field := inherited_fields.get(key)) is not None:
is_js = existing_field.is_javascript
else:
is_js = key[0] != "_"
default = value.default
# If no default or factory provided, default to None
# (same behavior as bare annotations without field())
if default is MISSING and value.default_factory is None:
default = None
value = ComponentField(
default=default,
default_factory=value.default_factory,
is_javascript=is_js,
annotated_type=annotation,
doc=value.doc,
)
own_fields[key] = value
return own_fields
@classmethod
def _create_field(
cls,
annotated_type: Any,
default: Any = MISSING,
default_factory: Callable[[], Any] | None = None,
) -> ComponentField:
return ComponentField(
annotated_type=annotated_type,
default=default,
default_factory=default_factory,
is_javascript=True, # Default for components
)
@classmethod
def _process_field_overrides(
cls,
namespace: dict[str, Any],
annotations: dict[str, Any],
inherited_fields: dict[str, Any],
) -> dict[str, ComponentField]:
own_fields: dict[str, ComponentField] = {}
for key, value, inherited_field in [
(key, value, inherited_field)
for key, value in namespace.items()
if key not in annotations
and ((inherited_field := inherited_fields.get(key)) is not None)
]:
new_field = ComponentField(
default=value,
is_javascript=inherited_field.is_javascript,
annotated_type=inherited_field.annotated_type,
)
own_fields[key] = new_field
return own_fields
@classmethod
def _finalize_fields(
cls,
namespace: dict[str, Any],
inherited_fields: dict[str, ComponentField],
own_fields: dict[str, ComponentField],
) -> None:
# Call parent implementation
super()._finalize_fields(namespace, inherited_fields, own_fields)
# Add JavaScript fields mapping
all_fields = namespace["_fields"]
namespace["_js_fields"] = {
key: value
for key, value in all_fields.items()
if value.is_javascript is True
}
class BaseComponent(metaclass=BaseComponentMeta):
"""The base class for all Reflex components.
This is something that can be rendered as a Component via the Reflex compiler.
"""
children: list[BaseComponent] = field(
doc="The children nested within the component.",
default_factory=list,
is_javascript_property=False,
)
# The library that the component is based on.
library: str | None = field(default=None, is_javascript_property=False)
lib_dependencies: list[str] = field(
doc="List here the non-react dependency needed by `library`",
default_factory=list,
is_javascript_property=False,
)
# The tag to use when rendering the component.
tag: str | None = field(default=None, is_javascript_property=False)
def __init__(
self,
**kwargs,
):
"""Initialize the component.
Args:
**kwargs: The kwargs to pass to the component.
"""
for key, value in kwargs.items():
setattr(self, key, value)
for name, value in self.get_fields().items():
if name not in kwargs:
setattr(self, name, value.default_value())
def set(self, **kwargs):
"""Set the component props.
Args:
**kwargs: The kwargs to set.
Returns:
The component with the updated props.
"""
for key, value in kwargs.items():
setattr(self, key, value)
return self
def __copy__(self) -> BaseComponent:
"""Return a shallow copy suitable for compile-time mutation.
Bypasses ``copy.copy``'s generic ``__reduce_ex__`` dispatch. Nested
mutable containers (``children``, ``style``, ``event_triggers``) are
shared with the original until the caller explicitly rebinds them.
Render-path caches populated on the original are dropped so the clone
recomputes against its (potentially rebound) fields.
Returns:
A new instance of the same class with ``__dict__`` shallow-copied.
"""
new = self.__class__.__new__(self.__class__)
new_dict = vars(new)
new_dict.update(vars(self))
for attr in (
"_cached_render_result",
"_vars_cache",
"_imports_cache",
"_hooks_internal_cache",
"_get_component_prop_property",
):
new_dict.pop(attr, None)
return new
def __eq__(self, value: Any) -> bool:
"""Check if the component is equal to another value.
Args:
value: The value to compare to.
Returns:
Whether the component is equal to the value.
"""
return type(self) is type(value) and bool(
getattr(self, key) == getattr(value, key) for key in self.get_fields()
)
@classmethod
def get_fields(cls) -> Mapping[str, ComponentField]:
"""Get the fields of the component.
Returns:
The fields of the component.
"""
return cls._fields
@classmethod
def get_js_fields(cls) -> Mapping[str, ComponentField]:
"""Get the javascript fields of the component.
Returns:
The javascript fields of the component.
"""
return cls._js_fields
@abstractmethod
def render(self) -> dict:
"""Render the component.
Returns:
The dictionary for template of the component.
"""
@abstractmethod
def _get_all_hooks_internal(self) -> dict[str, VarData | None]:
"""Get the reflex internal hooks for the component and its children.
Returns:
The code that should appear just before user-defined hooks.
"""
@abstractmethod
def _get_all_hooks(self) -> dict[str, VarData | None]:
"""Get the React hooks for this component.
Returns:
The code that should appear just before returning the rendered component.
"""
@abstractmethod
def _get_all_imports(self) -> ParsedImportDict:
"""Get all the libraries and fields that are used by the component.
Returns:
The import dict with the required imports.
"""
@abstractmethod
def _get_all_dynamic_imports(self) -> set[str]:
"""Get dynamic imports for the component.
Returns:
The dynamic imports.
"""
@abstractmethod
def _get_all_custom_code(self) -> dict[str, None]:
"""Get custom code for the component.
Returns:
The custom code.
"""
@abstractmethod
def _get_all_refs(self) -> dict[str, None]:
"""Get the refs for the children of the component.
Returns:
The refs for the children.
"""
class ComponentNamespace(SimpleNamespace):
"""A namespace to manage components with subcomponents."""
def __hash__(self) -> int: # pyright: ignore [reportIncompatibleVariableOverride]
"""Get the hash of the namespace.
Returns:
The hash of the namespace.
"""
return hash(type(self).__name__)
def evaluate_style_namespaces(style: ComponentStyle) -> dict:
"""Evaluate namespaces in the style.
Args:
style: The style to evaluate.
Returns:
The evaluated style.
"""
return {
k.__call__ if isinstance(k, ComponentNamespace) else k: v
for k, v in style.items()
}
# Map from component to styling.
ComponentStyle = dict[str | type[BaseComponent] | Callable | ComponentNamespace, Any]
ComponentChildTypes = (*types.PrimitiveTypes, Var, BaseComponent, type(None))
def _satisfies_type_hint(obj: Any, type_hint: Any) -> bool:
return types._isinstance(
obj,
type_hint,
nested=1,
treat_var_as_type=True,
treat_mutable_obj_as_immutable=(
isinstance(obj, Var) and not isinstance(obj, LiteralVar)
),
)
def satisfies_type_hint(obj: Any, type_hint: Any) -> bool:
"""Check if an object satisfies a type hint.
Args:
obj: The object to check.
type_hint: The type hint to check against.
Returns:
Whether the object satisfies the type hint.
"""
if _satisfies_type_hint(obj, type_hint):
return True
if _satisfies_type_hint(obj, type_hint | None):
obj = (
obj
if not isinstance(obj, Var)
else (obj._var_value if isinstance(obj, LiteralVar) else obj)
)
console.warn(
"Passing None to a Var that is not explicitly marked as Optional (| None) is deprecated. "
f"Passed {obj!s} of type {escape(str(type(obj) if not isinstance(obj, Var) else obj._var_type))} to {escape(str(type_hint))}."
)
return True
return False
def _components_from(
component_or_var: BaseComponent | Var,
) -> tuple[BaseComponent, ...]:
"""Get the components from a component or Var.
Args:
component_or_var: The component or Var to get the components from.
Returns:
The components.
"""
if isinstance(component_or_var, Var):
var_data = component_or_var._get_all_var_data()
return var_data.components if var_data else ()
if isinstance(component_or_var, BaseComponent):
return (component_or_var,)
return ()
def _hash_str(value: str) -> str:
return md5(f'"{value}"'.encode(), usedforsecurity=False).hexdigest()
def _update_deterministic_hash(hasher: Any, value: object) -> None:
"""Feed ``value`` into ``hasher`` using a self-delimiting, type-tagged encoding.
Each branch writes a distinct type tag plus length-prefixed payload, which
keeps the encoding injective without building intermediate strings — the
nested ``str([...])`` approach this replaces was the dominant cost of
``_deterministic_hash`` (~4x speedup on synthetic, ~2x on real renders).
Args:
hasher: A ``hashlib`` hasher (must accept ``.update(bytes)``).
value: The value to fold into the hasher.
Raises:
TypeError: If the value is not hashable.
"""
if value is None:
hasher.update(b"N")
elif isinstance(value, bool):
hasher.update(b"T" if value else b"F")
elif isinstance(value, (int, float, enum.Enum)):
hasher.update(b"n")
hasher.update(str(value).encode())
elif isinstance(value, str):
encoded = value.encode()
hasher.update(b"s")
hasher.update(len(encoded).to_bytes(8, "little"))
hasher.update(encoded)
elif isinstance(value, dict):
items = sorted(value.items(), key=operator.itemgetter(0))
hasher.update(b"d")
hasher.update(len(items).to_bytes(8, "little"))
for k, v in items:
_update_deterministic_hash(hasher, k)
_update_deterministic_hash(hasher, v)
elif isinstance(value, (tuple, list)):
hasher.update(b"l")
hasher.update(len(value).to_bytes(8, "little"))
for item in value:
_update_deterministic_hash(hasher, item)
elif isinstance(value, Var):
hasher.update(b"v")
_update_deterministic_hash(hasher, value._js_expr)
_update_deterministic_hash(hasher, value._get_all_var_data())
elif dataclasses.is_dataclass(value):
fields = dataclasses.fields(value)
hasher.update(b"D")
hasher.update(len(fields).to_bytes(8, "little"))
for field in fields:
hasher.update(field.name.encode())
_update_deterministic_hash(hasher, getattr(value, field.name))
elif isinstance(value, BaseComponent):
hasher.update(b"C")
_update_deterministic_hash(hasher, value.render())
else:
msg = (
f"Cannot hash value `{value}` of type `{type(value).__name__}`. "
"Only BaseComponent, Var, VarData, dict, str, tuple, and enum.Enum are supported."
)
raise TypeError(msg)
def _deterministic_hash(value: object) -> str:
"""Hash a rendered dictionary.
Args:
value: The dictionary to hash.
Returns:
The hash of the dictionary.
Raises:
TypeError: If the value is not hashable.
"""
hasher = md5(usedforsecurity=False)
_update_deterministic_hash(hasher, value)
return hasher.hexdigest()
@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
class TriggerDefinition:
"""A default event trigger with its args spec and description."""
spec: types.ArgsSpec | Sequence[types.ArgsSpec]
description: str
DEFAULT_TRIGGERS_AND_DESC: Mapping[str, TriggerDefinition] = {
EventTriggers.ON_FOCUS: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the element (or some element inside of it) receives focus. For example, it is called when the user clicks on a text input.",
),
EventTriggers.ON_BLUR: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when focus has left the element (or left some element inside of it). For example, it is called when the user clicks outside of a focused text input.",
),
EventTriggers.ON_CLICK: TriggerDefinition(
spec=pointer_event_spec, # pyright: ignore [reportArgumentType]
description="Fired when the user clicks on an element. For example, it's called when the user clicks on a button.",
),
EventTriggers.ON_CONTEXT_MENU: TriggerDefinition(
spec=pointer_event_spec, # pyright: ignore [reportArgumentType]
description="Fired when the user right-clicks on an element.",
),
EventTriggers.ON_DOUBLE_CLICK: TriggerDefinition(
spec=pointer_event_spec, # pyright: ignore [reportArgumentType]
description="Fired when the user double-clicks on an element.",
),
EventTriggers.ON_MOUSE_DOWN: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the user presses a mouse button on an element.",
),
EventTriggers.ON_MOUSE_ENTER: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the mouse pointer enters the element.",
),
EventTriggers.ON_MOUSE_LEAVE: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the mouse pointer leaves the element.",
),
EventTriggers.ON_MOUSE_MOVE: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the mouse pointer moves over the element.",
),
EventTriggers.ON_MOUSE_OUT: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the mouse pointer moves out of the element.",
),
EventTriggers.ON_MOUSE_OVER: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the mouse pointer moves onto the element.",
),
EventTriggers.ON_MOUSE_UP: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the user releases a mouse button on an element.",
),
EventTriggers.ON_SCROLL: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the user scrolls the element.",
),
EventTriggers.ON_SCROLL_END: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when scrolling ends on the element.",
),
EventTriggers.ON_MOUNT: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the component is mounted to the page.",
),
EventTriggers.ON_UNMOUNT: TriggerDefinition(
spec=no_args_event_spec,
description="Fired when the component is removed from the page. Only called during navigation, not on page refresh.",
),
}
DEFAULT_TRIGGERS = {
name: trigger.spec for name, trigger in DEFAULT_TRIGGERS_AND_DESC.items()
}
T = TypeVar("T", bound="Component")
class Component(BaseComponent, ABC):
"""A component with style, event trigger and other props."""
style: Style = field(
doc="The style of the component.",
default_factory=Style,
is_javascript_property=False,
)
event_triggers: dict[str, EventChain | Var] = field(
doc="A mapping from event triggers to event chains.",
default_factory=dict,
is_javascript_property=False,
)
# The alias for the tag.
alias: str | None = field(default=None, is_javascript_property=False)
# Whether the component is a global scope tag. True for tags like `html`, `head`, `body`.
_is_tag_in_global_scope: ClassVar[bool] = False
# Whether the import is default or named.
is_default: bool | None = field(default=False, is_javascript_property=False)
key: Any = field(
doc="A unique key for the component.",
default=None,
is_javascript_property=False,
)
id: Any = field(
doc="The id for the component.", default=None, is_javascript_property=False
)
ref: Var | None = field(
doc="The Var to pass as the ref to the component.",
default=None,
is_javascript_property=False,
)
class_name: Any = field(
doc="The class name for the component.",
default=None,
is_javascript_property=False,
)
special_props: list[Var] = field(
doc="Special component props.",
default_factory=list,
is_javascript_property=False,
)
# components that cannot be children
_invalid_children: ClassVar[list[str]] = []
# only components that are allowed as children
_valid_children: ClassVar[list[str]] = []
# only components that are allowed as parent
_valid_parents: ClassVar[list[str]] = []
# props to change the name of
_rename_props: ClassVar[dict[str, str]] = {}
custom_attrs: dict[str, Var | Any] = field(
doc="Attributes passed directly to the component.",
default_factory=dict,
is_javascript_property=False,
)
_memoization_mode: MemoizationMode = field(
doc="When to memoize this component and its children.",
default_factory=MemoizationMode,
is_javascript_property=False,
)
# State class associated with this component instance
State: type[reflex.state.State] | None = field(
default=None, is_javascript_property=False
)
def add_imports(self) -> ImportDict | list[ImportDict]:
"""Add imports for the component.
This method should be implemented by subclasses to add new imports for the component.
Implementations do NOT need to call super(). The result of calling
add_imports in each parent class will be merged internally.
Returns:
The additional imports for this component subclass.
The format of the return value is a dictionary where the keys are the
library names (with optional npm-style version specifications) mapping
to a single name to be imported, or a list names to be imported.
For advanced use cases, the values can be ImportVar instances (for
example, to provide an alias or mark that an import is the default
export from the given library).
```python
return {
"react": "useEffect",
"react-draggable": ["DraggableCore", rx.ImportVar(tag="Draggable", is_default=True)],
}
```
"""
return {}
def add_hooks(self) -> list[str | Var]:
"""Add hooks inside the component function.
Hooks are pieces of literal Javascript code that is inserted inside the
React component function.
Each logical hook should be a separate string in the list.
Common strings will be deduplicated and inserted into the component
function only once, so define const variables and other identical code
in their own strings to avoid defining the same const or hook multiple
times.
If a hook depends on specific data from the component instance, be sure
to use unique values inside the string to _avoid_ deduplication.
Implementations do NOT need to call super(). The result of calling
add_hooks in each parent class will be merged and deduplicated internally.
Returns:
The additional hooks for this component subclass.
```python
return [
"const [count, setCount] = useState(0);",
"useEffect(() => { setCount((prev) => prev + 1); console.log(`mounted ${count} times`); }, []);",
]
```
"""
return []
def add_custom_code(self) -> list[str]:
"""Add custom Javascript code into the page that contains this component.
Custom code is inserted at module level, after any imports.
Each string of custom code is deduplicated per-page, so take care to
avoid defining the same const or function differently from different
component instances.
Custom code is useful for defining global functions or constants which
can then be referenced inside hooks or used by component vars.
Implementations do NOT need to call super(). The result of calling
add_custom_code in each parent class will be merged and deduplicated internally.
Returns:
The additional custom code for this component subclass.
```python
return [
"const translatePoints = (event) => { return { x: event.clientX, y: event.clientY }; };",
]
```
"""
return []
@classmethod
def __init_subclass__(cls, **kwargs):
"""Set default properties.
Args:
**kwargs: The kwargs to pass to the superclass.
"""
super().__init_subclass__(**kwargs)
# Ensure renamed props from parent classes are applied to the subclass.
if cls._rename_props:
inherited_rename_props = {}
for parent in reversed(cls.mro()):
if issubclass(parent, Component) and parent._rename_props:
inherited_rename_props.update(parent._rename_props)
cls._rename_props = inherited_rename_props
def __init__(self, **kwargs):
"""Initialize the custom component.
Args:
**kwargs: The kwargs to pass to the component.
"""
console.error(
"Instantiating components directly is not supported."
f" Use `{self.__class__.__name__}.create` method instead."
)
def _post_init(self, *args, **kwargs):
"""Initialize the component.
Args:
*args: Args to initialize the component.
**kwargs: Kwargs to initialize the component.
Raises:
TypeError: If an invalid prop is passed.
ValueError: If an event trigger passed is not valid.
"""
# Set the id and children initially.
children = kwargs.get("children", [])
self._validate_component_children(children)
# Get the component fields, triggers, and props.
fields = self.get_fields()
component_specific_triggers = self.get_event_triggers()
props = self.get_props()
# Add any events triggers.
if "event_triggers" not in kwargs:
kwargs["event_triggers"] = {}
kwargs["event_triggers"] = kwargs["event_triggers"].copy()
# Iterate through the kwargs and set the props.
for key, value in kwargs.items():
if (
key.startswith("on_")
and key not in component_specific_triggers
and key not in props
):
valid_triggers = sorted(component_specific_triggers.keys())
msg = (
f"The {(comp_name := type(self).__name__)} does not take in an `{key}` event trigger. "
f"Valid triggers for {comp_name}: {valid_triggers}. "
f"If {comp_name} is a third party component make sure to add `{key}` to the component's event triggers. "
f"visit https://reflex.dev/docs/wrapping-react/guide/#event-triggers for more info."
)
raise ValueError(msg)
if key in component_specific_triggers:
# Event triggers are bound to event chains.
is_var = False
elif key in props:
# Set the field type.
is_var = (
field.type_origin is Var if (field := fields.get(key)) else False
)
else:
continue
# Check whether the key is a component prop.
if is_var:
try:
kwargs[key] = LiteralVar.create(value)
# Get the passed type and the var type.
passed_type = kwargs[key]._var_type
expected_type = typing.get_args(
types.get_field_type(type(self), key)
)[0]
except TypeError:
# If it is not a valid var, check the base types.
passed_type = type(value)
expected_type = types.get_field_type(type(self), key)
if not satisfies_type_hint(value, expected_type):
value_name = value._js_expr if isinstance(value, Var) else value
additional_info = (
" You can call `.bool()` on the value to convert it to a boolean."
if expected_type is bool and isinstance(value, Var)
else ""
)
raise TypeError(
f"Invalid var passed for prop {type(self).__name__}.{key}, expected type {expected_type}, got value {value_name} of type {passed_type}."
+ additional_info
)
# Check if the key is an event trigger.
if key in component_specific_triggers:
kwargs["event_triggers"][key] = EventChain.create(
value=value,
args_spec=component_specific_triggers[key],
key=key,
)
# Remove any keys that were added as events.
for key in kwargs["event_triggers"]:
kwargs.pop(key, None)
# Place data_ and aria_ attributes into custom_attrs
special_attributes = [
key
for key in kwargs
if key not in fields and SpecialAttributes.is_special(key)
]
if special_attributes:
custom_attrs = kwargs.setdefault("custom_attrs", {})
custom_attrs.update({
format.to_kebab_case(key): kwargs.pop(key) for key in special_attributes
})
# Add style props to the component.
style = kwargs.get("style", {})
if isinstance(style, Sequence):
if any(not isinstance(s, Mapping) for s in style):
msg = "Style must be a dictionary or a list of dictionaries."
raise TypeError(msg)
# Merge styles, the later ones overriding keys in the earlier ones.
style = {
k: v
for style_dict in style
for k, v in cast(Mapping, style_dict).items()
}
if isinstance(style, (Breakpoints, Var)):
style = {
# Assign the Breakpoints to the self-referential selector to avoid squashing down to a regular dict.
"&": style,
}
fields_style = self.get_fields()["style"]
kwargs["style"] = Style({
**fields_style.default_value(),
**style,
**{attr: value for attr, value in kwargs.items() if attr not in fields},
})
# Convert class_name to str if it's list
class_name = kwargs.get("class_name", "")
if isinstance(class_name, (list, tuple)):
has_var = False
for c in class_name:
if isinstance(c, str):
continue
if isinstance(c, Var):
if not isinstance(c, StringVar) and not issubclass(
c._var_type, str
):
msg = f"Invalid class_name passed for prop {type(self).__name__}.class_name, expected type str, got value {c._js_expr} of type {c._var_type}."
raise TypeError(msg)
has_var = True
else:
msg = f"Invalid class_name passed for prop {type(self).__name__}.class_name, expected type str, got value {c} of type {type(c)}."
raise TypeError(msg)
if has_var:
kwargs["class_name"] = LiteralArrayVar.create(
class_name, _var_type=list[str]
).join(" ")
else:
kwargs["class_name"] = " ".join(class_name)
elif (
isinstance(class_name, Var)
and not isinstance(class_name, StringVar)
and not issubclass(class_name._var_type, str)
):
msg = f"Invalid class_name passed for prop {type(self).__name__}.class_name, expected type str, got value {class_name._js_expr} of type {class_name._var_type}."
raise TypeError(msg)
# Construct the component.
for key, value in kwargs.items():
setattr(self, key, value)
@classmethod
def get_event_triggers(cls) -> dict[str, types.ArgsSpec | Sequence[types.ArgsSpec]]:
"""Get the event triggers for the component.
Returns:
The event triggers.
"""
# Look for component specific triggers,
# e.g. variable declared as EventHandler types.
return DEFAULT_TRIGGERS | args_specs_from_fields(cls.get_fields()) # pyright: ignore [reportOperatorIssue]
def __repr__(self) -> str:
"""Represent the component in React.
Returns:
The code to render the component.
"""
return format.json_dumps(self.render())
def __str__(self) -> str:
"""Represent the component in React.
Returns:
The code to render the component.
"""
from reflex.compiler.compiler import _compile_component
return _compile_component(self)
def _exclude_props(self) -> list[str]:
"""Props to exclude when adding the component props to the Tag.
Returns:
A list of component props to exclude.
"""
return []
def _render(self, props: dict[str, Any] | None = None) -> Tag:
"""Define how to render the component in React.
Args:
props: The props to render (if None, then use get_props).
Returns:
The tag to render.
"""
# Create the base tag.
name = (self.tag if not self.alias else self.alias) or ""
if self._is_tag_in_global_scope and self.library is None:
name = '"' + name + '"'
# Create the base tag.
tag = Tag(
name=name,
special_props=self.special_props.copy(),
)
if props is None:
# Add component props to the tag.
props = {
attr.removesuffix("_"): getattr(self, attr) for attr in self.get_props()
}
# Add ref to element if `ref` is None and `id` is not None.
if self.ref is not None:
props["ref"] = self.ref
elif (ref := self.get_ref()) is not None:
props["ref"] = Var(_js_expr=ref)
else:
props = props.copy()
props.update(
**{
trigger: handler
for trigger, handler in self.event_triggers.items()
if trigger not in {EventTriggers.ON_MOUNT, EventTriggers.ON_UNMOUNT}
},
key=self.key,
id=self.id,
class_name=self.class_name,
)
props.update(self._get_style())
props.update(self.custom_attrs)
# remove excluded props from prop dict before adding to tag.
for prop_to_exclude in self._exclude_props():
props.pop(prop_to_exclude, None)
return tag.add_props(**props)
@classmethod
@functools.cache
def get_props(cls) -> Iterable[str]:
"""Get the unique fields for the component.
Returns:
The unique fields.
"""
return cls.get_js_fields()
@classmethod
@functools.cache
def get_initial_props(cls) -> set[str]:
"""Get the initial props to set for the component.
Returns:
The initial props to set.
"""
return set()
@functools.cached_property
def _get_component_prop_property(self) -> Sequence[BaseComponent]:
return [
component
for prop in self.get_props()
if (value := getattr(self, prop)) is not None
and isinstance(value, (BaseComponent, Var))
for component in _components_from(value)
]
def _get_components_in_props(self) -> Sequence[BaseComponent]:
"""Get the components in the props.
Returns:
The components in the props
"""
return self._get_component_prop_property
@classmethod
def _validate_children(cls, children: tuple | list):
from reflex_base.utils.exceptions import ChildrenTypeError
for child in children:
if isinstance(child, (tuple, list)):
cls._validate_children(child)
# Make sure the child is a valid type.
if isinstance(child, dict) or not isinstance(child, ComponentChildTypes):
raise ChildrenTypeError(component=cls.__name__, child=child)
@classmethod
def create(cls: type[T], *children, **props) -> T:
"""Create the component.
Args:
*children: The children of the component.
**props: The props of the component.
Returns:
The component.
"""
# Import here to avoid circular imports.
from reflex_components_core.base.bare import Bare
from reflex_components_core.base.fragment import Fragment
# Filter out None props
props = {key: value for key, value in props.items() if value is not None}
# Validate all the children.
cls._validate_children(children)
children_normalized = [
(
child
if isinstance(child, Component)
else (
Fragment.create(*child)
if isinstance(child, tuple)
else Bare.create(contents=LiteralVar.create(child))
)
)
for child in children
]
return cls._create(children_normalized, **props)
@classmethod
def _create(cls: type[T], children: Sequence[BaseComponent], **props: Any) -> T:
"""Create the component.
Args:
children: The children of the component.
**props: The props of the component.
Returns:
The component.
"""
comp = cls.__new__(cls)
super(Component, comp).__init__(id=props.get("id"), children=list(children))
comp._post_init(children=list(children), **props)
return comp
@classmethod
def _unsafe_create(
cls: type[T], children: Sequence[BaseComponent], **props: Any
) -> T:
"""Create the component without running post_init.
Args:
children: The children of the component.
**props: The props of the component.
Returns:
The component.
"""
comp = cls.__new__(cls)
super(Component, comp).__init__(id=props.get("id"), children=list(children))
for prop, value in props.items():
setattr(comp, prop, value)
return comp
def add_style(self) -> dict[str, Any] | None:
"""Add style to the component.
Downstream components can override this method to return a style dict
that will be applied to the component.
Returns:
The style to add.
"""
return None
def _add_style(self) -> Style:
"""Call add_style for all bases in the MRO.
Downstream components should NOT override. Use add_style instead.
Returns:
The style to add.
"""
styles = []
# Walk the MRO to call all `add_style` methods.
for base in self._iter_parent_classes_with_method("add_style"):
s = base.add_style(self)
if s is not None:
styles.append(s)
style_ = Style()
for s in reversed(styles):
style_.update(s)
return style_
def _get_component_style(self, styles: ComponentStyle | Style) -> Style | None:
"""Get the style to the component from `App.style`.
Args:
styles: The style to apply.
Returns:
The style of the component.
"""
component_style = None
if (style := styles.get(type(self))) is not None: # pyright: ignore [reportArgumentType]
component_style = Style(style)
if (style := styles.get(self.create)) is not None: # pyright: ignore [reportArgumentType]
component_style = Style(style)
return component_style
def _add_style_recursive(
self, style: ComponentStyle | Style, theme: Component | None = None
) -> Component:
"""Add additional style to the component and its children.
Apply order is as follows (with the latest overriding the earliest):
1. Default style from `_add_style`/`add_style`.
2. User-defined style from `App.style`.
3. User-defined style from `Component.style`.
4. style dict and css props passed to the component instance.
Args:
style: A dict from component to styling.
theme: The theme to apply. (for retro-compatibility with deprecated _apply_theme API)
Returns:
The component with the additional style.
Raises:
UserWarning: If `_add_style` has been overridden.
"""
# 1. Default style from `_add_style`/`add_style`.
if type(self)._add_style != Component._add_style:
msg = "Do not override _add_style directly. Use add_style instead."
raise UserWarning(msg)
new_style = self._add_style()
style_vars = [new_style._var_data]
# 2. User-defined style from `App.style`.
component_style = self._get_component_style(style)
if component_style:
new_style.update(component_style)
style_vars.append(component_style._var_data)
# 4. style dict and css props passed to the component instance.
new_style.update(self.style)
style_vars.append(self.style._var_data)
new_style._var_data = VarData.merge(*style_vars)
# Assign the new style
self.style = new_style
# Recursively add style to the children.
for child in self.children:
# Skip non-Component children.
if not isinstance(child, Component):
continue
child._add_style_recursive(style, theme)
return self
def _get_style(self) -> dict:
"""Get the style for the component.
Returns:
The dictionary of the component style as value and the style notation as key.
"""
if isinstance(self.style, Var):
return {"css": self.style}
emotion_style = format_as_emotion(self.style)
return (
{"css": LiteralVar.create(emotion_style)}
if emotion_style is not None
else {}
)
def render(self) -> dict:
"""Render the component.
Returns:
The dictionary for template of component.
"""
try:
return self._cached_render_result
except AttributeError:
pass
tag = self._render()
rendered_dict = dict(
tag.set(
children=[child.render() for child in self.children],
)
)
self._replace_prop_names(rendered_dict)
self._cached_render_result = rendered_dict
return rendered_dict
def _get_component_hash(self, shallow: bool = False) -> str:
"""Get a stable content hash for this component.
The hash incorporates the rendered JSX dict plus the component's
recursive imports, hooks (including internal lifecycle hooks),
custom code, and app-wrap components, so two components that
compile to semantically distinct JS modules hash differently
even when their ``render()`` output happens to match (e.g. two
components differing only in ``on_mount``, which is excluded
from ``_render`` props but lives in the lifecycle hook).
Args:
shallow: If True, only hash the component's own render output and
directly defined hooks, imports, custom code, and app-wrap
components, excluding any of those from child components.
Returns:
The hex digest content hash.
"""
hasher = md5(usedforsecurity=False)
_update_deterministic_hash(hasher, self.render())
if shallow:
# For non-snapshot strategies, we only hash the component's own hooks, imports, custom code, and app-wrap components
_update_deterministic_hash(hasher, dict(self._get_imports()))
_update_deterministic_hash(hasher, dict(self._get_hooks_internal()))
_update_deterministic_hash(hasher, dict(self._get_added_hooks()))
_update_deterministic_hash(hasher, self._get_hooks())
_update_deterministic_hash(hasher, self._get_custom_code())
_update_deterministic_hash(hasher, dict(self._get_app_wrap_components()))
else:
_update_deterministic_hash(hasher, dict(self._get_all_imports()))
_update_deterministic_hash(hasher, dict(self._get_all_hooks_internal()))
_update_deterministic_hash(hasher, dict(self._get_all_hooks()))
_update_deterministic_hash(hasher, dict(self._get_all_custom_code()))
_update_deterministic_hash(
hasher, dict(self._get_all_app_wrap_components())
)
return hasher.hexdigest()
def _compute_memo_tag(self) -> str:
"""Compute a stable tag name for memoizing this component.
The class qualname is encoded directly in the tag prefix so that
distinct classes which happen to render identically never collide
on a tag. Tag collision would silently share a single cached memo
wrapper across classes and drop the later class's class-level
metadata (e.g. ``_get_app_wrap_components``, which carries
providers like ``UploadFilesProvider`` that must reach the app
root).
Returns:
The stable tag name.
"""
from reflex_base.components.memoize_helpers import (
MemoizationStrategy,
get_memoization_strategy,
)
comp_hash = self._get_component_hash(
shallow=get_memoization_strategy(self) == MemoizationStrategy.PASSTHROUGH
)
return format.format_state_name(
f"{type(self).__qualname__}_{self.tag or 'Comp'}_{comp_hash}"
).capitalize()
def _replace_prop_names(self, rendered_dict: dict) -> None:
"""Replace the prop names in the render dictionary.
Args:
rendered_dict: The render dictionary with all the component props and event handlers.
"""
# fast path
if not self._rename_props:
return
for ix, prop in enumerate(rendered_dict["props"]):
for old_prop, new_prop in self._rename_props.items():
if prop.startswith(old_prop):
rendered_dict["props"][ix] = prop.replace(old_prop, new_prop, 1)
def _validate_component_children(self, children: list[Component]):
"""Validate the children components.
Args:
children: The children of the component.
"""
from reflex_components_core.base.fragment import Fragment
from reflex_components_core.core.cond import Cond
from reflex_components_core.core.foreach import Foreach
from reflex_components_core.core.match import Match
no_valid_parents_defined = all(child._valid_parents == [] for child in children)
if (
not self._invalid_children
and not self._valid_children
and no_valid_parents_defined
):
return
comp_name = type(self).__name__
allowed_components = [
comp.__name__ for comp in (Fragment, Foreach, Cond, Match)
]
def validate_child(child: Any):
child_name = type(child).__name__
# Iterate through the immediate children of fragment
if isinstance(child, Fragment):
for c in child.children:
validate_child(c)
if isinstance(child, Cond):
validate_child(child.children[0])
validate_child(child.children[1])
if isinstance(child, Match):
for cases in child.match_cases:
validate_child(cases[-1])
validate_child(child.default)
if self._invalid_children and child_name in self._invalid_children:
msg = f"The component `{comp_name}` cannot have `{child_name}` as a child component"
raise ValueError(msg)
if self._valid_children and child_name not in [
*self._valid_children,
*allowed_components,
]:
valid_child_list = ", ".join([
f"`{v_child}`" for v_child in self._valid_children
])
msg = f"The component `{comp_name}` only allows the components: {valid_child_list} as children. Got `{child_name}` instead."
raise ValueError(msg)
if child._valid_parents and all(
clz_name not in [*child._valid_parents, *allowed_components]
for clz_name in self._iter_parent_classes_names()
):
valid_parent_list = ", ".join([
f"`{v_parent}`" for v_parent in child._valid_parents
])
msg = f"The component `{child_name}` can only be a child of the components: {valid_parent_list}. Got `{comp_name}` instead."
raise ValueError(msg)
for child in children:
validate_child(child)
@staticmethod
def _get_vars_from_event_triggers(
event_triggers: dict[str, EventChain | Var],
) -> Iterator[tuple[str, list[Var]]]:
"""Get the Vars associated with each event trigger.
Args:
event_triggers: The event triggers from the component instance.
Yields:
tuple of (event_name, event_vars)
"""
for event_trigger, event in event_triggers.items():
if isinstance(event, Var):
yield event_trigger, [event]
elif isinstance(event, EventChain):
event_args = []
for spec in event.events:
if isinstance(spec, EventSpec):
for args in spec.args:
event_args.extend(args)
else:
event_args.append(spec)
yield event_trigger, event_args
def _get_vars(
self, include_children: bool = False, ignore_ids: set[int] | None = None
) -> Iterator[Var]:
"""Walk all Vars used in this component.
Args:
include_children: Whether to include Vars from children.
ignore_ids: The ids to ignore.
Yields:
Each var referenced by the component (props, styles, event handlers).
"""
if not include_children and ignore_ids is None:
cached = self.__dict__.get("_vars_cache")
if cached is not None:
yield from cached
return
ignore_ids = ignore_ids or set()
vars: list[Var] = []
# Get Vars associated with event trigger arguments.
for _, event_vars in self._get_vars_from_event_triggers(self.event_triggers):
vars.extend(event_vars)
# Get Vars associated with component props.
for prop in self.get_props():
prop_var = getattr(self, prop)
if isinstance(prop_var, Var):
vars.append(prop_var)
# Style keeps track of its own VarData instance, so embed in a temp Var that is yielded.
if (isinstance(self.style, dict) and self.style) or isinstance(self.style, Var):
vars.append(
Var(
_js_expr="style",
_var_type=str,
_var_data=VarData.merge(self.style._var_data),
)
)
# Special props are always Var instances.
vars.extend(self.special_props)
# Get Vars associated with common Component props.
for comp_prop in (
self.class_name,
self.id,
self.key,
*self.custom_attrs.values(),
):
if isinstance(comp_prop, Var):
vars.append(comp_prop)
elif isinstance(comp_prop, str):
# Collapse VarData encoded in f-strings.
var = LiteralStringVar.create(comp_prop)
if var._get_all_var_data() is not None:
vars.append(var)
if include_children:
yield from vars
for child in self.children:
if not isinstance(child, Component) or id(child) in ignore_ids:
continue
ignore_ids.add(id(child))
yield from child._get_vars(
include_children=include_children, ignore_ids=ignore_ids
)
return
# Freeze and cache the default-args result.
self._vars_cache = tuple(vars)
yield from vars
def _event_trigger_values_use_state(self) -> bool:
"""Check if the values of a component's event trigger use state.
Returns:
True if any of the component's event trigger values uses State.
"""
for trigger in self.event_triggers.values():
if isinstance(trigger, EventChain):
for event in trigger.events:
if isinstance(event, EventCallback):
continue
if isinstance(event, EventSpec):
if event.handler.state is not None:
return True
else:
if event._var_state:
return True
elif isinstance(trigger, Var) and trigger._var_state:
return True
return False
def _has_stateful_event_triggers(self):
"""Check if component or children have any event triggers that use state.
Returns:
True if the component or children have any event triggers that uses state.
"""
if self.event_triggers and self._event_trigger_values_use_state():
return True
for child in self.children:
if isinstance(child, Component) and child._has_stateful_event_triggers():
return True
return False
@classmethod
def _iter_parent_classes_names(cls) -> Iterator[str]:
for clz in cls.mro():
if clz is Component:
break
yield clz.__name__
@classmethod
@functools.cache
def _iter_parent_classes_with_method(cls, method: str) -> Sequence[type[Component]]:
"""Iterate through parent classes that define a given method.
Used for handling the `add_*` API functions that internally simulate a super() call chain.
Args:
method: The method to look for.
Returns:
A sequence of parent classes that define the method (differently than the base).
"""
current_class_method = getattr(Component, method, None)
seen_methods = (
{current_class_method} if current_class_method is not None else set()
)
clzs: list[type[Component]] = []
for clz in cls.mro():
if clz is Component:
break
if not issubclass(clz, Component):
continue
method_func = getattr(clz, method, None)
if not callable(method_func) or method_func in seen_methods:
continue
seen_methods.add(method_func)
clzs.append(clz)
return tuple(clzs)
def _get_custom_code(self) -> str | None:
"""Get custom code for the component.
Returns:
The custom code.
"""
return None
def _get_all_custom_code(self) -> dict[str, None]:
"""Get custom code for the component and its children.
Returns:
The custom code.
"""
# Store the code in a set to avoid duplicates.
code: dict[str, None] = {}
# Add the custom code for this component.
custom_code = self._get_custom_code()
if custom_code is not None:
code[custom_code] = None
for component in self._get_components_in_props():
code |= component._get_all_custom_code()
# Add the custom code from add_custom_code method.
for clz in self._iter_parent_classes_with_method("add_custom_code"):
for item in clz.add_custom_code(self):
code[item] = None
# Add the custom code for the children.
for child in self.children:
code |= child._get_all_custom_code()
# Return the code.
return code
def _get_dynamic_imports(self) -> str | None:
"""Get dynamic import for the component.
Returns:
The dynamic import.
"""
return None
def _get_all_dynamic_imports(self) -> set[str]:
"""Get dynamic imports for the component and its children.
Returns:
The dynamic imports.
"""
# Store the import in a set to avoid duplicates.
dynamic_imports: set[str] = set()
# Get dynamic import for this component.
dynamic_import = self._get_dynamic_imports()
if dynamic_import:
dynamic_imports.add(dynamic_import)
# Get the dynamic imports from children
for child in self.children:
dynamic_imports |= child._get_all_dynamic_imports()
for component in self._get_components_in_props():
dynamic_imports |= component._get_all_dynamic_imports()
# Return the dynamic imports
return dynamic_imports
def _get_dependencies_imports(self) -> ParsedImportDict:
"""Get the imports from lib_dependencies for installing.
Returns:
The dependencies imports of the component.
"""
return {
dep: [ImportVar(tag=None, render=False)] for dep in self.lib_dependencies
}
def _get_hooks_imports(self) -> ParsedImportDict:
"""Get the imports required by certain hooks.
Returns:
The imports required for all selected hooks.
"""
imports_ = {}
if self._get_ref_hook() is not None:
# Handle hooks needed for attaching react refs to DOM nodes.
imports_.setdefault("react", set()).add(ImportVar(tag="useRef"))
imports_.setdefault(f"$/{Dirs.STATE_PATH}", set()).add(
ImportVar(tag="refs")
)
if self._get_mount_lifecycle_hook():
# Handle hooks for `on_mount` / `on_unmount`.
imports_.setdefault("react", set()).add(ImportVar(tag="useEffect"))
other_imports = []
user_hooks = self._get_hooks()
user_hooks_data = (
VarData.merge(user_hooks._get_all_var_data())
if user_hooks is not None and isinstance(user_hooks, Var)
else None
)
if user_hooks_data is not None:
other_imports.append(user_hooks_data.imports)
other_imports.extend(
hook_vardata.imports
for hook_vardata in self._get_added_hooks().values()
if hook_vardata is not None
)
return imports.merge_imports(imports_, *other_imports)
def _get_imports(self) -> ParsedImportDict:
"""Get all the libraries and fields that are used by the component.
Returns:
The imports needed by the component.
"""
cached = self.__dict__.get("_imports_cache")
if cached is not None:
return cached
imports_ = (
{self.library: [self.import_var]}
if self.library is not None and self.tag is not None
else {}
)
# Get static imports required for event processing.
event_imports = Imports.EVENTS if self.event_triggers else {}
# Collect imports from Vars used directly by this component.
var_imports = [
dict(var_data.imports)
for var in self._get_vars()
if (var_data := var._get_all_var_data()) is not None
]
added_import_dicts: list[ParsedImportDict] = []
for clz in self._iter_parent_classes_with_method("add_imports"):
list_of_import_dict = clz.add_imports(self)
if not isinstance(list_of_import_dict, list):
added_import_dicts.append(imports.parse_imports(list_of_import_dict))
else:
added_import_dicts.extend([
imports.parse_imports(item) for item in list_of_import_dict
])
result = imports.merge_parsed_imports(
self._get_dependencies_imports(),
self._get_hooks_imports(),
imports_,
event_imports,
*var_imports,
*added_import_dicts,
)
self._imports_cache = result
return result
def _get_all_imports(self, collapse: bool = False) -> ParsedImportDict:
"""Get all the libraries and fields that are used by the component and its children.
Args:
collapse: Whether to collapse the imports by removing duplicates.
Returns:
The import dict with the required imports.
"""
imports_ = imports.merge_parsed_imports(
self._get_imports(),
*[child._get_all_imports() for child in self.children],
*[
component._get_all_imports()
for component in self._get_components_in_props()
],
)
return imports.collapse_imports(imports_) if collapse else imports_
def _get_mount_lifecycle_hook(self) -> str | None:
"""Generate the component lifecycle hook.
Returns:
The useEffect hook for managing `on_mount` and `on_unmount` events.
"""
# pop on_mount and on_unmount from event_triggers since these are handled by
# hooks, not as actually props in the component
on_mount = self.event_triggers.get(EventTriggers.ON_MOUNT, None)
on_unmount = self.event_triggers.get(EventTriggers.ON_UNMOUNT, None)
if on_mount is not None:
on_mount = str(LiteralVar.create(on_mount)) + "()"
if on_unmount is not None:
on_unmount = str(LiteralVar.create(on_unmount)) + "()"
if on_mount is not None or on_unmount is not None:
return f"""
useEffect(() => {{
{on_mount or ""}
return () => {{
{on_unmount or ""}
}}
}}, []);"""
return None
def _get_ref_hook(self) -> Var | None:
"""Generate the ref hook for the component.
Returns:
The useRef hook for managing refs.
"""
ref = self.get_ref()
if ref is not None:
return Var(
f"const {ref} = useRef(null); {Var(_js_expr=ref)._as_ref()!s} = {ref};",
_var_data=VarData(position=Hooks.HookPosition.INTERNAL),
)
return None
def _get_vars_hooks(self) -> dict[str, VarData | None]:
"""Get the hooks required by vars referenced in this component.
Returns:
The hooks for the vars.
"""
vars_hooks = {}
for var in self._get_vars():
var_data = var._get_all_var_data()
if var_data is not None:
vars_hooks.update(
var_data.hooks
if isinstance(var_data.hooks, dict)
else {
k: VarData(position=Hooks.HookPosition.INTERNAL)
for k in var_data.hooks
}
)
for component in var_data.components:
vars_hooks.update(component._get_all_hooks())
return vars_hooks
def _get_events_hooks(self) -> dict[str, VarData | None]:
"""Get the hooks required by events referenced in this component.
Returns:
The hooks for the events.
"""
return (
{Hooks.EVENTS: VarData(position=Hooks.HookPosition.INTERNAL)}
if self.event_triggers
else {}
)
def _get_hooks_internal(self) -> dict[str, VarData | None]:
"""Get the React hooks for this component managed by the framework.
Downstream components should NOT override this method to avoid breaking
framework functionality.
Returns:
The internally managed hooks.
"""
cached = self.__dict__.get("_hooks_internal_cache")
if cached is not None:
return cached
result = {
**self._get_events_hooks(),
**{
str(hook): VarData(position=Hooks.HookPosition.INTERNAL)
for hook in [self._get_ref_hook(), self._get_mount_lifecycle_hook()]
if hook is not None
},
**self._get_vars_hooks(),
}
self._hooks_internal_cache = result
return result
def _get_added_hooks(self) -> dict[str, VarData | None]:
"""Get the hooks added via `add_hooks` method.
Returns:
The deduplicated hooks and imports added by the component and parent components.
"""
code = {}
def extract_var_hooks(hook: Var):
var_data = VarData.merge(hook._get_all_var_data())
if var_data is not None:
for sub_hook in var_data.hooks:
code[sub_hook] = None
if str(hook) in code:
code[str(hook)] = VarData.merge(var_data, code[str(hook)])
else:
code[str(hook)] = var_data
# Add the hook code from add_hooks for each parent class (this is reversed to preserve
# the order of the hooks in the final output)
for clz in reversed(self._iter_parent_classes_with_method("add_hooks")):
for hook in clz.add_hooks(self):
if isinstance(hook, Var):
extract_var_hooks(hook)
else:
code[hook] = None
return code
def _get_hooks(self) -> str | None:
"""Get the React hooks for this component.
Downstream components should override this method to add their own hooks.
Returns:
The hooks for just this component.
"""
return
def _get_all_hooks_internal(self) -> dict[str, VarData | None]:
"""Get the reflex internal hooks for the component and its children.
Returns:
The code that should appear just before user-defined hooks.
"""
# Store the code in a set to avoid duplicates.
code = self._get_hooks_internal()
# Add the hook code for the children.
for child in self.children:
code.update(child._get_all_hooks_internal())
return code
def _get_all_hooks(self) -> dict[str, VarData | None]:
"""Get the React hooks for this component and its children.
Returns:
The code that should appear just before returning the rendered component.
"""
code = {}
# Add the internal hooks for this component.
code.update(self._get_hooks_internal())
# Add the hook code for this component.
hooks = self._get_hooks()
if hooks is not None:
code[hooks] = None
code.update(self._get_added_hooks())
# Add the hook code for the children.
for child in self.children:
code.update(child._get_all_hooks())
return code
def get_ref(self) -> str | None:
"""Get the name of the ref for the component.
Returns:
The ref name.
"""
# do not create a ref if the id is dynamic or unspecified
if self.id is None or isinstance(self.id, Var):
return None
return format.format_ref(self.id)
def _get_all_refs(self) -> dict[str, None]:
"""Get the refs for the children of the component.
Returns:
The refs for the children.
"""
refs = {}
ref = self.get_ref()
if ref is not None:
refs[ref] = None
for child in self.children:
refs |= child._get_all_refs()
for component in self._get_components_in_props():
refs |= component._get_all_refs()
return refs
@property
def import_var(self):
"""The tag to import.
Returns:
An import var.
"""
# If the tag is dot-qualified, only import the left-most name.
tag = self.tag.partition(".")[0] if self.tag else None
alias = self.alias.partition(".")[0] if self.alias else None
return ImportVar(tag=tag, is_default=self.is_default, alias=alias)
@staticmethod
def _get_app_wrap_components() -> dict[tuple[int, str], Component]:
"""Get the app wrap components for the component.
Returns:
The app wrap components.
"""
return {}
def _get_all_app_wrap_components(
self, *, ignore_ids: set[int] | None = None
) -> dict[tuple[int, str], Component]:
"""Get the app wrap components for the component and its children.
Args:
ignore_ids: A set of component IDs to ignore. Used to avoid duplicates.
Returns:
The app wrap components.
"""
ignore_ids = ignore_ids or set()
# Store the components in a set to avoid duplicates.
components = self._get_app_wrap_components()
for component in tuple(components.values()):
component_id = id(component)
if component_id in ignore_ids:
continue
ignore_ids.add(component_id)
components.update(
component._get_all_app_wrap_components(ignore_ids=ignore_ids)
)
# Add the app wrap components for the children.
for child in self.children:
child_id = id(child)
# Skip non-Component children.
if not isinstance(child, Component) or child_id in ignore_ids:
continue
ignore_ids.add(child_id)
components.update(child._get_all_app_wrap_components(ignore_ids=ignore_ids))
# Return the components.
return components
class CustomComponent(Component):
"""A custom user-defined component."""
# Use the components library.
library = f"$/{Dirs.COMPONENTS_PATH}"
component_fn: Callable[..., Component] = field(
doc="The function that creates the component.", default=Component.create
)
props: dict[str, Any] = field(
doc="The props of the component.", default_factory=dict
)
def _post_init(self, **kwargs):
"""Initialize the custom component.
Args:
**kwargs: The kwargs to pass to the component.
"""
component_fn = kwargs.get("component_fn")
# Set the props.
props_types = typing.get_type_hints(component_fn) if component_fn else {}
props = {key: value for key, value in kwargs.items() if key in props_types}
kwargs = {key: value for key, value in kwargs.items() if key not in props_types}
event_types = {
key
for key in props
if (
(get_origin((annotation := props_types.get(key))) or annotation)
== EventHandler
)
}
def get_args_spec(key: str) -> types.ArgsSpec | Sequence[types.ArgsSpec]:
type_ = props_types[key]
return (
args[0]
if (args := get_args(type_))
else (
annotation_args[1]
if get_origin(
annotation := inspect.getfullargspec(component_fn).annotations[
key
]
)
is typing.Annotated
and (annotation_args := get_args(annotation))
else no_args_event_spec
)
)
super()._post_init(
event_triggers={
key: EventChain.create(
value=props[key],
args_spec=get_args_spec(key),
key=key,
)
for key in event_types
},
**kwargs,
)
to_camel_cased_props = {
format.to_camel_case(key): None for key in props if key not in event_types
}
self.get_props = lambda: to_camel_cased_props # pyright: ignore [reportIncompatibleVariableOverride]
# Unset the style.
self.style = Style()
# Set the tag to the name of the function.
self.tag = format.to_title_case(self.component_fn.__name__)
for key, value in props.items():
# Skip kwargs that are not props.
if key not in props_types:
continue
camel_cased_key = format.to_camel_case(key)
# Get the type based on the annotation.
type_ = props_types[key]
# Handle event chains.
if type_ is EventHandler:
inspect.getfullargspec(component_fn).annotations[key]
self.props[camel_cased_key] = EventChain.create(
value=value, args_spec=get_args_spec(key), key=key
)
continue
value = LiteralVar.create(value)
self.props[camel_cased_key] = value
setattr(self, camel_cased_key, value)
def __eq__(self, other: Any) -> bool:
"""Check if the component is equal to another.
Args:
other: The other component.
Returns:
Whether the component is equal to the other.
"""
return isinstance(other, CustomComponent) and self.tag == other.tag
def __hash__(self) -> int:
"""Get the hash of the component.
Returns:
The hash of the component.
"""
return hash(self.tag)
@classmethod
def get_props(cls) -> Iterable[str]:
"""Get the props for the component.
Returns:
The set of component props.
"""
return ()
@staticmethod
def _get_event_spec_from_args_spec(name: str, event: EventChain) -> Callable:
"""Get the event spec from the args spec.
Args:
name: The name of the event
event: The args spec.
Returns:
The event spec.
"""
def fn(*args):
return run_script(Var(name).to(FunctionVar).call(*args))
if event.args_spec:
arg_spec = (
event.args_spec
if not isinstance(event.args_spec, Sequence)
else event.args_spec[0]
)
names = inspect.getfullargspec(arg_spec).args
fn.__signature__ = inspect.Signature( # pyright: ignore[reportFunctionMemberAccess]
parameters=[
inspect.Parameter(
name=name,
kind=inspect.Parameter.POSITIONAL_ONLY,
annotation=arg._var_type,
)
for name, arg in zip(
names, parse_args_spec(event.args_spec)[0], strict=True
)
]
)
return fn
def get_prop_vars(self) -> list[Var | Callable]:
"""Get the prop vars.
Returns:
The prop vars.
"""
return [
Var(
_js_expr=name + CAMEL_CASE_MEMO_MARKER,
_var_type=(prop._var_type if isinstance(prop, Var) else type(prop)),
).guess_type()
if isinstance(prop, Var) or not isinstance(prop, EventChain)
else CustomComponent._get_event_spec_from_args_spec(
name + CAMEL_CASE_MEMO_MARKER, prop
)
for name, prop in self.props.items()
]
@functools.cache # noqa: B019
def get_component(self) -> Component:
"""Render the component.
Returns:
The code to render the component.
"""
component = self.component_fn(*self.get_prop_vars())
try:
from reflex.utils.prerequisites import get_and_validate_app
style = get_and_validate_app().app.style
except Exception:
style = {}
component._add_style_recursive(style)
return component
def _get_all_app_wrap_components(
self, *, ignore_ids: set[int] | None = None
) -> dict[tuple[int, str], Component]:
"""Get the app wrap components for the custom component.
Args:
ignore_ids: A set of IDs to ignore to avoid infinite recursion.
Returns:
The app wrap components.
"""
ignore_ids = ignore_ids or set()
component = self.get_component()
if id(component) in ignore_ids:
return {}
ignore_ids.add(id(component))
return self.get_component()._get_all_app_wrap_components(ignore_ids=ignore_ids)
CUSTOM_COMPONENTS: dict[str, CustomComponent] = {}
def _register_custom_component(
component_fn: Callable[..., Component],
):
"""Register a custom component to be compiled.
Args:
component_fn: The function that creates the component.
Returns:
The custom component.
Raises:
TypeError: If the tag name cannot be determined.
"""
dummy_props = {
prop: (
Var(
"",
_var_type=unwrap_var_annotation(annotation),
).guess_type()
if not types.safe_issubclass(annotation, EventHandler)
else EventSpec(handler=EventHandler(fn=no_args_event_spec))
)
for prop, annotation in typing.get_type_hints(component_fn).items()
if prop != "return"
}
dummy_component = CustomComponent._create(
children=[],
component_fn=component_fn,
**dummy_props,
)
if dummy_component.tag is None:
msg = f"Could not determine the tag name for {component_fn!r}"
raise TypeError(msg)
CUSTOM_COMPONENTS[dummy_component.tag] = dummy_component
return dummy_component
def custom_component(
component_fn: Callable[..., Component],
) -> Callable[..., CustomComponent]:
"""Create a custom component from a function.
Args:
component_fn: The function that creates the component.
Returns:
The decorated function.
"""
@wraps(component_fn)
def wrapper(*children, **props) -> CustomComponent:
# Remove the children from the props.
props.pop("children", None)
return CustomComponent._create(
children=list(children), component_fn=component_fn, **props
)
# Register this component so it can be compiled.
dummy_component = _register_custom_component(component_fn)
if tag := dummy_component.tag:
object.__setattr__(
wrapper,
"_as_var",
lambda: Var(
tag,
_var_type=type[Component],
_var_data=VarData(
imports={
f"$/{constants.Dirs.UTILS}/components": [ImportVar(tag=tag)],
"@emotion/react": [
ImportVar(tag="jsx"),
],
}
),
),
)
return wrapper
# Alias memo to custom_component.
memo = custom_component
class NoSSRComponent(Component):
"""A dynamic component that is not rendered on the server."""
def _get_import_name(self) -> str | None:
if not self.library:
return None
return f"${self.library}" if self.library.startswith("/") else self.library
def _get_imports(self) -> ParsedImportDict:
"""Get the imports for the component.
Returns:
The imports for dynamically importing the component at module load time.
"""
# React lazy import mechanism.
dynamic_import = {
f"$/{constants.Dirs.UTILS}/context": [ImportVar(tag="ClientSide")],
}
# The normal imports for this component.
imports_ = super()._get_imports()
# Do NOT import the main library/tag statically.
import_name = self._get_import_name()
if import_name is not None:
with contextlib.suppress(ValueError):
imports_[import_name].remove(self.import_var)
imports_[import_name].append(ImportVar(tag=None, render=False))
return imports.merge_imports(
dynamic_import,
imports_,
self._get_dependencies_imports(),
)
def _get_dynamic_imports(self) -> str:
# extract the correct import name from library name
base_import_name = self._get_import_name()
if base_import_name is None:
msg = "Undefined library for NoSSRComponent"
raise ValueError(msg)
import_name = format.format_library_name(base_import_name)
library_import = f"import('{import_name}')"
mod_import = (
# https://nextjs.org/docs/pages/building-your-application/optimizing/lazy-loading#with-named-exports
f".then((mod) => mod.{self.tag})"
if not self.is_default
else ".then((mod) => mod.default.default ?? mod.default)"
)
return (
f"const {self.alias or self.tag} = ClientSide(() => "
+ library_import
+ mod_import
+ ")"
)
class MemoizationLeaf(Component):
"""A component that does not separately memoize its children.
Any component which depends on finding the exact names of children
components within it, should be a memoization leaf so the compiler
does not replace the provided child tags with memoized tags.
Whether the leaf is wrapped in a memo definition is decided by the
compiler's snapshot-boundary subtree scan, not by a class-local
disposition override — so leaves and components that explicitly set
``_memoization_mode = MemoizationMode(recursive=False)`` are handled
identically.
"""
_memoization_mode = MemoizationMode(recursive=False)
load_dynamic_serializer()
class ComponentVar(Var[Component], python_types=BaseComponent):
"""A Var that represents a Component."""
def empty_component() -> Component:
"""Create an empty component.
Returns:
An empty component.
"""
from reflex_components_core.base.bare import Bare
return Bare.create("")
def render_dict_to_var(tag: dict | Component | str) -> Var:
"""Convert a render dict to a Var.
Args:
tag: The render dict.
Returns:
The Var.
"""
if not isinstance(tag, dict):
if isinstance(tag, Component):
return render_dict_to_var(tag.render())
return Var.create(tag)
if "contents" in tag:
return Var(tag["contents"])
if "iterable" in tag:
function_return = LiteralArrayVar.create([
render_dict_to_var(child.render()) for child in tag["children"]
])
func = ArgsFunctionOperation.create(
(tag["arg_var_name"], tag["index_var_name"]),
function_return,
)
return FunctionStringVar.create("Array.prototype.map.call").call(
tag["iterable"]
if not isinstance(tag["iterable"], ObjectVar)
else tag["iterable"].items(),
func,
)
if "match_cases" in tag:
element = Var(tag["cond"])
conditionals = render_dict_to_var(tag["default"])
for case in tag["match_cases"][::-1]:
conditions, return_value = case
condition = Var.create(False)
for pattern in conditions:
condition = condition | (
Var(pattern).to_string() == element.to_string()
)
conditionals = ternary_operation(
condition,
render_dict_to_var(return_value),
conditionals,
)
return conditionals
if "cond_state" in tag:
return ternary_operation(
Var(tag["cond_state"]),
render_dict_to_var(tag["true_value"]),
render_dict_to_var(tag["false_value"])
if tag["false_value"] is not None
else LiteralNoneVar.create(),
)
props = Var("({" + ",".join(tag["props"]) + "})")
raw_tag_name = tag.get("name")
tag_name = Var(raw_tag_name or "Fragment")
return FunctionStringVar.create(
"jsx",
).call(
tag_name,
props,
*[render_dict_to_var(child) for child in tag["children"]],
)
@dataclasses.dataclass(
eq=False,
frozen=True,
slots=True,
)
class LiteralComponentVar(CachedVarOperation, LiteralVar[Component], ComponentVar):
"""A Var that represents a Component."""
_var_value: BaseComponent = dataclasses.field(default_factory=empty_component)
@cached_property_no_lock
def _cached_var_name(self) -> str:
"""Get the name of the var.
Returns:
The name of the var.
"""
return str(render_dict_to_var(self._var_value.render()))
@cached_property_no_lock
def _cached_get_all_var_data(self) -> VarData | None:
"""Get the VarData for the var.
Returns:
The VarData for the var.
"""
return VarData.merge(
self._var_data,
VarData(
imports={
"@emotion/react": ["jsx"],
"react": ["Fragment"],
},
),
VarData(
imports=self._var_value._get_all_imports(),
),
)
def __hash__(self) -> int:
"""Get the hash of the var.
Returns:
The hash of the var.
"""
return hash((type(self).__name__, self._js_expr))
@classmethod
def create(
cls,
value: Component,
_var_data: VarData | None = None,
):
"""Create a var from a value.
Args:
value: The value of the var.
_var_data: Additional hooks and imports associated with the Var.
Returns:
The var.
"""
var_datas = [
var_data
for var in value._get_vars(include_children=True)
if (var_data := var._get_all_var_data())
]
return LiteralComponentVar(
_js_expr="",
_var_type=type(value),
_var_data=VarData.merge(
_var_data,
*var_datas,
VarData(
components=(value,),
),
),
_var_value=value,
)