123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941 |
- """Define event classes to connect the frontend and backend."""
- from __future__ import annotations
- import dataclasses
- import inspect
- import sys
- import types
- import urllib.parse
- from base64 import b64encode
- from functools import partial
- from typing import (
- TYPE_CHECKING,
- Any,
- Callable,
- Dict,
- Generic,
- List,
- Optional,
- Sequence,
- Tuple,
- Type,
- Union,
- get_type_hints,
- overload,
- )
- import typing_extensions
- from typing_extensions import (
- Concatenate,
- ParamSpec,
- Protocol,
- TypeAliasType,
- TypedDict,
- TypeVar,
- get_args,
- get_origin,
- )
- from reflex import constants
- from reflex.constants.state import FRONTEND_EVENT_STATE
- from reflex.utils import console, format
- from reflex.utils.exceptions import EventFnArgMismatch, EventHandlerArgTypeMismatch
- from reflex.utils.types import ArgsSpec, GenericType, typehint_issubclass
- from reflex.vars import VarData
- from reflex.vars.base import LiteralVar, Var
- from reflex.vars.function import (
- ArgsFunctionOperation,
- ArgsFunctionOperationBuilder,
- BuilderFunctionVar,
- FunctionArgs,
- FunctionStringVar,
- FunctionVar,
- VarOperationCall,
- )
- from reflex.vars.object import ObjectVar
- try:
- from typing import Annotated
- except ImportError:
- from typing_extensions import Annotated
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class Event:
- """An event that describes any state change in the app."""
- # The token to specify the client that the event is for.
- token: str
- # The event name.
- name: str
- # The routing data where event occurred
- router_data: Dict[str, Any] = dataclasses.field(default_factory=dict)
- # The event payload.
- payload: Dict[str, Any] = dataclasses.field(default_factory=dict)
- @property
- def substate_token(self) -> str:
- """Get the substate token for the event.
- Returns:
- The substate token.
- """
- substate = self.name.rpartition(".")[0]
- return f"{self.token}_{substate}"
- BACKGROUND_TASK_MARKER = "_reflex_background_task"
- def background(fn, *, __internal_reflex_call: bool = False):
- """Decorator to mark event handler as running in the background.
- Args:
- fn: The function to decorate.
- Returns:
- The same function, but with a marker set.
- Raises:
- TypeError: If the function is not a coroutine function or async generator.
- """
- if not __internal_reflex_call:
- console.deprecate(
- "background-decorator",
- "Use `rx.event(background=True)` instead.",
- "0.6.5",
- "0.7.0",
- )
- if not inspect.iscoroutinefunction(fn) and not inspect.isasyncgenfunction(fn):
- raise TypeError("Background task must be async function or generator.")
- setattr(fn, BACKGROUND_TASK_MARKER, True)
- return fn
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class EventActionsMixin:
- """Mixin for DOM event actions."""
- # Whether to `preventDefault` or `stopPropagation` on the event.
- event_actions: Dict[str, Union[bool, int]] = dataclasses.field(default_factory=dict)
- @property
- def stop_propagation(self):
- """Stop the event from bubbling up the DOM tree.
- Returns:
- New EventHandler-like with stopPropagation set to True.
- """
- return dataclasses.replace(
- self,
- event_actions={"stopPropagation": True, **self.event_actions},
- )
- @property
- def prevent_default(self):
- """Prevent the default behavior of the event.
- Returns:
- New EventHandler-like with preventDefault set to True.
- """
- return dataclasses.replace(
- self,
- event_actions={"preventDefault": True, **self.event_actions},
- )
- def throttle(self, limit_ms: int):
- """Throttle the event handler.
- Args:
- limit_ms: The time in milliseconds to throttle the event handler.
- Returns:
- New EventHandler-like with throttle set to limit_ms.
- """
- return dataclasses.replace(
- self,
- event_actions={"throttle": limit_ms, **self.event_actions},
- )
- def debounce(self, delay_ms: int):
- """Debounce the event handler.
- Args:
- delay_ms: The time in milliseconds to debounce the event handler.
- Returns:
- New EventHandler-like with debounce set to delay_ms.
- """
- return dataclasses.replace(
- self,
- event_actions={"debounce": delay_ms, **self.event_actions},
- )
- @property
- def temporal(self):
- """Do not queue the event if the backend is down.
- Returns:
- New EventHandler-like with temporal set to True.
- """
- return dataclasses.replace(
- self,
- event_actions={"temporal": True, **self.event_actions},
- )
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class EventHandler(EventActionsMixin):
- """An event handler responds to an event to update the state."""
- # The function to call in response to the event.
- fn: Any = dataclasses.field(default=None)
- # The full name of the state class this event handler is attached to.
- # Empty string means this event handler is a server side event.
- state_full_name: str = dataclasses.field(default="")
- @classmethod
- def __class_getitem__(cls, args_spec: str) -> Annotated:
- """Get a typed EventHandler.
- Args:
- args_spec: The args_spec of the EventHandler.
- Returns:
- The EventHandler class item.
- """
- return Annotated[cls, args_spec]
- @property
- def is_background(self) -> bool:
- """Whether the event handler is a background task.
- Returns:
- True if the event handler is marked as a background task.
- """
- return getattr(self.fn, BACKGROUND_TASK_MARKER, False)
- def __call__(self, *args: Any) -> EventSpec:
- """Pass arguments to the handler to get an event spec.
- This method configures event handlers that take in arguments.
- Args:
- *args: The arguments to pass to the handler.
- Returns:
- The event spec, containing both the function and args.
- Raises:
- EventHandlerTypeError: If the arguments are invalid.
- """
- from reflex.utils.exceptions import EventHandlerTypeError
- # Get the function args.
- fn_args = inspect.getfullargspec(self.fn).args[1:]
- fn_args = (Var(_js_expr=arg) for arg in fn_args)
- # Construct the payload.
- values = []
- for arg in args:
- # Special case for file uploads.
- if isinstance(arg, FileUpload):
- return arg.as_event_spec(handler=self)
- # Otherwise, convert to JSON.
- try:
- values.append(LiteralVar.create(arg))
- except TypeError as e:
- raise EventHandlerTypeError(
- f"Arguments to event handlers must be Vars or JSON-serializable. Got {arg} of type {type(arg)}."
- ) from e
- payload = tuple(zip(fn_args, values))
- # Return the event spec.
- return EventSpec(
- handler=self, args=payload, event_actions=self.event_actions.copy()
- )
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class EventSpec(EventActionsMixin):
- """An event specification.
- Whereas an Event object is passed during runtime, a spec is used
- during compile time to outline the structure of an event.
- """
- # The event handler.
- handler: EventHandler = dataclasses.field(default=None) # type: ignore
- # The handler on the client to process event.
- client_handler_name: str = dataclasses.field(default="")
- # The arguments to pass to the function.
- args: Tuple[Tuple[Var, Var], ...] = dataclasses.field(default_factory=tuple)
- def __init__(
- self,
- handler: EventHandler,
- event_actions: Dict[str, Union[bool, int]] | None = None,
- client_handler_name: str = "",
- args: Tuple[Tuple[Var, Var], ...] = (),
- ):
- """Initialize an EventSpec.
- Args:
- event_actions: The event actions.
- handler: The event handler.
- client_handler_name: The client handler name.
- args: The arguments to pass to the function.
- """
- if event_actions is None:
- event_actions = {}
- object.__setattr__(self, "event_actions", event_actions)
- object.__setattr__(self, "handler", handler)
- object.__setattr__(self, "client_handler_name", client_handler_name)
- object.__setattr__(self, "args", args or ())
- def with_args(self, args: Tuple[Tuple[Var, Var], ...]) -> EventSpec:
- """Copy the event spec, with updated args.
- Args:
- args: The new args to pass to the function.
- Returns:
- A copy of the event spec, with the new args.
- """
- return type(self)(
- handler=self.handler,
- client_handler_name=self.client_handler_name,
- args=args,
- event_actions=self.event_actions.copy(),
- )
- def add_args(self, *args: Var) -> EventSpec:
- """Add arguments to the event spec.
- Args:
- *args: The arguments to add positionally.
- Returns:
- The event spec with the new arguments.
- Raises:
- EventHandlerTypeError: If the arguments are invalid.
- """
- from reflex.utils.exceptions import EventHandlerTypeError
- # Get the remaining unfilled function args.
- fn_args = inspect.getfullargspec(self.handler.fn).args[1 + len(self.args) :]
- fn_args = (Var(_js_expr=arg) for arg in fn_args)
- # Construct the payload.
- values = []
- arg = None
- try:
- for arg in args:
- values.append(LiteralVar.create(value=arg)) # noqa: PERF401
- except TypeError as e:
- raise EventHandlerTypeError(
- f"Arguments to event handlers must be Vars or JSON-serializable. Got {arg} of type {type(arg)}."
- ) from e
- new_payload = tuple(zip(fn_args, values))
- return self.with_args(self.args + new_payload)
- @dataclasses.dataclass(
- frozen=True,
- )
- class CallableEventSpec(EventSpec):
- """Decorate an EventSpec-returning function to act as both a EventSpec and a function.
- This is used as a compatibility shim for replacing EventSpec objects in the
- API with functions that return a family of EventSpec.
- """
- fn: Optional[Callable[..., EventSpec]] = None
- def __init__(self, fn: Callable[..., EventSpec] | None = None, **kwargs):
- """Initialize a CallableEventSpec.
- Args:
- fn: The function to decorate.
- **kwargs: The kwargs to pass to pydantic initializer
- """
- if fn is not None:
- default_event_spec = fn()
- super().__init__(
- event_actions=default_event_spec.event_actions,
- client_handler_name=default_event_spec.client_handler_name,
- args=default_event_spec.args,
- handler=default_event_spec.handler,
- **kwargs,
- )
- object.__setattr__(self, "fn", fn)
- else:
- super().__init__(**kwargs)
- def __call__(self, *args, **kwargs) -> EventSpec:
- """Call the decorated function.
- Args:
- *args: The args to pass to the function.
- **kwargs: The kwargs to pass to the function.
- Returns:
- The EventSpec returned from calling the function.
- Raises:
- EventHandlerTypeError: If the CallableEventSpec has no associated function.
- """
- from reflex.utils.exceptions import EventHandlerTypeError
- if self.fn is None:
- raise EventHandlerTypeError("CallableEventSpec has no associated function.")
- return self.fn(*args, **kwargs)
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class EventChain(EventActionsMixin):
- """Container for a chain of events that will be executed in order."""
- events: Sequence[Union[EventSpec, EventVar, EventCallback]] = dataclasses.field(
- default_factory=list
- )
- args_spec: Optional[Union[Callable, Sequence[Callable]]] = dataclasses.field(
- default=None
- )
- invocation: Optional[Var] = dataclasses.field(default=None)
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class JavascriptHTMLInputElement:
- """Interface for a Javascript HTMLInputElement https://developer.mozilla.org/en-US/docs/Web/API/HTMLInputElement."""
- value: str = ""
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class JavascriptInputEvent:
- """Interface for a Javascript InputEvent https://developer.mozilla.org/en-US/docs/Web/API/InputEvent."""
- target: JavascriptHTMLInputElement = JavascriptHTMLInputElement() # noqa: RUF009
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class JavasciptKeyboardEvent:
- """Interface for a Javascript KeyboardEvent https://developer.mozilla.org/en-US/docs/Web/API/KeyboardEvent."""
- key: str = ""
- altKey: bool = False
- ctrlKey: bool = False
- metaKey: bool = False
- shiftKey: bool = False
- def input_event(e: Var[JavascriptInputEvent]) -> Tuple[Var[str]]:
- """Get the value from an input event.
- Args:
- e: The input event.
- Returns:
- The value from the input event.
- """
- return (e.target.value,)
- class KeyInputInfo(TypedDict):
- """Information about a key input event."""
- alt_key: bool
- ctrl_key: bool
- meta_key: bool
- shift_key: bool
- def key_event(e: Var[JavasciptKeyboardEvent]) -> Tuple[Var[str], Var[KeyInputInfo]]:
- """Get the key from a keyboard event.
- Args:
- e: The keyboard event.
- Returns:
- The key from the keyboard event.
- """
- return (
- e.key,
- Var.create(
- {
- "alt_key": e.altKey,
- "ctrl_key": e.ctrlKey,
- "meta_key": e.metaKey,
- "shift_key": e.shiftKey,
- },
- ),
- )
- def no_args_event_spec() -> Tuple[()]:
- """Empty event handler.
- Returns:
- An empty tuple.
- """
- return () # type: ignore
- # These chains can be used for their side effects when no other events are desired.
- stop_propagation = EventChain(events=[], args_spec=no_args_event_spec).stop_propagation
- prevent_default = EventChain(events=[], args_spec=no_args_event_spec).prevent_default
- T = TypeVar("T")
- U = TypeVar("U")
- class IdentityEventReturn(Generic[T], Protocol):
- """Protocol for an identity event return."""
- def __call__(self, *values: Var[T]) -> Tuple[Var[T], ...]:
- """Return the input values.
- Args:
- *values: The values to return.
- Returns:
- The input values.
- """
- return values
- @overload
- def passthrough_event_spec(
- event_type: Type[T], /
- ) -> Callable[[Var[T]], Tuple[Var[T]]]: ... # type: ignore
- @overload
- def passthrough_event_spec(
- event_type_1: Type[T], event_type2: Type[U], /
- ) -> Callable[[Var[T], Var[U]], Tuple[Var[T], Var[U]]]: ...
- @overload
- def passthrough_event_spec(*event_types: Type[T]) -> IdentityEventReturn[T]: ...
- def passthrough_event_spec(*event_types: Type[T]) -> IdentityEventReturn[T]: # type: ignore
- """A helper function that returns the input event as output.
- Args:
- *event_types: The types of the events.
- Returns:
- A function that returns the input event as output.
- """
- def inner(*values: Var[T]) -> Tuple[Var[T], ...]:
- return values
- inner_type = tuple(Var[event_type] for event_type in event_types)
- return_annotation = Tuple[inner_type] # type: ignore
- inner.__signature__ = inspect.signature(inner).replace( # type: ignore
- parameters=[
- inspect.Parameter(
- f"ev_{i}",
- kind=inspect.Parameter.POSITIONAL_OR_KEYWORD,
- annotation=Var[event_type],
- )
- for i, event_type in enumerate(event_types)
- ],
- return_annotation=return_annotation,
- )
- for i, event_type in enumerate(event_types):
- inner.__annotations__[f"ev_{i}"] = Var[event_type]
- inner.__annotations__["return"] = return_annotation
- return inner
- @dataclasses.dataclass(
- init=True,
- frozen=True,
- )
- class FileUpload:
- """Class to represent a file upload."""
- upload_id: Optional[str] = None
- on_upload_progress: Optional[Union[EventHandler, Callable]] = None
- @staticmethod
- def on_upload_progress_args_spec(_prog: Var[Dict[str, Union[int, float, bool]]]):
- """Args spec for on_upload_progress event handler.
- Returns:
- The arg mapping passed to backend event handler
- """
- return [_prog]
- def as_event_spec(self, handler: EventHandler) -> EventSpec:
- """Get the EventSpec for the file upload.
- Args:
- handler: The event handler.
- Returns:
- The event spec for the handler.
- Raises:
- ValueError: If the on_upload_progress is not a valid event handler.
- """
- from reflex.components.core.upload import (
- DEFAULT_UPLOAD_ID,
- upload_files_context_var_data,
- )
- upload_id = self.upload_id or DEFAULT_UPLOAD_ID
- spec_args = [
- (
- Var(_js_expr="files"),
- Var(
- _js_expr="filesById",
- _var_type=Dict[str, Any],
- _var_data=VarData.merge(upload_files_context_var_data),
- ).to(ObjectVar)[LiteralVar.create(upload_id)],
- ),
- (
- Var(_js_expr="upload_id"),
- LiteralVar.create(upload_id),
- ),
- ]
- if self.on_upload_progress is not None:
- on_upload_progress = self.on_upload_progress
- if isinstance(on_upload_progress, EventHandler):
- events = [
- call_event_handler(
- on_upload_progress,
- self.on_upload_progress_args_spec,
- ),
- ]
- elif isinstance(on_upload_progress, Callable):
- # Call the lambda to get the event chain.
- events = call_event_fn(
- on_upload_progress, self.on_upload_progress_args_spec
- ) # type: ignore
- else:
- raise ValueError(f"{on_upload_progress} is not a valid event handler.")
- if isinstance(events, Var):
- raise ValueError(f"{on_upload_progress} cannot return a var {events}.")
- on_upload_progress_chain = EventChain(
- events=[*events],
- args_spec=self.on_upload_progress_args_spec,
- )
- formatted_chain = str(format.format_prop(on_upload_progress_chain))
- spec_args.append(
- (
- Var(_js_expr="on_upload_progress"),
- FunctionStringVar(
- formatted_chain.strip("{}"),
- ).to(FunctionVar, EventChain),
- ),
- )
- return EventSpec(
- handler=handler,
- client_handler_name="uploadFiles",
- args=tuple(spec_args),
- event_actions=handler.event_actions.copy(),
- )
- # Alias for rx.upload_files
- upload_files = FileUpload
- # Special server-side events.
- def server_side(name: str, sig: inspect.Signature, **kwargs) -> EventSpec:
- """A server-side event.
- Args:
- name: The name of the event.
- sig: The function signature of the event.
- **kwargs: The arguments to pass to the event.
- Returns:
- An event spec for a server-side event.
- """
- def fn():
- return None
- fn.__qualname__ = name
- fn.__signature__ = sig
- return EventSpec(
- handler=EventHandler(fn=fn, state_full_name=FRONTEND_EVENT_STATE),
- args=tuple(
- (
- Var(_js_expr=k),
- LiteralVar.create(v),
- )
- for k, v in kwargs.items()
- ),
- )
- @overload
- def redirect(
- path: str | Var[str],
- is_external: Optional[bool] = None,
- replace: bool = False,
- ) -> EventSpec: ...
- @overload
- @typing_extensions.deprecated("`external` is deprecated use `is_external` instead")
- def redirect(
- path: str | Var[str],
- is_external: Optional[bool] = None,
- replace: bool = False,
- external: Optional[bool] = None,
- ) -> EventSpec: ...
- def redirect(
- path: str | Var[str],
- is_external: Optional[bool] = None,
- replace: bool = False,
- external: Optional[bool] = None,
- ) -> EventSpec:
- """Redirect to a new path.
- Args:
- path: The path to redirect to.
- is_external: Whether to open in new tab or not.
- replace: If True, the current page will not create a new history entry.
- external(Deprecated): Whether to open in new tab or not.
- Returns:
- An event to redirect to the path.
- """
- if external is not None:
- console.deprecate(
- "The `external` prop in `rx.redirect`",
- "use `is_external` instead.",
- "0.6.6",
- "0.7.0",
- )
- # is_external should take precedence over external.
- is_external = (
- (False if external is None else external)
- if is_external is None
- else is_external
- )
- return server_side(
- "_redirect",
- get_fn_signature(redirect),
- path=path,
- external=is_external,
- replace=replace,
- )
- def console_log(message: str | Var[str]) -> EventSpec:
- """Do a console.log on the browser.
- Args:
- message: The message to log.
- Returns:
- An event to log the message.
- """
- return run_script(Var("console").to(dict).log.to(FunctionVar).call(message))
- def noop() -> EventSpec:
- """Do nothing.
- Returns:
- An event to do nothing.
- """
- return run_script(Var.create(None))
- def back() -> EventSpec:
- """Do a history.back on the browser.
- Returns:
- An event to go back one page.
- """
- return run_script(
- Var("window").to(dict).history.to(dict).back.to(FunctionVar).call()
- )
- def window_alert(message: str | Var[str]) -> EventSpec:
- """Create a window alert on the browser.
- Args:
- message: The message to alert.
- Returns:
- An event to alert the message.
- """
- return run_script(Var("window").to(dict).alert.to(FunctionVar).call(message))
- def set_focus(ref: str) -> EventSpec:
- """Set focus to specified ref.
- Args:
- ref: The ref.
- Returns:
- An event to set focus on the ref
- """
- return server_side(
- "_set_focus",
- get_fn_signature(set_focus),
- ref=LiteralVar.create(format.format_ref(ref)),
- )
- def scroll_to(elem_id: str, align_to_top: bool | Var[bool] = True) -> EventSpec:
- """Select the id of a html element for scrolling into view.
- Args:
- elem_id: The id of the element to scroll to.
- align_to_top: Whether to scroll to the top (True) or bottom (False) of the element.
- Returns:
- An EventSpec to scroll the page to the selected element.
- """
- get_element_by_id = FunctionStringVar.create("document.getElementById")
- return run_script(
- get_element_by_id.call(elem_id)
- .to(ObjectVar)
- .scrollIntoView.to(FunctionVar)
- .call(align_to_top),
- )
- def set_value(ref: str, value: Any) -> EventSpec:
- """Set the value of a ref.
- Args:
- ref: The ref.
- value: The value to set.
- Returns:
- An event to set the ref.
- """
- return server_side(
- "_set_value",
- get_fn_signature(set_value),
- ref=LiteralVar.create(format.format_ref(ref)),
- value=value,
- )
- def remove_cookie(key: str, options: dict[str, Any] | None = None) -> EventSpec:
- """Remove a cookie on the frontend.
- Args:
- key: The key identifying the cookie to be removed.
- options: Support all the cookie options from RFC 6265
- Returns:
- EventSpec: An event to remove a cookie.
- """
- options = options or {}
- options["path"] = options.get("path", "/")
- return server_side(
- "_remove_cookie",
- get_fn_signature(remove_cookie),
- key=key,
- options=options,
- )
- def clear_local_storage() -> EventSpec:
- """Set a value in the local storage on the frontend.
- Returns:
- EventSpec: An event to clear the local storage.
- """
- return server_side(
- "_clear_local_storage",
- get_fn_signature(clear_local_storage),
- )
- def remove_local_storage(key: str) -> EventSpec:
- """Set a value in the local storage on the frontend.
- Args:
- key: The key identifying the variable in the local storage to remove.
- Returns:
- EventSpec: An event to remove an item based on the provided key in local storage.
- """
- return server_side(
- "_remove_local_storage",
- get_fn_signature(remove_local_storage),
- key=key,
- )
- def clear_session_storage() -> EventSpec:
- """Set a value in the session storage on the frontend.
- Returns:
- EventSpec: An event to clear the session storage.
- """
- return server_side(
- "_clear_session_storage",
- get_fn_signature(clear_session_storage),
- )
- def remove_session_storage(key: str) -> EventSpec:
- """Set a value in the session storage on the frontend.
- Args:
- key: The key identifying the variable in the session storage to remove.
- Returns:
- EventSpec: An event to remove an item based on the provided key in session storage.
- """
- return server_side(
- "_remove_session_storage",
- get_fn_signature(remove_session_storage),
- key=key,
- )
- def set_clipboard(content: Union[str, Var[str]]) -> EventSpec:
- """Set the text in content in the clipboard.
- Args:
- content: The text to add to clipboard.
- Returns:
- EventSpec: An event to set some content in the clipboard.
- """
- return run_script(
- Var("navigator")
- .to(dict)
- .clipboard.to(dict)
- .writeText.to(FunctionVar)
- .call(content)
- )
- def download(
- url: str | Var | None = None,
- filename: Optional[str | Var] = None,
- data: str | bytes | Var | None = None,
- ) -> EventSpec:
- """Download the file at a given path or with the specified data.
- Args:
- url: The URL to the file to download.
- filename: The name that the file should be saved as after download.
- data: The data to download.
- Raises:
- ValueError: If the URL provided is invalid, both URL and data are provided,
- or the data is not an expected type.
- Returns:
- EventSpec: An event to download the associated file.
- """
- from reflex.components.core.cond import cond
- if isinstance(url, str):
- if not url.startswith("/"):
- raise ValueError("The URL argument should start with a /")
- # if filename is not provided, infer it from url
- if filename is None:
- filename = url.rpartition("/")[-1]
- if filename is None:
- filename = ""
- if data is not None:
- if url is not None:
- raise ValueError("Cannot provide both URL and data to download.")
- if isinstance(data, str):
- # Caller provided a plain text string to download.
- url = "data:text/plain," + urllib.parse.quote(data)
- elif isinstance(data, Var):
- # Need to check on the frontend if the Var already looks like a data: URI.
- is_data_url = (data.js_type() == "string") & (
- data.to(str).startswith("data:")
- ) # type: ignore
- # If it's a data: URI, use it as is, otherwise convert the Var to JSON in a data: URI.
- url = cond( # type: ignore
- is_data_url,
- data.to(str),
- "data:text/plain," + data.to_string(), # type: ignore
- )
- elif isinstance(data, bytes):
- # Caller provided bytes, so base64 encode it as a data: URI.
- b64_data = b64encode(data).decode("utf-8")
- url = "data:application/octet-stream;base64," + b64_data
- else:
- raise ValueError(
- f"Invalid data type {type(data)} for download. Use `str` or `bytes`."
- )
- return server_side(
- "_download",
- get_fn_signature(download),
- url=url,
- filename=filename,
- )
- def _callback_arg_spec(eval_result):
- """ArgSpec for call_script callback function.
- Args:
- eval_result: The result of the javascript execution.
- Returns:
- Args for the callback function
- """
- return [eval_result]
- def call_script(
- javascript_code: str | Var[str],
- callback: EventType | None = None,
- ) -> EventSpec:
- """Create an event handler that executes arbitrary javascript code.
- Args:
- javascript_code: The code to execute.
- callback: EventHandler that will receive the result of evaluating the javascript code.
- Returns:
- EventSpec: An event that will execute the client side javascript.
- """
- callback_kwargs = {}
- if callback is not None:
- callback_kwargs = {
- "callback": format.format_queue_events(
- callback,
- args_spec=lambda result: [result],
- )._js_expr,
- }
- if isinstance(javascript_code, str):
- # When there is VarData, include it and eval the JS code inline on the client.
- javascript_code, original_code = (
- LiteralVar.create(javascript_code),
- javascript_code,
- )
- if not javascript_code._get_all_var_data():
- # Without VarData, cast to string and eval the code in the event loop.
- javascript_code = str(Var(_js_expr=original_code))
- return server_side(
- "_call_script",
- get_fn_signature(call_script),
- javascript_code=javascript_code,
- **callback_kwargs,
- )
- def call_function(
- javascript_code: str | Var,
- callback: EventType | None = None,
- ) -> EventSpec:
- """Create an event handler that executes arbitrary javascript code.
- Args:
- javascript_code: The code to execute.
- callback: EventHandler that will receive the result of evaluating the javascript code.
- Returns:
- EventSpec: An event that will execute the client side javascript.
- """
- callback_kwargs = {}
- if callback is not None:
- callback_kwargs = {
- "callback": format.format_queue_events(
- callback,
- args_spec=lambda result: [result],
- ),
- }
- javascript_code = (
- Var(javascript_code) if isinstance(javascript_code, str) else javascript_code
- )
- return server_side(
- "_call_function",
- get_fn_signature(call_function),
- function=javascript_code,
- **callback_kwargs,
- )
- def run_script(
- javascript_code: str | Var,
- callback: EventType | None = None,
- ) -> EventSpec:
- """Create an event handler that executes arbitrary javascript code.
- Args:
- javascript_code: The code to execute.
- callback: EventHandler that will receive the result of evaluating the javascript code.
- Returns:
- EventSpec: An event that will execute the client side javascript.
- """
- javascript_code = (
- Var(javascript_code) if isinstance(javascript_code, str) else javascript_code
- )
- return call_function(ArgsFunctionOperation.create((), javascript_code), callback)
- def get_event(state, event):
- """Get the event from the given state.
- Args:
- state: The state.
- event: The event.
- Returns:
- The event.
- """
- return f"{state.get_name()}.{event}"
- def get_hydrate_event(state) -> str:
- """Get the name of the hydrate event for the state.
- Args:
- state: The state.
- Returns:
- The name of the hydrate event.
- """
- return get_event(state, constants.CompileVars.HYDRATE)
- def call_event_handler(
- event_callback: EventHandler | EventSpec,
- event_spec: ArgsSpec | Sequence[ArgsSpec],
- key: Optional[str] = None,
- ) -> EventSpec:
- """Call an event handler to get the event spec.
- This function will inspect the function signature of the event handler.
- If it takes in an arg, the arg will be passed to the event handler.
- Otherwise, the event handler will be called with no args.
- Args:
- event_callback: The event handler.
- event_spec: The lambda that define the argument(s) to pass to the event handler.
- key: The key to pass to the event handler.
- Returns:
- The event spec from calling the event handler.
- # noqa: DAR401 failure
- """
- event_spec_args = parse_args_spec(event_spec) # type: ignore
- if isinstance(event_callback, EventSpec):
- check_fn_match_arg_spec(
- event_callback.handler.fn,
- event_spec,
- key,
- bool(event_callback.handler.state_full_name) + len(event_callback.args),
- event_callback.handler.fn.__qualname__,
- )
- # Handle partial application of EventSpec args
- return event_callback.add_args(*event_spec_args)
- check_fn_match_arg_spec(
- event_callback.fn,
- event_spec,
- key,
- bool(event_callback.state_full_name),
- event_callback.fn.__qualname__,
- )
- all_acceptable_specs = (
- [event_spec] if not isinstance(event_spec, Sequence) else event_spec
- )
- event_spec_return_types = list(
- filter(
- lambda event_spec_return_type: event_spec_return_type is not None
- and get_origin(event_spec_return_type) is tuple,
- (
- get_type_hints(arg_spec).get("return", None)
- for arg_spec in all_acceptable_specs
- ),
- )
- )
- if event_spec_return_types:
- failures = []
- event_callback_spec = inspect.getfullargspec(event_callback.fn)
- for event_spec_index, event_spec_return_type in enumerate(
- event_spec_return_types
- ):
- args = get_args(event_spec_return_type)
- args_types_without_vars = [
- arg if get_origin(arg) is not Var else get_args(arg)[0] for arg in args
- ]
- try:
- type_hints_of_provided_callback = get_type_hints(event_callback.fn)
- except NameError:
- type_hints_of_provided_callback = {}
- failed_type_check = False
- # check that args of event handler are matching the spec if type hints are provided
- for i, arg in enumerate(event_callback_spec.args[1:]):
- if arg not in type_hints_of_provided_callback:
- continue
- try:
- compare_result = typehint_issubclass(
- args_types_without_vars[i], type_hints_of_provided_callback[arg]
- )
- except TypeError:
- # TODO: In 0.7.0, remove this block and raise the exception
- # raise TypeError(
- # f"Could not compare types {args_types_without_vars[i]} and {type_hints_of_provided_callback[arg]} for argument {arg} of {event_handler.fn.__qualname__} provided for {key}." # noqa: ERA001
- # ) from e
- console.warn(
- f"Could not compare types {args_types_without_vars[i]} and {type_hints_of_provided_callback[arg]} for argument {arg} of {event_callback.fn.__qualname__} provided for {key}."
- )
- compare_result = False
- if compare_result:
- continue
- else:
- failure = EventHandlerArgTypeMismatch(
- f"Event handler {key} expects {args_types_without_vars[i]} for argument {arg} but got {type_hints_of_provided_callback[arg]} as annotated in {event_callback.fn.__qualname__} instead."
- )
- failures.append(failure)
- failed_type_check = True
- break
- if not failed_type_check:
- if event_spec_index:
- args = get_args(event_spec_return_types[0])
- args_types_without_vars = [
- arg if get_origin(arg) is not Var else get_args(arg)[0]
- for arg in args
- ]
- expect_string = ", ".join(
- repr(arg) for arg in args_types_without_vars
- ).replace("[", "\\[")
- given_string = ", ".join(
- repr(type_hints_of_provided_callback.get(arg, Any))
- for arg in event_callback_spec.args[1:]
- ).replace("[", "\\[")
- console.warn(
- f"Event handler {key} expects ({expect_string}) -> () but got ({given_string}) -> () as annotated in {event_callback.fn.__qualname__} instead. "
- f"This may lead to unexpected behavior but is intentionally ignored for {key}."
- )
- return event_callback(*event_spec_args)
- if failures:
- console.deprecate(
- "Mismatched event handler argument types",
- "\n".join([str(f) for f in failures]),
- "0.6.5",
- "0.7.0",
- )
- return event_callback(*event_spec_args) # type: ignore
- def unwrap_var_annotation(annotation: GenericType):
- """Unwrap a Var annotation or return it as is if it's not Var[X].
- Args:
- annotation: The annotation to unwrap.
- Returns:
- The unwrapped annotation.
- """
- if get_origin(annotation) is Var and (args := get_args(annotation)):
- return args[0]
- return annotation
- def resolve_annotation(annotations: dict[str, Any], arg_name: str):
- """Resolve the annotation for the given argument name.
- Args:
- annotations: The annotations.
- arg_name: The argument name.
- Returns:
- The resolved annotation.
- """
- annotation = annotations.get(arg_name)
- if annotation is None:
- console.deprecate(
- feature_name="Unannotated event handler arguments",
- reason="Provide type annotations for event handler arguments.",
- deprecation_version="0.6.3",
- removal_version="0.7.0",
- )
- # Allow arbitrary attribute access two levels deep until removed.
- return Dict[str, dict]
- return annotation
- def parse_args_spec(arg_spec: ArgsSpec | Sequence[ArgsSpec]):
- """Parse the args provided in the ArgsSpec of an event trigger.
- Args:
- arg_spec: The spec of the args.
- Returns:
- The parsed args.
- """
- # if there's multiple, the first is the default
- arg_spec = arg_spec[0] if isinstance(arg_spec, Sequence) else arg_spec
- spec = inspect.getfullargspec(arg_spec)
- annotations = get_type_hints(arg_spec)
- return list(
- arg_spec(
- *[
- Var(f"_{l_arg}").to(
- unwrap_var_annotation(resolve_annotation(annotations, l_arg))
- )
- for l_arg in spec.args
- ]
- )
- )
- def check_fn_match_arg_spec(
- user_func: Callable,
- arg_spec: ArgsSpec | Sequence[ArgsSpec],
- key: str | None = None,
- number_of_bound_args: int = 0,
- func_name: str | None = None,
- ):
- """Ensures that the function signature matches the passed argument specification
- or raises an EventFnArgMismatch if they do not.
- Args:
- user_func: The function to be validated.
- arg_spec: The argument specification for the event trigger.
- key: The key of the event trigger.
- number_of_bound_args: The number of bound arguments to the function.
- func_name: The name of the function to be validated.
- Raises:
- EventFnArgMismatch: Raised if the number of mandatory arguments do not match
- """
- user_args = inspect.getfullargspec(user_func).args
- # Drop the first argument if it's a bound method
- if inspect.ismethod(user_func) and user_func.__self__ is not None:
- user_args = user_args[1:]
- user_default_args = inspect.getfullargspec(user_func).defaults
- number_of_user_args = len(user_args) - number_of_bound_args
- number_of_user_default_args = len(user_default_args) if user_default_args else 0
- parsed_event_args = parse_args_spec(arg_spec)
- number_of_event_args = len(parsed_event_args)
- if number_of_user_args - number_of_user_default_args > number_of_event_args:
- raise EventFnArgMismatch(
- f"Event {key} only provides {number_of_event_args} arguments, but "
- f"{func_name or user_func} requires at least {number_of_user_args - number_of_user_default_args} "
- "arguments to be passed to the event handler.\n"
- "See https://reflex.dev/docs/events/event-arguments/"
- )
- def call_event_fn(
- fn: Callable,
- arg_spec: ArgsSpec | Sequence[ArgsSpec],
- key: Optional[str] = None,
- ) -> list[EventSpec] | Var:
- """Call a function to a list of event specs.
- The function should return a single EventSpec, a list of EventSpecs, or a
- single Var.
- Args:
- fn: The function to call.
- arg_spec: The argument spec for the event trigger.
- key: The key to pass to the event handler.
- Returns:
- The event specs from calling the function or a Var.
- Raises:
- EventHandlerValueError: If the lambda returns an unusable value.
- """
- # Import here to avoid circular imports.
- from reflex.event import EventHandler, EventSpec
- from reflex.utils.exceptions import EventHandlerValueError
- # Check that fn signature matches arg_spec
- check_fn_match_arg_spec(fn, arg_spec, key=key)
- parsed_args = parse_args_spec(arg_spec)
- number_of_fn_args = len(inspect.getfullargspec(fn).args)
- # Call the function with the parsed args.
- out = fn(*[*parsed_args][:number_of_fn_args])
- # If the function returns a Var, assume it's an EventChain and render it directly.
- if isinstance(out, Var):
- return out
- # Convert the output to a list.
- if not isinstance(out, list):
- out = [out]
- # Convert any event specs to event specs.
- events = []
- for e in out:
- if isinstance(e, EventHandler):
- # An un-called EventHandler gets all of the args of the event trigger.
- e = call_event_handler(e, arg_spec, key=key)
- # Make sure the event spec is valid.
- if not isinstance(e, EventSpec):
- raise EventHandlerValueError(
- f"Lambda {fn} returned an invalid event spec: {e}."
- )
- # Add the event spec to the chain.
- events.append(e)
- # Return the events.
- return events
- def get_handler_args(
- event_spec: EventSpec,
- ) -> tuple[tuple[Var, Var], ...]:
- """Get the handler args for the given event spec.
- Args:
- event_spec: The event spec.
- Returns:
- The handler args.
- """
- args = inspect.getfullargspec(event_spec.handler.fn).args
- return event_spec.args if len(args) > 1 else ()
- def fix_events(
- events: list[EventHandler | EventSpec] | None,
- token: str,
- router_data: dict[str, Any] | None = None,
- ) -> list[Event]:
- """Fix a list of events returned by an event handler.
- Args:
- events: The events to fix.
- token: The user token.
- router_data: The optional router data to set in the event.
- Raises:
- ValueError: If the event type is not what was expected.
- Returns:
- The fixed events.
- """
- # If the event handler returns nothing, return an empty list.
- if events is None:
- return []
- # If the handler returns a single event, wrap it in a list.
- if not isinstance(events, List):
- events = [events]
- # Fix the events created by the handler.
- out = []
- for e in events:
- if isinstance(e, Event):
- # If the event is already an event, append it to the list.
- out.append(e)
- continue
- if not isinstance(e, (EventHandler, EventSpec)):
- e = EventHandler(fn=e)
- # Otherwise, create an event from the event spec.
- if isinstance(e, EventHandler):
- e = e()
- if not isinstance(e, EventSpec):
- raise ValueError(f"Unexpected event type, {type(e)}.")
- name = format.format_event_handler(e.handler)
- payload = {k._js_expr: v._decode() for k, v in e.args} # type: ignore
- # Filter router_data to reduce payload size
- event_router_data = {
- k: v
- for k, v in (router_data or {}).items()
- if k in constants.route.ROUTER_DATA_INCLUDE
- }
- # Create an event and append it to the list.
- out.append(
- Event(
- token=token,
- name=name,
- payload=payload,
- router_data=event_router_data,
- )
- )
- return out
- def get_fn_signature(fn: Callable) -> inspect.Signature:
- """Get the signature of a function.
- Args:
- fn: The function.
- Returns:
- The signature of the function.
- """
- signature = inspect.signature(fn)
- new_param = inspect.Parameter(
- FRONTEND_EVENT_STATE, inspect.Parameter.POSITIONAL_OR_KEYWORD, annotation=Any
- )
- return signature.replace(parameters=(new_param, *signature.parameters.values()))
- class EventVar(ObjectVar, python_types=EventSpec):
- """Base class for event vars."""
- @dataclasses.dataclass(
- eq=False,
- frozen=True,
- **{"slots": True} if sys.version_info >= (3, 10) else {},
- )
- class LiteralEventVar(VarOperationCall, LiteralVar, EventVar):
- """A literal event var."""
- _var_value: EventSpec = dataclasses.field(default=None) # type: ignore
- 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: EventSpec,
- _var_data: VarData | None = None,
- ) -> LiteralEventVar:
- """Create a new LiteralEventVar instance.
- Args:
- value: The value of the var.
- _var_data: The data of the var.
- Returns:
- The created LiteralEventVar instance.
- """
- return cls(
- _js_expr="",
- _var_type=EventSpec,
- _var_data=_var_data,
- _var_value=value,
- _func=FunctionStringVar("Event"),
- _args=(
- # event handler name
- ".".join(
- filter(
- None,
- format.get_event_handler_parts(value.handler),
- )
- ),
- # event handler args
- {str(name): value for name, value in value.args},
- # event actions
- value.event_actions,
- # client handler name
- *([value.client_handler_name] if value.client_handler_name else []),
- ),
- )
- class EventChainVar(BuilderFunctionVar, python_types=EventChain):
- """Base class for event chain vars."""
- @dataclasses.dataclass(
- eq=False,
- frozen=True,
- **{"slots": True} if sys.version_info >= (3, 10) else {},
- )
- # Note: LiteralVar is second in the inheritance list allowing it act like a
- # CachedVarOperation (ArgsFunctionOperation) and get the _js_expr from the
- # _cached_var_name property.
- class LiteralEventChainVar(ArgsFunctionOperationBuilder, LiteralVar, EventChainVar):
- """A literal event chain var."""
- _var_value: EventChain = dataclasses.field(default=None) # type: ignore
- 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: EventChain,
- _var_data: VarData | None = None,
- ) -> LiteralEventChainVar:
- """Create a new LiteralEventChainVar instance.
- Args:
- value: The value of the var.
- _var_data: The data of the var.
- Returns:
- The created LiteralEventChainVar instance.
- """
- arg_spec = (
- value.args_spec[0]
- if isinstance(value.args_spec, Sequence)
- else value.args_spec
- )
- sig = inspect.signature(arg_spec) # type: ignore
- if sig.parameters:
- arg_def = tuple((f"_{p}" for p in sig.parameters))
- arg_def_expr = LiteralVar.create([Var(_js_expr=arg) for arg in arg_def])
- else:
- # add a default argument for addEvents if none were specified in value.args_spec
- # used to trigger the preventDefault() on the event.
- arg_def = ("...args",)
- arg_def_expr = Var(_js_expr="args")
- if value.invocation is None:
- invocation = FunctionStringVar.create("addEvents")
- else:
- invocation = value.invocation
- return cls(
- _js_expr="",
- _var_type=EventChain,
- _var_data=_var_data,
- _args=FunctionArgs(arg_def),
- _return_expr=invocation.call(
- LiteralVar.create([LiteralVar.create(event) for event in value.events]),
- arg_def_expr,
- value.event_actions,
- ),
- _var_value=value,
- )
- P = ParamSpec("P")
- Q = ParamSpec("Q")
- T = TypeVar("T")
- V = TypeVar("V")
- V2 = TypeVar("V2")
- V3 = TypeVar("V3")
- V4 = TypeVar("V4")
- V5 = TypeVar("V5")
- background_event_decorator = background
- class EventCallback(Generic[P, T]):
- """A descriptor that wraps a function to be used as an event."""
- def __init__(self, func: Callable[Concatenate[Any, P], T]):
- """Initialize the descriptor with the function to be wrapped.
- Args:
- func: The function to be wrapped.
- """
- self.func = func
- @property
- def prevent_default(self):
- """Prevent default behavior.
- Returns:
- The event callback with prevent default behavior.
- """
- return self
- @property
- def stop_propagation(self):
- """Stop event propagation.
- Returns:
- The event callback with stop propagation behavior.
- """
- return self
- @overload
- def __call__(
- self: EventCallback[Q, T],
- ) -> EventCallback[Q, T]: ...
- @overload
- def __call__(
- self: EventCallback[Concatenate[V, Q], T], value: V | Var[V]
- ) -> EventCallback[Q, T]: ...
- @overload
- def __call__(
- self: EventCallback[Concatenate[V, V2, Q], T],
- value: V | Var[V],
- value2: V2 | Var[V2],
- ) -> EventCallback[Q, T]: ...
- @overload
- def __call__(
- self: EventCallback[Concatenate[V, V2, V3, Q], T],
- value: V | Var[V],
- value2: V2 | Var[V2],
- value3: V3 | Var[V3],
- ) -> EventCallback[Q, T]: ...
- @overload
- def __call__(
- self: EventCallback[Concatenate[V, V2, V3, V4, Q], T],
- value: V | Var[V],
- value2: V2 | Var[V2],
- value3: V3 | Var[V3],
- value4: V4 | Var[V4],
- ) -> EventCallback[Q, T]: ...
- def __call__(self, *values) -> EventCallback: # type: ignore
- """Call the function with the values.
- Args:
- *values: The values to call the function with.
- Returns:
- The function with the values.
- """
- return self.func(*values) # type: ignore
- @overload
- def __get__(
- self: EventCallback[P, T], instance: None, owner
- ) -> EventCallback[P, T]: ...
- @overload
- def __get__(self, instance, owner) -> Callable[P, T]: ...
- def __get__(self, instance, owner) -> Callable: # type: ignore
- """Get the function with the instance bound to it.
- Args:
- instance: The instance to bind to the function.
- owner: The owner of the function.
- Returns:
- The function with the instance bound to it
- """
- if instance is None:
- return self.func # type: ignore
- return partial(self.func, instance) # type: ignore
- G = ParamSpec("G")
- if TYPE_CHECKING:
- from reflex.state import BaseState
- BASE_STATE = TypeVar("BASE_STATE", bound=BaseState)
- else:
- BASE_STATE = TypeVar("BASE_STATE")
- StateCallable = TypeAliasType(
- "StateCallable",
- Callable[Concatenate[BASE_STATE, G], Any],
- type_params=(G, BASE_STATE),
- )
- IndividualEventType = Union[
- EventSpec,
- EventHandler,
- Callable[G, Any],
- StateCallable[G, BASE_STATE],
- EventCallback[G, Any],
- Var[Any],
- ]
- ItemOrList = Union[V, List[V]]
- EventType = ItemOrList[IndividualEventType[G, BASE_STATE]]
- class EventNamespace(types.SimpleNamespace):
- """A namespace for event related classes."""
- Event = Event
- EventHandler = EventHandler
- EventSpec = EventSpec
- CallableEventSpec = CallableEventSpec
- EventChain = EventChain
- EventVar = EventVar
- LiteralEventVar = LiteralEventVar
- EventChainVar = EventChainVar
- LiteralEventChainVar = LiteralEventChainVar
- EventType = EventType
- EventCallback = EventCallback
- @overload
- @staticmethod
- def __call__(
- func: None = None, *, background: bool | None = None
- ) -> Callable[[Callable[Concatenate[BASE_STATE, P], T]], EventCallback[P, T]]: ...
- @overload
- @staticmethod
- def __call__(
- func: Callable[Concatenate[BASE_STATE, P], T],
- *,
- background: bool | None = None,
- ) -> EventCallback[P, T]: ...
- @staticmethod
- def __call__(
- func: Callable[Concatenate[BASE_STATE, P], T] | None = None,
- *,
- background: bool | None = None,
- ) -> Union[
- EventCallback[P, T],
- Callable[[Callable[Concatenate[BASE_STATE, P], T]], EventCallback[P, T]],
- ]:
- """Wrap a function to be used as an event.
- Args:
- func: The function to wrap.
- background: Whether the event should be run in the background. Defaults to False.
- Returns:
- The wrapped function.
- """
- def wrapper(
- func: Callable[Concatenate[BASE_STATE, P], T],
- ) -> EventCallback[P, T]:
- if background is True:
- return background_event_decorator(func, __internal_reflex_call=True) # type: ignore
- return func # type: ignore
- if func is not None:
- return wrapper(func)
- return wrapper
- get_event = staticmethod(get_event)
- get_hydrate_event = staticmethod(get_hydrate_event)
- fix_events = staticmethod(fix_events)
- call_event_handler = staticmethod(call_event_handler)
- call_event_fn = staticmethod(call_event_fn)
- get_handler_args = staticmethod(get_handler_args)
- check_fn_match_arg_spec = staticmethod(check_fn_match_arg_spec)
- resolve_annotation = staticmethod(resolve_annotation)
- parse_args_spec = staticmethod(parse_args_spec)
- passthrough_event_spec = staticmethod(passthrough_event_spec)
- input_event = staticmethod(input_event)
- key_event = staticmethod(key_event)
- no_args_event_spec = staticmethod(no_args_event_spec)
- server_side = staticmethod(server_side)
- redirect = staticmethod(redirect)
- console_log = staticmethod(console_log)
- noop = staticmethod(noop)
- back = staticmethod(back)
- window_alert = staticmethod(window_alert)
- set_focus = staticmethod(set_focus)
- scroll_to = staticmethod(scroll_to)
- set_value = staticmethod(set_value)
- remove_cookie = staticmethod(remove_cookie)
- clear_local_storage = staticmethod(clear_local_storage)
- remove_local_storage = staticmethod(remove_local_storage)
- clear_session_storage = staticmethod(clear_session_storage)
- remove_session_storage = staticmethod(remove_session_storage)
- set_clipboard = staticmethod(set_clipboard)
- download = staticmethod(download)
- call_script = staticmethod(call_script)
- call_function = staticmethod(call_function)
- run_script = staticmethod(run_script)
- event = EventNamespace()
|