reflex_copy/pynecone/event.py

"""Define event classes to connect the frontend and backend."""
from __future__ import annotations

import inspect
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union

from pynecone import constants
from pynecone.base import Base
from pynecone.utils import format
from pynecone.var import BaseVar, Var


class Event(Base):
    """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] = {}

    # The event payload.
    payload: Dict[str, Any] = {}


class EventHandler(Base):
    """An event handler responds to an event to update the state."""

    # The function to call in response to the event.
    fn: Callable

    class Config:
        """The Pydantic config."""

        # Needed to allow serialization of Callable.
        frozen = True

    def __call__(self, *args: Var) -> 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:
            TypeError: If the arguments are invalid.
        """
        # Get the function args.
        fn_args = inspect.getfullargspec(self.fn).args[1:]

        # Construct the payload.
        values = []
        for arg in args:
            # If it is a Var, add the full name.
            if isinstance(arg, Var):
                values.append(arg.full_name)
                continue

            if isinstance(arg, FileUpload):
                return EventSpec(handler=self, upload=True)

            # Otherwise, convert to JSON.
            try:
                values.append(format.json_dumps(arg))
            except TypeError as e:
                raise TypeError(
                    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)


class EventSpec(Base):
    """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

    # The local arguments on the frontend.
    local_args: Tuple[str, ...] = ()

    # The arguments to pass to the function.
    args: Tuple[Any, ...] = ()

    # Whether to upload files.
    upload: bool = False

    class Config:
        """The Pydantic config."""

        # Required to allow tuple fields.
        frozen = True


class EventChain(Base):
    """Container for a chain of events that will be executed in order."""

    events: List[EventSpec]

    # Whether events are in fully controlled input.
    full_control: bool = False

    # State name when fully controlled.
    state_name: str = ""


class Target(Base):
    """A Javascript event target."""

    checked: bool = False
    value: Any = None


class FrontendEvent(Base):
    """A Javascript event."""

    target: Target = Target()
    key: str = ""


# The default event argument.
EVENT_ARG = BaseVar(name="_e", type_=FrontendEvent, is_local=True)


class FileUpload(Base):
    """Class to represent a file upload."""

    pass


# Special server-side events.
def redirect(path: str) -> EventSpec:
    """Redirect to a new path.

    Args:
        path: The path to redirect to.

    Returns:
        An event to redirect to the path.
    """

    def fn():
        return None

    fn.__qualname__ = "_redirect"
    return EventSpec(
        handler=EventHandler(fn=fn),
        args=(("path", path),),
    )


def console_log(message: str) -> EventSpec:
    """Do a console.log on the browser.

    Args:
        message: The message to log.

    Returns:
        An event to log the message.
    """

    def fn():
        return None

    fn.__qualname__ = "_console"
    return EventSpec(
        handler=EventHandler(fn=fn),
        args=(("message", message),),
    )


def window_alert(message: str) -> EventSpec:
    """Create a window alert on the browser.

    Args:
        message: The message to alert.

    Returns:
        An event to alert the message.
    """

    def fn():
        return None

    fn.__qualname__ = "_alert"
    return EventSpec(
        handler=EventHandler(fn=fn),
        args=(("message", message),),
    )


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.HYDRATE)


def call_event_handler(event_handler: EventHandler, arg: Var) -> 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_handler: The event handler.
        arg: The argument to pass to the event handler.

    Returns:
        The event spec from calling the event handler.
    """
    args = inspect.getfullargspec(event_handler.fn).args
    if len(args) == 1:
        return event_handler()
    assert (
        len(args) == 2
    ), f"Event handler {event_handler.fn} must have 1 or 2 arguments."
    return event_handler(arg)


def call_event_fn(fn: Callable, arg: Var) -> List[EventSpec]:
    """Call a function to a list of event specs.

    The function should return either a single EventSpec or a list of EventSpecs.
    If the function takes in an arg, the arg will be passed to the function.
    Otherwise, the function will be called with no args.

    Args:
        fn: The function to call.
        arg: The argument to pass to the function.

    Returns:
        The event specs from calling the function.

    Raises:
        ValueError: If the lambda has an invalid signature.
    """
    # Import here to avoid circular imports.
    from pynecone.event import EventHandler, EventSpec

    # Get the args of the lambda.
    args = inspect.getfullargspec(fn).args

    # Call the lambda.
    if len(args) == 0:
        out = fn()
    elif len(args) == 1:
        out = fn(arg)
    else:
        raise ValueError(f"Lambda {fn} must have 0 or 1 arguments.")

    # 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:
        # Convert handlers to event specs.
        if isinstance(e, EventHandler):
            if len(args) == 0:
                e = e()
            elif len(args) == 1:
                e = e(arg)

        # Make sure the event spec is valid.
        if not isinstance(e, EventSpec):
            raise ValueError(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, arg: Var) -> Tuple[Tuple[str, str], ...]:
    """Get the handler args for the given event spec.

    Args:
        event_spec: The event spec.
        arg: The controlled event argument.

    Returns:
        The handler args.

    Raises:
        ValueError: If the event handler has an invalid signature.
    """
    args = inspect.getfullargspec(event_spec.handler.fn).args
    if len(args) < 2:
        raise ValueError(
            f"Event handler has an invalid signature, needed a method with a parameter, got {event_spec.handler}."
        )
    return event_spec.args if len(args) > 2 else ((args[1], arg.name),)


def fix_events(
    events: Optional[List[Union[EventHandler, EventSpec]]], token: str
) -> List[Event]:
    """Fix a list of events returned by an event handler.

    Args:
        events: The events to fix.
        token: The user token.

    Returns:
        The fixed events.
    """
    from pynecone.event import Event, EventHandler, EventSpec

    # 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 not isinstance(e, (EventHandler, EventSpec)):
            e = EventHandler(fn=e)
        # Otherwise, create an event from the event spec.
        if isinstance(e, EventHandler):
            e = e()
        assert isinstance(e, EventSpec), f"Unexpected event type, {type(e)}."
        name = format.format_event_handler(e.handler)
        payload = dict(e.args)

        # Create an event and append it to the list.
        out.append(
            Event(
                token=token,
                name=name,
                payload=payload,
            )
        )

    return out


# A set of common event triggers.
EVENT_TRIGGERS: Set[str] = {
    "on_focus",
    "on_blur",
    "on_click",
    "on_context_menu",
    "on_double_click",
    "on_mouse_down",
    "on_mouse_enter",
    "on_mouse_leave",
    "on_mouse_move",
    "on_mouse_out",
    "on_mouse_over",
    "on_mouse_up",
    "on_scroll",
}