313 lines
7.5 KiB
Python
313 lines
7.5 KiB
Python
"""Serializers used to convert Var types to JSON strings."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import types as builtin_types
|
|
from datetime import date, datetime, time, timedelta
|
|
from typing import Any, Callable, Dict, List, Set, Tuple, Type, Union, get_type_hints
|
|
|
|
from reflex.base import Base
|
|
from reflex.utils import exceptions, format, types
|
|
|
|
# Mapping from type to a serializer.
|
|
# The serializer should convert the type to a JSON object.
|
|
SerializedType = Union[str, bool, int, float, list, dict]
|
|
Serializer = Callable[[Type], SerializedType]
|
|
SERIALIZERS: dict[Type, Serializer] = {}
|
|
|
|
|
|
def serializer(fn: Serializer) -> Serializer:
|
|
"""Decorator to add a serializer for a given type.
|
|
|
|
Args:
|
|
fn: The function to decorate.
|
|
|
|
Returns:
|
|
The decorated function.
|
|
|
|
Raises:
|
|
ValueError: If the function does not take a single argument.
|
|
"""
|
|
# Get the global serializers.
|
|
global SERIALIZERS
|
|
|
|
# Check the type hints to get the type of the argument.
|
|
type_hints = get_type_hints(fn)
|
|
args = [arg for arg in type_hints if arg != "return"]
|
|
|
|
# Make sure the function takes a single argument.
|
|
if len(args) != 1:
|
|
raise ValueError("Serializer must take a single argument.")
|
|
|
|
# Get the type of the argument.
|
|
type_ = type_hints[args[0]]
|
|
|
|
# Make sure the type is not already registered.
|
|
registered_fn = SERIALIZERS.get(type_)
|
|
if registered_fn is not None and registered_fn != fn:
|
|
raise ValueError(
|
|
f"Serializer for type {type_} is already registered as {registered_fn.__qualname__}."
|
|
)
|
|
|
|
# Register the serializer.
|
|
SERIALIZERS[type_] = fn
|
|
|
|
# Return the function.
|
|
return fn
|
|
|
|
|
|
def serialize(value: Any) -> SerializedType | None:
|
|
"""Serialize the value to a JSON string.
|
|
|
|
Args:
|
|
value: The value to serialize.
|
|
|
|
Returns:
|
|
The serialized value, or None if a serializer is not found.
|
|
"""
|
|
# Get the serializer for the type.
|
|
serializer = get_serializer(type(value))
|
|
|
|
# If there is no serializer, return None.
|
|
if serializer is None:
|
|
return None
|
|
|
|
# Serialize the value.
|
|
return serializer(value)
|
|
|
|
|
|
def get_serializer(type_: Type) -> Serializer | None:
|
|
"""Get the serializer for the type.
|
|
|
|
Args:
|
|
type_: The type to get the serializer for.
|
|
|
|
Returns:
|
|
The serializer for the type, or None if there is no serializer.
|
|
"""
|
|
global SERIALIZERS
|
|
|
|
# First, check if the type is registered.
|
|
serializer = SERIALIZERS.get(type_)
|
|
if serializer is not None:
|
|
return serializer
|
|
|
|
# If the type is not registered, check if it is a subclass of a registered type.
|
|
for registered_type, serializer in reversed(SERIALIZERS.items()):
|
|
if types._issubclass(type_, registered_type):
|
|
return serializer
|
|
|
|
# If there is no serializer, return None.
|
|
return None
|
|
|
|
|
|
def has_serializer(type_: Type) -> bool:
|
|
"""Check if there is a serializer for the type.
|
|
|
|
Args:
|
|
type_: The type to check.
|
|
|
|
Returns:
|
|
Whether there is a serializer for the type.
|
|
"""
|
|
return get_serializer(type_) is not None
|
|
|
|
|
|
@serializer
|
|
def serialize_type(value: type) -> str:
|
|
"""Serialize a python type.
|
|
|
|
Args:
|
|
value: the type to serialize.
|
|
|
|
Returns:
|
|
The serialized type.
|
|
"""
|
|
return value.__name__
|
|
|
|
|
|
@serializer
|
|
def serialize_str(value: str) -> str:
|
|
"""Serialize a string.
|
|
|
|
Args:
|
|
value: The string to serialize.
|
|
|
|
Returns:
|
|
The serialized string.
|
|
"""
|
|
return value
|
|
|
|
|
|
@serializer
|
|
def serialize_primitive(value: Union[bool, int, float, None]) -> str:
|
|
"""Serialize a primitive type.
|
|
|
|
Args:
|
|
value: The number/bool/None to serialize.
|
|
|
|
Returns:
|
|
The serialized number/bool/None.
|
|
"""
|
|
return format.json_dumps(value)
|
|
|
|
|
|
@serializer
|
|
def serialize_base(value: Base) -> str:
|
|
"""Serialize a Base instance.
|
|
|
|
Args:
|
|
value : The Base to serialize.
|
|
|
|
Returns:
|
|
The serialized Base.
|
|
"""
|
|
return value.json()
|
|
|
|
|
|
@serializer
|
|
def serialize_list(value: Union[List, Tuple, Set]) -> str:
|
|
"""Serialize a list to a JSON string.
|
|
|
|
Args:
|
|
value: The list to serialize.
|
|
|
|
Returns:
|
|
The serialized list.
|
|
"""
|
|
# Dump the list to a string.
|
|
fprop = format.json_dumps(list(value))
|
|
|
|
# Unwrap var values.
|
|
return format.unwrap_vars(fprop)
|
|
|
|
|
|
@serializer
|
|
def serialize_dict(prop: Dict[str, Any]) -> str:
|
|
"""Serialize a dictionary to a JSON string.
|
|
|
|
Args:
|
|
prop: The dictionary to serialize.
|
|
|
|
Returns:
|
|
The serialized dictionary.
|
|
|
|
Raises:
|
|
InvalidStylePropError: If the style prop is invalid.
|
|
"""
|
|
# Import here to avoid circular imports.
|
|
from reflex.event import EventHandler
|
|
|
|
prop_dict = {}
|
|
|
|
for key, value in prop.items():
|
|
if types._issubclass(type(value), Callable):
|
|
raise exceptions.InvalidStylePropError(
|
|
f"The style prop `{format.to_snake_case(key)}` cannot have " # type: ignore
|
|
f"`{value.fn.__qualname__ if isinstance(value, EventHandler) else value.__qualname__ if isinstance(value, builtin_types.FunctionType) else value}`, "
|
|
f"an event handler or callable as its value"
|
|
)
|
|
prop_dict[key] = value
|
|
|
|
# Dump the dict to a string.
|
|
fprop = format.json_dumps(prop_dict)
|
|
|
|
# Unwrap var values.
|
|
return format.unwrap_vars(fprop)
|
|
|
|
|
|
@serializer
|
|
def serialize_datetime(dt: Union[date, datetime, time, timedelta]) -> str:
|
|
"""Serialize a datetime to a JSON string.
|
|
|
|
Args:
|
|
dt: The datetime to serialize.
|
|
|
|
Returns:
|
|
The serialized datetime.
|
|
"""
|
|
return str(dt)
|
|
|
|
|
|
try:
|
|
from pandas import DataFrame
|
|
|
|
def format_dataframe_values(df: DataFrame) -> List[List[Any]]:
|
|
"""Format dataframe values to a list of lists.
|
|
|
|
Args:
|
|
df: The dataframe to format.
|
|
|
|
Returns:
|
|
The dataframe as a list of lists.
|
|
"""
|
|
return [
|
|
[str(d) if isinstance(d, (list, tuple)) else d for d in data]
|
|
for data in list(df.values.tolist())
|
|
]
|
|
|
|
@serializer
|
|
def serialize_dataframe(df: DataFrame) -> dict:
|
|
"""Serialize a pandas dataframe.
|
|
|
|
Args:
|
|
df: The dataframe to serialize.
|
|
|
|
Returns:
|
|
The serialized dataframe.
|
|
"""
|
|
return {
|
|
"columns": df.columns.tolist(),
|
|
"data": format_dataframe_values(df),
|
|
}
|
|
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
from plotly.graph_objects import Figure
|
|
from plotly.io import to_json
|
|
|
|
@serializer
|
|
def serialize_figure(figure: Figure) -> list:
|
|
"""Serialize a plotly figure.
|
|
|
|
Args:
|
|
figure: The figure to serialize.
|
|
|
|
Returns:
|
|
The serialized figure.
|
|
"""
|
|
return json.loads(str(to_json(figure)))["data"]
|
|
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
try:
|
|
import base64
|
|
import io
|
|
|
|
from PIL.Image import Image as Img
|
|
|
|
@serializer
|
|
def serialize_image(image: Img) -> str:
|
|
"""Serialize a plotly figure.
|
|
|
|
Args:
|
|
image: The image to serialize.
|
|
|
|
Returns:
|
|
The serialized image.
|
|
"""
|
|
buff = io.BytesIO()
|
|
image.save(buff, format=getattr(image, "format", None) or "PNG")
|
|
image_bytes = buff.getvalue()
|
|
base64_image = base64.b64encode(image_bytes).decode("utf-8")
|
|
mime_type = getattr(image, "get_format_mimetype", lambda: "image/png")()
|
|
return f"data:{mime_type};base64,{base64_image}"
|
|
|
|
except ImportError:
|
|
pass
|