diff --git a/reflex/vars/__init__.py b/reflex/vars/__init__.py index c81e9a9bff3..5b3e3cbbbf8 100644 --- a/reflex/vars/__init__.py +++ b/reflex/vars/__init__.py @@ -13,6 +13,7 @@ var_operation, var_operation_return, ) +from .blob import BlobVar, LiteralBlobVar from .color import ColorVar, LiteralColorVar from .datetime import DateTimeVar from .function import FunctionStringVar, FunctionVar, VarOperationCall @@ -20,8 +21,10 @@ from .object import LiteralObjectVar, ObjectVar from .sequence import ( ArrayVar, + BytesVar, ConcatVarOperation, LiteralArrayVar, + LiteralBytesVar, LiteralStringVar, StringVar, ) @@ -29,7 +32,9 @@ __all__ = [ "ArrayVar", "BaseStateMeta", + "BlobVar", "BooleanVar", + "BytesVar", "ColorVar", "ConcatVarOperation", "DateTimeVar", @@ -38,7 +43,9 @@ "FunctionStringVar", "FunctionVar", "LiteralArrayVar", + "LiteralBlobVar", "LiteralBooleanVar", + "LiteralBytesVar", "LiteralColorVar", "LiteralNumberVar", "LiteralObjectVar", diff --git a/reflex/vars/base.py b/reflex/vars/base.py index 412c0e41c70..12907b73218 100644 --- a/reflex/vars/base.py +++ b/reflex/vars/base.py @@ -76,10 +76,18 @@ from reflex.constants.colors import Color from reflex.state import BaseState + from .blob import Blob, BlobVar, LiteralBlobVar from .color import LiteralColorVar from .number import BooleanVar, LiteralBooleanVar, LiteralNumberVar, NumberVar from .object import LiteralObjectVar, ObjectVar - from .sequence import ArrayVar, LiteralArrayVar, LiteralStringVar, StringVar + from .sequence import ( + ArrayVar, + BytesVar, + LiteralArrayVar, + LiteralBytesVar, + LiteralStringVar, + StringVar, + ) VAR_TYPE = TypeVar("VAR_TYPE", covariant=True) @@ -655,6 +663,22 @@ def create( # pyright: ignore [reportOverlappingOverload] _var_data: VarData | None = None, ) -> LiteralStringVar: ... + @overload + @classmethod + def create( # pyright: ignore [reportOverlappingOverload] + cls, + value: bytes, + _var_data: VarData | None = None, + ) -> LiteralBytesVar: ... + + @overload + @classmethod + def create( # pyright: ignore [reportOverlappingOverload] + cls, + value: Blob, + _var_data: VarData | None = None, + ) -> LiteralBlobVar: ... + @overload @classmethod def create( # pyright: ignore [reportOverlappingOverload] @@ -845,6 +869,9 @@ def guess_type(self: Var[NoReturn]) -> Var[Any]: ... # pyright: ignore [reportO @overload def guess_type(self: Var[str]) -> StringVar: ... + @overload + def guess_type(self: Var[bytes | Sequence[bytes]]) -> BytesVar: ... + @overload def guess_type(self: Var[bool]) -> BooleanVar: ... @@ -1686,6 +1713,20 @@ def var_operation( ) -> Callable[P, StringVar]: ... +@overload +def var_operation( + func: Callable[P, CustomVarOperationReturn[bytes]] + | Callable[P, CustomVarOperationReturn[bytes | None]], +) -> Callable[P, BytesVar]: ... + + +@overload +def var_operation( + func: Callable[P, CustomVarOperationReturn[Blob]] + | Callable[P, CustomVarOperationReturn[Blob | None]], +) -> Callable[P, BlobVar]: ... + + LIST_T = TypeVar("LIST_T", bound=Sequence) @@ -2306,6 +2347,13 @@ def __get__( owner: type, ) -> StringVar: ... + @overload + def __get__( + self: ComputedVar[bytes], + instance: None, + owner: type, + ) -> BytesVar: ... + @overload def __get__( self: ComputedVar[MAPPING_TYPE], @@ -3402,6 +3450,11 @@ def __get__( self: Field[str] | Field[str | None], instance: None, owner: Any ) -> StringVar: ... + @overload + def __get__( + self: Field[bytes] | Field[bytes | None], instance: None, owner: Any + ) -> BytesVar: ... + @overload def __get__( self: Field[list[V]] diff --git a/reflex/vars/blob.py b/reflex/vars/blob.py new file mode 100644 index 00000000000..287d32c0c67 --- /dev/null +++ b/reflex/vars/blob.py @@ -0,0 +1,163 @@ +"""Blob variable types for representing JavaScript Blob objects in Reflex.""" + +import dataclasses +from typing import TYPE_CHECKING, TypeVar + +from reflex.vars.base import ( + LiteralVar, + Var, + VarData, + var_operation, + var_operation_return, +) + +if TYPE_CHECKING: + from reflex.vars import Var + + +@dataclasses.dataclass +class Blob: + """Represents a JavaScript Blob object.""" + + data: str | bytes = "" + mime_type: str = "" + + +BLOB_T = TypeVar("BLOB_T", bound=bytes | str, covariant=True) + + +class BlobVar(Var[BLOB_T], python_types=Blob): + """A variable representing a JavaScript Blob object.""" + + @classmethod + def _determine_mime_type(cls, value: str | bytes | Blob | Var) -> str: + mime_type = "" + if isinstance(value, str | bytes | Blob): + match value: + case str(): + mime_type = "text/plain" + case bytes(): + mime_type = "application/octet-stream" + case Blob(): + mime_type = value.mime_type + + elif isinstance(value, Var): + if isinstance(value._var_type, str): + mime_type = "text/plain" + if isinstance(value._var_type, bytes): + mime_type = "application/octet-stream" + + if not mime_type: + msg = "Unable to determine mime type for blob creation." + raise ValueError(msg) + + return mime_type + + @classmethod + def create( + cls, + value: str | bytes | Blob | Var, + mime_type: str | Var | None = None, + _var_data: VarData | None = None, + ): + """Create a BlobVar from the given value and MIME type. + + Args: + value: The data to create the Blob from (string, bytes, or Var). + mime_type: The MIME type of the Blob (string or Var). + _var_data: Optional variable data. + + Returns: + A BlobVar instance representing the JavaScript Blob object. + """ + if mime_type is None: + mime_type = cls._determine_mime_type(value) + + if not isinstance(mime_type, Var): + mime_type = LiteralVar.create(mime_type) + + if isinstance(value, str | bytes): + value = LiteralVar.create(value) + + elif isinstance(value, Blob): + value = LiteralVar.create(value.data) + + if isinstance(value._var_type, bytes): + value = f"new Uint8Array({value})" + + return cls( + _js_expr=f"new Blob([{value}], {{ type: {mime_type} }})", + _var_type=Blob, + _var_data=_var_data, + ) + + def create_object_url(self): + """Create a URL from this Blob object using window.URL.createObjectURL. + + Returns: + A URL string representing the Blob object. + """ + return create_url_from_blob_operation(self) + + +@var_operation +def create_url_from_blob_operation(value: BlobVar): + """Create a URL from a Blob variable using window.URL.createObjectURL. + + Args: + value: The Blob variable to create a URL from. + + Returns: + A URL string representing the Blob object. + """ + return var_operation_return( + js_expression=f"window.URL.createObjectURL({value})", + var_type=str, + ) + + +@dataclasses.dataclass( + eq=False, + frozen=True, + slots=True, +) +class LiteralBlobVar(LiteralVar, BlobVar): + """A literal version of a Blob variable.""" + + _var_value: Blob = dataclasses.field(default_factory=Blob) + + @classmethod + def create( + cls, + value: bytes | str | Blob, + mime_type: str | None = None, + _var_data: VarData | None = None, + ) -> BlobVar: + """Create a literal Blob variable from bytes or string data. + + Args: + value: The data to create the Blob from (bytes or string). + mime_type: The MIME type of the Blob. + _var_data: Optional variable data. + + Returns: + A BlobVar instance representing the Blob. + """ + if not mime_type: + mime_type = cls._determine_mime_type(value) + + if isinstance(value, Blob): + value = value.data + + var_type = type(value) + + if isinstance(value, bytes): + value = f"new Uint8Array({list(value)})" + else: + value = f"'{value}'" + + return cls( + _js_expr=f"new Blob([{value}], {{ type: '{mime_type}' }})", + _var_type=var_type, + _var_data=_var_data, + ) diff --git a/reflex/vars/sequence.py b/reflex/vars/sequence.py index 3824ba8760b..736e9cef798 100644 --- a/reflex/vars/sequence.py +++ b/reflex/vars/sequence.py @@ -33,6 +33,7 @@ var_operation, var_operation_return, ) +from .blob import Blob from .number import ( BooleanVar, LiteralNumberVar, @@ -860,6 +861,45 @@ def replace(self, search_value: Any, new_value: Any) -> StringVar: # pyright: i return string_replace_operation(self, search_value, new_value) + def encode(self, encoding: StringVar | str = "utf-8"): + """Encode the string to bytes using the specified encoding. + + Args: + encoding: The character encoding to use. Defaults to "utf-8". + + Returns: + The encoded bytes. + """ + return string_encode_operation(self, encoding) + + def to_blob(self, mime_type: StringVar | str = "text/plain"): + """Convert the string to a Blob object. + + Args: + mime_type: The MIME type of the blob. Defaults to "text/plain". + + Returns: + A Blob object containing the string data. + """ + return blob_object_create_operation(self, mime_type=mime_type) + + +@var_operation +def string_encode_operation(value: StringVar[Any] | str, encoding: StringVar | str): + """Encode a string to bytes using the specified encoding. + + Args: + value: The string to encode. + encoding: The character encoding to use. + + Returns: + The encoded bytes. + """ + return var_operation_return( + f"(new TextEncoder({encoding})).encode({value})", + var_type=bytes, + ) + @var_operation def string_lt_operation(lhs: StringVar[Any] | str, rhs: StringVar[Any] | str): @@ -1287,6 +1327,120 @@ def json(self) -> str: return json.dumps(self._var_value) +class BytesVar(Var[bytes], python_types=bytes): + """A variable that represents Python bytes as JavaScript Uint8Array.""" + + @classmethod + def create( + cls, value: str | bytes, _var_data: VarData | None = None + ) -> LiteralBytesVar: + """Create a BytesVar from a string or bytes value. + + Args: + value: The string or bytes value to create the var from. + _var_data: Additional hooks and imports associated with the Var. + + Returns: + A LiteralBytesVar representing the bytes value. + """ + if isinstance(value, str): + value = value.encode() + return LiteralVar.create(value, _var_data=_var_data) + + def decode(self, encoding: StringVar | str = "utf-8"): + """Decode bytes to a string using the specified encoding. + + Args: + encoding: The character encoding to use for decoding. Defaults to "utf-8". + + Returns: + A StringVar containing the decoded string. + """ + return bytes_decode_operation(self, encoding) + + def to_blob(self, mime_type: StringVar | str = "application/octet-stream"): + """Convert the bytes to a Blob object. + + Args: + mime_type: The MIME type of the blob. Defaults to "application/octet-stream". + + Returns: + A Blob object containing the bytes data. + """ + return blob_object_create_operation(self, mime_type=mime_type) + + +@var_operation +def bytes_decode_operation(value: BytesVar, encoding: StringVar | str): + """Decode bytes to a string using the specified encoding. + + Args: + value: The BytesVar to decode. + encoding: The character encoding to use for decoding. + + Returns: + A StringVar containing the decoded string. + """ + return var_operation_return( + f"(new TextDecoder({encoding})).decode(new Uint8Array({value}))", + var_type=str, + ) + + +@dataclasses.dataclass( + eq=False, + frozen=True, + slots=True, +) +class LiteralBytesVar(LiteralVar, BytesVar): + """A literal version of BytesVar.""" + + _var_value: bytes = dataclasses.field(default=b"") + + @classmethod + def create(cls, value: bytes, _var_data: VarData | None = None) -> BytesVar: + """Create a LiteralBytesVar from a bytes value. + + Args: + value: The bytes value to create the variable from. + _var_data: Additional variable data, by default None. + + Returns: + A literal bytes variable representing the given bytes value. + """ + return cls( + _js_expr=f"new Uint8Array({list(value)})", + _var_type=bytes, + _var_data=_var_data, + _var_value=value, + ) + + +@var_operation +def blob_object_create_operation( + value: StringVar[Any] | BytesVar, + mime_type: StringVar[Any] | str, +) -> CustomVarOperationReturn[Blob]: + """Create a Blob object from string or bytes data. + + Args: + value: The string or bytes data to convert to a Blob. + mime_type: The MIME type of the blob. + + Returns: + A Blob object containing the data. + """ + if isinstance(value, BytesVar): + return var_operation_return( + js_expression=f"new Blob([new Uint8Array({value})], {{ type: {mime_type} }})", + var_type=Blob, + ) + return var_operation_return( + js_expression=f"new Blob([{value}], {{ type: {mime_type} }})", + var_type=Blob, + ) + + @dataclasses.dataclass( eq=False, frozen=True,