123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455 |
- """Collection of string classes and utilities."""
- from __future__ import annotations
- import dataclasses
- import functools
- import inspect
- import json
- import re
- import sys
- import typing
- from typing import (
- TYPE_CHECKING,
- Any,
- Callable,
- ClassVar,
- List,
- Sequence,
- Set,
- Tuple,
- Type,
- Union,
- cast,
- )
- from typing_extensions import TypeAliasType, TypeVar
- from reflex import constants
- from reflex.constants.base import REFLEX_VAR_OPENING_TAG
- from reflex.constants.colors import Color
- from reflex.utils.exceptions import VarTypeError
- from reflex.utils.types import GenericType, get_origin
- from .base import (
- CachedVarOperation,
- CustomVarOperationReturn,
- LiteralVar,
- ReflexCallable,
- Var,
- VarData,
- VarWithDefault,
- _global_vars,
- cached_property_no_lock,
- figure_out_type,
- get_python_literal,
- get_unique_variable_name,
- nary_type_computer,
- passthrough_unary_type_computer,
- unionize,
- unwrap_reflex_callalbe,
- var_operation,
- var_operation_return,
- )
- from .number import (
- _AT_SLICE_IMPORT,
- _IS_TRUE_IMPORT,
- _RANGE_IMPORT,
- LiteralNumberVar,
- NumberVar,
- raise_unsupported_operand_types,
- ternary_operation,
- )
- if TYPE_CHECKING:
- from .function import FunctionVar
- STRING_TYPE = TypeVar("STRING_TYPE", default=str)
- ARRAY_VAR_TYPE = TypeVar("ARRAY_VAR_TYPE", bound=Union[Set, Tuple, Sequence])
- OTHER_ARRAY_VAR_TYPE = TypeVar(
- "OTHER_ARRAY_VAR_TYPE", bound=Union[Set, Tuple, Sequence]
- )
- INNER_ARRAY_VAR = TypeVar("INNER_ARRAY_VAR", covariant=True)
- ANOTHER_ARRAY_VAR = TypeVar("ANOTHER_ARRAY_VAR", covariant=True)
- KEY_TYPE = TypeVar("KEY_TYPE")
- VALUE_TYPE = TypeVar("VALUE_TYPE")
- @var_operation
- def string_lt_operation(lhs: Var[str], rhs: Var[str]):
- """Check if a string is less than another string.
- Args:
- lhs: The left-hand side string.
- rhs: The right-hand side string.
- Returns:
- The string less than operation.
- """
- return var_operation_return(js_expression=f"{lhs} < {rhs}", var_type=bool)
- @var_operation
- def string_gt_operation(lhs: Var[str], rhs: Var[str]):
- """Check if a string is greater than another string.
- Args:
- lhs: The left-hand side string.
- rhs: The right-hand side string.
- Returns:
- The string greater than operation.
- """
- return var_operation_return(js_expression=f"{lhs} > {rhs}", var_type=bool)
- @var_operation
- def string_le_operation(lhs: Var[str], rhs: Var[str]):
- """Check if a string is less than or equal to another string.
- Args:
- lhs: The left-hand side string.
- rhs: The right-hand side string.
- Returns:
- The string less than or equal operation.
- """
- return var_operation_return(js_expression=f"{lhs} <= {rhs}", var_type=bool)
- @var_operation
- def string_ge_operation(lhs: Var[str], rhs: Var[str]):
- """Check if a string is greater than or equal to another string.
- Args:
- lhs: The left-hand side string.
- rhs: The right-hand side string.
- Returns:
- The string greater than or equal operation.
- """
- return var_operation_return(js_expression=f"{lhs} >= {rhs}", var_type=bool)
- @var_operation
- def string_lower_operation(string: Var[str]):
- """Convert a string to lowercase.
- Args:
- string: The string to convert.
- Returns:
- The lowercase string.
- """
- return var_operation_return(js_expression=f"{string}.toLowerCase()", var_type=str)
- @var_operation
- def string_upper_operation(string: Var[str]):
- """Convert a string to uppercase.
- Args:
- string: The string to convert.
- Returns:
- The uppercase string.
- """
- return var_operation_return(js_expression=f"{string}.toUpperCase()", var_type=str)
- @var_operation
- def string_strip_operation(string: Var[str]):
- """Strip a string.
- Args:
- string: The string to strip.
- Returns:
- The stripped string.
- """
- return var_operation_return(js_expression=f"{string}.trim()", var_type=str)
- @var_operation
- def string_contains_field_operation(
- haystack: Var[str],
- needle: Var[str],
- field: VarWithDefault[str] = VarWithDefault(""),
- ):
- """Check if a string contains another string.
- Args:
- haystack: The haystack.
- needle: The needle.
- field: The field to check.
- Returns:
- The string contains operation.
- """
- return var_operation_return(
- js_expression=f"isTrue({field}) ? {haystack}.some(obj => obj[{field}] === {needle}) : {haystack}.some(obj => obj === {needle})",
- var_type=bool,
- var_data=VarData(
- imports=_IS_TRUE_IMPORT,
- ),
- )
- @var_operation
- def string_contains_operation(haystack: Var[str], needle: Var[str]):
- """Check if a string contains another string.
- Args:
- haystack: The haystack.
- needle: The needle.
- Returns:
- The string contains operation.
- """
- return var_operation_return(
- js_expression=f"{haystack}.includes({needle})", var_type=bool
- )
- @var_operation
- def string_starts_with_operation(full_string: Var[str], prefix: Var[str]):
- """Check if a string starts with a prefix.
- Args:
- full_string: The full string.
- prefix: The prefix.
- Returns:
- Whether the string starts with the prefix.
- """
- return var_operation_return(
- js_expression=f"{full_string}.startsWith({prefix})", var_type=bool
- )
- @var_operation
- def string_ends_with_operation(full_string: Var[str], suffix: Var[str]):
- """Check if a string ends with a suffix.
- Args:
- full_string: The full string.
- suffix: The suffix.
- Returns:
- Whether the string ends with the suffix.
- """
- return var_operation_return(
- js_expression=f"{full_string}.endsWith({suffix})", var_type=bool
- )
- @var_operation
- def string_item_operation(string: Var[str], index: Var[int]):
- """Get an item from a string.
- Args:
- string: The string.
- index: The index of the item.
- Returns:
- The item from the string.
- """
- return var_operation_return(js_expression=f"{string}.at({index})", var_type=str)
- @var_operation
- def string_replace_operation(
- string: Var[str], search_value: Var[str], new_value: Var[str]
- ):
- """Replace a string with a value.
- Args:
- string: The string.
- search_value: The string to search.
- new_value: The value to be replaced with.
- Returns:
- The string replace operation.
- """
- return var_operation_return(
- js_expression=f"{string}.replace({search_value}, {new_value})",
- var_type=str,
- )
- @var_operation
- def array_pluck_operation(
- array: Var[Sequence[Any]],
- field: Var[str],
- ) -> CustomVarOperationReturn[Sequence[Any]]:
- """Pluck a field from an array of objects.
- Args:
- array: The array to pluck from.
- field: The field to pluck from the objects in the array.
- Returns:
- The reversed array.
- """
- return var_operation_return(
- js_expression=f"{array}.map(e=>e?.[{field}])",
- var_type=List[Any],
- )
- @var_operation
- def array_join_operation(
- array: Var[Sequence[Any]], sep: VarWithDefault[str] = VarWithDefault("")
- ):
- """Join the elements of an array.
- Args:
- array: The array.
- sep: The separator.
- Returns:
- The joined elements.
- """
- return var_operation_return(js_expression=f"{array}.join({sep})", var_type=str)
- @var_operation
- def array_reverse_operation(
- array: Var[Sequence[INNER_ARRAY_VAR]],
- ) -> CustomVarOperationReturn[Sequence[INNER_ARRAY_VAR]]:
- """Reverse an array.
- Args:
- array: The array to reverse.
- Returns:
- The reversed array.
- """
- return var_operation_return(
- js_expression=f"{array}.slice().reverse()",
- type_computer=passthrough_unary_type_computer(ReflexCallable[[List], List]),
- )
- @var_operation
- def array_lt_operation(lhs: Var[ARRAY_VAR_TYPE], rhs: Var[ARRAY_VAR_TYPE]):
- """Check if an array is less than another array.
- Args:
- lhs: The left-hand side array.
- rhs: The right-hand side array.
- Returns:
- The array less than operation.
- """
- return var_operation_return(js_expression=f"{lhs} < {rhs}", var_type=bool)
- @var_operation
- def array_gt_operation(lhs: Var[ARRAY_VAR_TYPE], rhs: Var[ARRAY_VAR_TYPE]):
- """Check if an array is greater than another array.
- Args:
- lhs: The left-hand side array.
- rhs: The right-hand side array.
- Returns:
- The array greater than operation.
- """
- return var_operation_return(js_expression=f"{lhs} > {rhs}", var_type=bool)
- @var_operation
- def array_le_operation(lhs: Var[ARRAY_VAR_TYPE], rhs: Var[ARRAY_VAR_TYPE]):
- """Check if an array is less than or equal to another array.
- Args:
- lhs: The left-hand side array.
- rhs: The right-hand side array.
- Returns:
- The array less than or equal operation.
- """
- return var_operation_return(js_expression=f"{lhs} <= {rhs}", var_type=bool)
- @var_operation
- def array_ge_operation(lhs: Var[ARRAY_VAR_TYPE], rhs: Var[ARRAY_VAR_TYPE]):
- """Check if an array is greater than or equal to another array.
- Args:
- lhs: The left-hand side array.
- rhs: The right-hand side array.
- Returns:
- The array greater than or equal operation.
- """
- return var_operation_return(js_expression=f"{lhs} >= {rhs}", var_type=bool)
- @var_operation
- def array_length_operation(array: Var[ARRAY_VAR_TYPE]):
- """Get the length of an array.
- Args:
- array: The array.
- Returns:
- The length of the array.
- """
- return var_operation_return(
- js_expression=f"{array}.length",
- var_type=int,
- )
- @var_operation
- def string_split_operation(
- string: Var[str], sep: VarWithDefault[str] = VarWithDefault("")
- ):
- """Split a string.
- Args:
- string: The string to split.
- sep: The separator.
- Returns:
- The split string.
- """
- return var_operation_return(
- js_expression=f"isTrue({sep}) ? {string}.split({sep}) : [...{string}]",
- var_type=Sequence[str],
- var_data=VarData(imports=_IS_TRUE_IMPORT),
- )
- def _element_type(array: Var, index: Var) -> Any:
- array_args = typing.get_args(array._var_type)
- if (
- array_args
- and isinstance(index, LiteralNumberVar)
- and is_tuple_type(array._var_type)
- ):
- index_value = int(index._var_value)
- return array_args[index_value % len(array_args)]
- return unionize(*(array_arg for array_arg in array_args if array_arg is not ...))
- @var_operation
- def array_item_or_slice_operation(
- array: Var[Sequence[INNER_ARRAY_VAR]],
- index_or_slice: Var[Union[int, slice]],
- ) -> CustomVarOperationReturn[Union[INNER_ARRAY_VAR, Sequence[INNER_ARRAY_VAR]]]:
- """Get an item or slice from an array.
- Args:
- array: The array.
- index_or_slice: The index or slice.
- Returns:
- The item or slice from the array.
- """
- return var_operation_return(
- js_expression=f"Array.isArray({index_or_slice}) ? at_slice({array}, {index_or_slice}) : {array}.at({index_or_slice})",
- type_computer=nary_type_computer(
- ReflexCallable[[Sequence, Union[int, slice]], Any],
- ReflexCallable[[Union[int, slice]], Any],
- computer=lambda args: (
- args[0]._var_type
- if args[1]._var_type is slice
- else (_element_type(args[0], args[1]))
- ),
- ),
- var_data=VarData(
- imports=_AT_SLICE_IMPORT,
- ),
- )
- @var_operation
- def array_slice_operation(
- array: Var[Sequence[INNER_ARRAY_VAR]],
- slice: Var[slice],
- ) -> CustomVarOperationReturn[Sequence[INNER_ARRAY_VAR]]:
- """Get a slice from an array.
- Args:
- array: The array.
- slice: The slice.
- Returns:
- The item or slice from the array.
- """
- return var_operation_return(
- js_expression=f"atSlice({array}, {slice})",
- type_computer=nary_type_computer(
- ReflexCallable[[List, slice], Any],
- ReflexCallable[[slice], Any],
- computer=lambda args: args[0]._var_type,
- ),
- var_data=VarData(
- imports=_AT_SLICE_IMPORT,
- ),
- )
- @var_operation
- def array_item_operation(
- array: Var[Sequence[INNER_ARRAY_VAR]], index: Var[int]
- ) -> CustomVarOperationReturn[INNER_ARRAY_VAR]:
- """Get an item from an array.
- Args:
- array: The array.
- index: The index of the item.
- Returns:
- The item from the array.
- """
- def type_computer(*args):
- if len(args) == 0:
- return (
- ReflexCallable[[List[Any], int], Any],
- functools.partial(type_computer, *args),
- )
- array = args[0]
- array_args = typing.get_args(array._var_type)
- if len(args) == 1:
- return (
- ReflexCallable[[int], unionize(*array_args)],
- functools.partial(type_computer, *args),
- )
- index = args[1]
- if (
- array_args
- and isinstance(index, LiteralNumberVar)
- and is_tuple_type(array._var_type)
- ):
- index_value = int(index._var_value)
- element_type = array_args[index_value % len(array_args)]
- else:
- element_type = unionize(*array_args)
- return (ReflexCallable[[], element_type], None)
- return var_operation_return(
- js_expression=f"{array}.at({index})",
- type_computer=type_computer,
- )
- @var_operation
- def array_range_operation(
- e1: Var[int],
- e2: VarWithDefault[int | None] = VarWithDefault(None),
- step: VarWithDefault[int] = VarWithDefault(1),
- ) -> CustomVarOperationReturn[Sequence[int]]:
- """Create a range of numbers.
- Args:
- e1: The end of the range if e2 is not provided, otherwise the start of the range.
- e2: The end of the range.
- step: The step of the range.
- Returns:
- The range of numbers.
- """
- return var_operation_return(
- js_expression=f"range({e1}, {e2}, {step})",
- var_type=List[int],
- var_data=VarData(
- imports=_RANGE_IMPORT,
- ),
- )
- @var_operation
- def array_contains_field_operation(
- haystack: Var[ARRAY_VAR_TYPE],
- needle: Var[Any],
- field: VarWithDefault[str] = VarWithDefault(""),
- ):
- """Check if an array contains an element.
- Args:
- haystack: The array to check.
- needle: The element to check for.
- field: The field to check.
- Returns:
- The array contains operation.
- """
- return var_operation_return(
- js_expression=f"isTrue({field}) ? {haystack}.some(obj => obj[{field}] === {needle}) : {haystack}.some(obj => obj === {needle})",
- var_type=bool,
- var_data=VarData(
- imports=_IS_TRUE_IMPORT,
- ),
- )
- @var_operation
- def array_contains_operation(haystack: Var[ARRAY_VAR_TYPE], needle: Var):
- """Check if an array contains an element.
- Args:
- haystack: The array to check.
- needle: The element to check for.
- Returns:
- The array contains operation.
- """
- return var_operation_return(
- js_expression=f"{haystack}.includes({needle})",
- var_type=bool,
- )
- @var_operation
- def repeat_array_operation(
- array: Var[Sequence[INNER_ARRAY_VAR]], count: Var[int]
- ) -> CustomVarOperationReturn[Sequence[INNER_ARRAY_VAR]]:
- """Repeat an array a number of times.
- Args:
- array: The array to repeat.
- count: The number of times to repeat the array.
- Returns:
- The repeated array.
- """
- def type_computer(*args: Var):
- if not args:
- return (
- ReflexCallable[[List[Any], int], List[Any]],
- type_computer,
- )
- if len(args) == 1:
- return (
- ReflexCallable[[int], args[0]._var_type],
- functools.partial(type_computer, *args),
- )
- return (ReflexCallable[[], args[0]._var_type], None)
- return var_operation_return(
- js_expression=f"Array.from({{ length: {count} }}).flatMap(() => {array})",
- type_computer=type_computer,
- )
- @var_operation
- def repeat_string_operation(
- string: Var[str], count: Var[int]
- ) -> CustomVarOperationReturn[str]:
- """Repeat a string a number of times.
- Args:
- string: The string to repeat.
- count: The number of times to repeat the string.
- Returns:
- The repeated string.
- """
- return var_operation_return(
- js_expression=f"{string}.repeat({count})",
- var_type=str,
- )
- if TYPE_CHECKING:
- pass
- @var_operation
- def map_array_operation(
- array: Var[Sequence[INNER_ARRAY_VAR]],
- function: Var[
- ReflexCallable[[INNER_ARRAY_VAR], ANOTHER_ARRAY_VAR]
- | ReflexCallable[[], ANOTHER_ARRAY_VAR]
- ],
- ) -> CustomVarOperationReturn[Sequence[ANOTHER_ARRAY_VAR]]:
- """Map a function over an array.
- Args:
- array: The array.
- function: The function to map.
- Returns:
- The mapped array.
- """
- def type_computer(*args: Var):
- if not args:
- return (
- ReflexCallable[[List[Any], ReflexCallable], List[Any]],
- type_computer,
- )
- if len(args) == 1:
- return (
- ReflexCallable[[ReflexCallable], List[Any]],
- functools.partial(type_computer, *args),
- )
- return (ReflexCallable[[], List[args[0]._var_type]], None)
- return var_operation_return(
- js_expression=f"{array}.map({function})",
- type_computer=nary_type_computer(
- ReflexCallable[[List[Any], ReflexCallable], List[Any]],
- ReflexCallable[[ReflexCallable], List[Any]],
- computer=lambda args: List[unwrap_reflex_callalbe(args[1]._var_type)[1]], # type: ignore
- ),
- )
- @var_operation
- def array_concat_operation(
- lhs: Var[Sequence[INNER_ARRAY_VAR]], rhs: Var[Sequence[ANOTHER_ARRAY_VAR]]
- ) -> CustomVarOperationReturn[Sequence[INNER_ARRAY_VAR | ANOTHER_ARRAY_VAR]]:
- """Concatenate two arrays.
- Args:
- lhs: The left-hand side array.
- rhs: The right-hand side array.
- Returns:
- The concatenated array.
- """
- return var_operation_return(
- js_expression=f"[...{lhs}, ...{rhs}]",
- type_computer=nary_type_computer(
- ReflexCallable[[List[Any], List[Any]], List[Any]],
- ReflexCallable[[List[Any]], List[Any]],
- computer=lambda args: unionize(args[0]._var_type, args[1]._var_type),
- ),
- )
- @var_operation
- def string_concat_operation(
- lhs: Var[str], rhs: Var[str]
- ) -> CustomVarOperationReturn[str]:
- """Concatenate two strings.
- Args:
- lhs: The left-hand side string.
- rhs: The right-hand side string.
- Returns:
- The concatenated string.
- """
- return var_operation_return(
- js_expression=f"{lhs} + {rhs}",
- var_type=str,
- )
- @var_operation
- def reverse_string_concat_operation(
- lhs: Var[str], rhs: Var[str]
- ) -> CustomVarOperationReturn[str]:
- """Concatenate two strings in reverse order.
- Args:
- lhs: The left-hand side string.
- rhs: The right-hand side string.
- Returns:
- The concatenated string.
- """
- return var_operation_return(
- js_expression=f"{rhs} + {lhs}",
- var_type=str,
- )
- class SliceVar(Var[slice], python_types=slice):
- """Base class for immutable slice vars."""
- @dataclasses.dataclass(
- eq=False,
- frozen=True,
- **{"slots": True} if sys.version_info >= (3, 10) else {},
- )
- class LiteralSliceVar(CachedVarOperation, LiteralVar, SliceVar):
- """Base class for immutable literal slice vars."""
- _var_value: slice = dataclasses.field(default_factory=lambda: slice(None))
- @cached_property_no_lock
- def _cached_var_name(self) -> str:
- """The name of the var.
- Returns:
- The name of the var.
- """
- return f"[{LiteralVar.create(self._var_value.start)!s}, {LiteralVar.create(self._var_value.stop)!s}, {LiteralVar.create(self._var_value.step)!s}]"
- @cached_property_no_lock
- def _cached_get_all_var_data(self) -> VarData | None:
- """Get all the VarData asVarDatae Var.
- Returns:
- The VarData associated with the Var.
- """
- return VarData.merge(
- *[
- var._get_all_var_data()
- for var in [
- self._var_value.start,
- self._var_value.stop,
- self._var_value.step,
- ]
- if isinstance(var, Var)
- ],
- self._var_data,
- )
- @classmethod
- def create(
- cls,
- value: slice,
- _var_type: Type[slice] | None = None,
- _var_data: VarData | None = None,
- ) -> SliceVar:
- """Create a var from a slice value.
- Args:
- value: The value to create the var from.
- _var_type: The type of the var.
- _var_data: Additional hooks and imports associated with the Var.
- Returns:
- The var.
- """
- return cls(
- _js_expr="",
- _var_type=slice if _var_type is None else _var_type,
- _var_data=_var_data,
- _var_value=value,
- )
- def __hash__(self) -> int:
- """Get the hash of the var.
- Returns:
- The hash of the var.
- """
- return hash(
- (
- self.__class__.__name__,
- self._var_value.start,
- self._var_value.stop,
- self._var_value.step,
- )
- )
- def json(self) -> str:
- """Get the JSON representation of the var.
- Returns:
- The JSON representation of the var.
- """
- return json.dumps(
- [self._var_value.start, self._var_value.stop, self._var_value.step]
- )
- class ArrayVar(Var[ARRAY_VAR_TYPE], python_types=(Sequence, set)):
- """Base class for immutable array vars."""
- join = array_join_operation
- reverse = array_reverse_operation
- __add__ = array_concat_operation
- __getitem__ = array_item_or_slice_operation
- at = array_item_operation
- slice = array_slice_operation
- length = array_length_operation
- range: ClassVar[
- FunctionVar[
- ReflexCallable[
- [int, VarWithDefault[int | None], VarWithDefault[int]], Sequence[int]
- ]
- ]
- ] = array_range_operation
- contains = array_contains_field_operation
- pluck = array_pluck_operation
- __rmul__ = __mul__ = repeat_array_operation
- __lt__ = array_lt_operation
- __gt__ = array_gt_operation
- __le__ = array_le_operation
- __ge__ = array_ge_operation
- def foreach(
- self: ArrayVar[Sequence[INNER_ARRAY_VAR]],
- fn: Callable[[Var[INNER_ARRAY_VAR]], ANOTHER_ARRAY_VAR]
- | Callable[[], ANOTHER_ARRAY_VAR],
- ) -> ArrayVar[Sequence[ANOTHER_ARRAY_VAR]]:
- """Apply a function to each element of the array.
- Args:
- fn: The function to apply.
- Returns:
- The array after applying the function.
- Raises:
- VarTypeError: If the function takes more than one argument.
- """
- from .function import ArgsFunctionOperation
- if not callable(fn):
- raise_unsupported_operand_types("foreach", (type(self), type(fn)))
- # get the number of arguments of the function
- num_args = len(inspect.signature(fn).parameters)
- if num_args > 1:
- raise VarTypeError(
- "The function passed to foreach should take at most one argument."
- )
- if num_args == 0:
- return_value = fn() # type: ignore
- simple_function_var: FunctionVar[ReflexCallable[[], ANOTHER_ARRAY_VAR]] = (
- ArgsFunctionOperation.create((), return_value)
- )
- return map_array_operation(self, simple_function_var).guess_type()
- # generic number var
- number_var = Var("").to(NumberVar, int)
- first_arg_type = self.__getitem__(number_var)._var_type
- arg_name = get_unique_variable_name()
- # get first argument type
- first_arg = cast(
- Var[Any],
- Var(
- _js_expr=arg_name,
- _var_type=first_arg_type,
- ).guess_type(),
- )
- function_var = cast(
- Var[ReflexCallable[[INNER_ARRAY_VAR], ANOTHER_ARRAY_VAR]],
- ArgsFunctionOperation.create(
- (arg_name,),
- Var.create(fn(first_arg)), # type: ignore
- ),
- )
- return map_array_operation.call(self, function_var).guess_type()
- LIST_ELEMENT = TypeVar("LIST_ELEMENT", covariant=True)
- ARRAY_VAR_OF_LIST_ELEMENT = TypeAliasType(
- "ARRAY_VAR_OF_LIST_ELEMENT",
- Union[
- ArrayVar[Sequence[LIST_ELEMENT]],
- ArrayVar[Set[LIST_ELEMENT]],
- ],
- type_params=(LIST_ELEMENT,),
- )
- @dataclasses.dataclass(
- eq=False,
- frozen=True,
- **{"slots": True} if sys.version_info >= (3, 10) else {},
- )
- class LiteralArrayVar(CachedVarOperation, LiteralVar, ArrayVar[ARRAY_VAR_TYPE]):
- """Base class for immutable literal array vars."""
- _var_value: Union[
- Sequence[Union[Var, Any]],
- Set[Union[Var, Any]],
- ] = dataclasses.field(default_factory=list)
- @cached_property_no_lock
- def _cached_var_name(self) -> str:
- """The name of the var.
- Returns:
- The name of the var.
- """
- return (
- "["
- + ", ".join(
- [str(LiteralVar.create(element)) for element in self._var_value]
- )
- + "]"
- )
- @cached_property_no_lock
- def _cached_get_all_var_data(self) -> VarData | None:
- """Get all the VarData associated with the Var.
- Returns:
- The VarData associated with the Var.
- """
- return VarData.merge(
- *[
- LiteralVar.create(element)._get_all_var_data()
- for element in self._var_value
- ],
- self._var_data,
- )
- def __hash__(self) -> int:
- """Get the hash of the var.
- Returns:
- The hash of the var.
- """
- return hash((self.__class__.__name__, self._js_expr))
- def json(self) -> str:
- """Get the JSON representation of the var.
- Returns:
- The JSON representation of the var.
- """
- return (
- "["
- + ", ".join(
- [LiteralVar.create(element).json() for element in self._var_value]
- )
- + "]"
- )
- @classmethod
- def create(
- cls,
- value: ARRAY_VAR_TYPE,
- _var_type: Type[ARRAY_VAR_TYPE] | None = None,
- _var_data: VarData | None = None,
- ) -> LiteralArrayVar[ARRAY_VAR_TYPE]:
- """Create a var from a string value.
- Args:
- value: The value to create the var from.
- _var_data: Additional hooks and imports associated with the Var.
- Returns:
- The var.
- """
- return cls(
- _js_expr="",
- _var_type=figure_out_type(value) if _var_type is None else _var_type,
- _var_data=_var_data,
- _var_value=value,
- )
- class StringVar(Var[STRING_TYPE], python_types=str):
- """Base class for immutable string vars."""
- __add__ = string_concat_operation
- __radd__ = reverse_string_concat_operation
- __getitem__ = string_item_operation
- lower = string_lower_operation
- upper = string_upper_operation
- strip = string_strip_operation
- contains = string_contains_field_operation
- split = string_split_operation
- length = split.chain(array_length_operation)
- reversed = split.chain(array_reverse_operation).chain(array_join_operation)
- startswith = string_starts_with_operation
- __rmul__ = __mul__ = repeat_string_operation
- __lt__ = string_lt_operation
- __gt__ = string_gt_operation
- __le__ = string_le_operation
- __ge__ = string_ge_operation
- # Compile regex for finding reflex var tags.
- _decode_var_pattern_re = (
- rf"{constants.REFLEX_VAR_OPENING_TAG}(.*?){constants.REFLEX_VAR_CLOSING_TAG}"
- )
- _decode_var_pattern = re.compile(_decode_var_pattern_re, flags=re.DOTALL)
- @dataclasses.dataclass(
- eq=False,
- frozen=True,
- **{"slots": True} if sys.version_info >= (3, 10) else {},
- )
- class LiteralStringVar(LiteralVar, StringVar[str]):
- """Base class for immutable literal string vars."""
- _var_value: str = dataclasses.field(default="")
- @classmethod
- def create(
- cls,
- value: str,
- _var_type: GenericType | None = None,
- _var_data: VarData | None = None,
- ) -> StringVar:
- """Create a var from a string value.
- Args:
- value: The value to create the var from.
- _var_type: The type of the var.
- _var_data: Additional hooks and imports associated with the Var.
- Returns:
- The var.
- """
- # Determine var type in case the value is inherited from str.
- _var_type = _var_type or type(value) or str
- if REFLEX_VAR_OPENING_TAG in value:
- strings_and_vals: list[Var | str] = []
- offset = 0
- # Find all tags
- while m := _decode_var_pattern.search(value):
- start, end = m.span()
- strings_and_vals.append(value[:start])
- serialized_data = m.group(1)
- if serialized_data.isnumeric() or (
- serialized_data[0] == "-" and serialized_data[1:].isnumeric()
- ):
- # This is a global immutable var.
- var = _global_vars[int(serialized_data)]
- strings_and_vals.append(var)
- value = value[(end + len(var._js_expr)) :]
- offset += end - start
- strings_and_vals.append(value)
- filtered_strings_and_vals = [
- s for s in strings_and_vals if isinstance(s, Var) or s
- ]
- if len(filtered_strings_and_vals) == 1:
- only_string = filtered_strings_and_vals[0]
- if isinstance(only_string, str):
- return LiteralVar.create(only_string).to(StringVar, _var_type)
- else:
- return only_string.to(StringVar, only_string._var_type)
- if len(
- literal_strings := [
- s
- for s in filtered_strings_and_vals
- if isinstance(s, (str, LiteralStringVar))
- ]
- ) == len(filtered_strings_and_vals):
- return LiteralStringVar.create(
- "".join(
- s._var_value if isinstance(s, LiteralStringVar) else s
- for s in literal_strings
- ),
- _var_type=_var_type,
- _var_data=VarData.merge(
- _var_data,
- *(
- s._get_all_var_data()
- for s in filtered_strings_and_vals
- if isinstance(s, Var)
- ),
- ),
- )
- concat_result = ConcatVarOperation.create(
- *filtered_strings_and_vals,
- _var_data=_var_data,
- )
- return (
- concat_result
- if _var_type is str
- else concat_result.to(StringVar, _var_type)
- )
- return LiteralStringVar(
- _js_expr=json.dumps(value),
- _var_type=_var_type,
- _var_data=_var_data,
- _var_value=value,
- )
- def __hash__(self) -> int:
- """Get the hash of the var.
- Returns:
- The hash of the var.
- """
- return hash((self.__class__.__name__, self._var_value))
- def json(self) -> str:
- """Get the JSON representation of the var.
- Returns:
- The JSON representation of the var.
- """
- return json.dumps(self._var_value)
- @dataclasses.dataclass(
- eq=False,
- frozen=True,
- **{"slots": True} if sys.version_info >= (3, 10) else {},
- )
- class ConcatVarOperation(CachedVarOperation, StringVar[str]):
- """Representing a concatenation of literal string vars."""
- _var_value: Tuple[Var, ...] = dataclasses.field(default_factory=tuple)
- @cached_property_no_lock
- def _cached_var_name(self) -> str:
- """The name of the var.
- Returns:
- The name of the var.
- """
- list_of_strs: List[Union[str, Var]] = []
- last_string = ""
- for var in self._var_value:
- if isinstance(var, LiteralStringVar):
- last_string += var._var_value
- else:
- if last_string:
- list_of_strs.append(last_string)
- last_string = ""
- list_of_strs.append(var)
- if last_string:
- list_of_strs.append(last_string)
- list_of_strs_filtered = [
- str(LiteralVar.create(s)) for s in list_of_strs if isinstance(s, Var) or s
- ]
- if len(list_of_strs_filtered) == 1:
- return list_of_strs_filtered[0]
- return "(" + "+".join(list_of_strs_filtered) + ")"
- @cached_property_no_lock
- def _cached_get_all_var_data(self) -> VarData | None:
- """Get all the VarData asVarDatae Var.
- Returns:
- The VarData associated with the Var.
- """
- return VarData.merge(
- *[
- var._get_all_var_data()
- for var in self._var_value
- if isinstance(var, Var)
- ],
- self._var_data,
- )
- @classmethod
- def create(
- cls,
- *value: Var | str,
- _var_data: VarData | None = None,
- ) -> ConcatVarOperation:
- """Create a var from a string value.
- Args:
- value: The values to concatenate.
- _var_data: Additional hooks and imports associated with the Var.
- Returns:
- The var.
- """
- return cls(
- _js_expr="",
- _var_type=str,
- _var_data=_var_data,
- _var_value=tuple(map(LiteralVar.create, value)),
- )
- def is_tuple_type(t: GenericType) -> bool:
- """Check if a type is a tuple type.
- Args:
- t: The type to check.
- Returns:
- Whether the type is a tuple type.
- """
- if inspect.isclass(t):
- return issubclass(t, tuple)
- return get_origin(t) is tuple
- class ColorVar(StringVar[Color], python_types=Color):
- """Base class for immutable color vars."""
- @dataclasses.dataclass(
- eq=False,
- frozen=True,
- **{"slots": True} if sys.version_info >= (3, 10) else {},
- )
- class LiteralColorVar(CachedVarOperation, LiteralVar, ColorVar):
- """Base class for immutable literal color vars."""
- _var_value: Color = dataclasses.field(default_factory=lambda: Color(color="black"))
- @classmethod
- def create(
- cls,
- value: Color,
- _var_type: Type[Color] | None = None,
- _var_data: VarData | None = None,
- ) -> ColorVar:
- """Create a var from a string value.
- Args:
- value: The value to create the var from.
- _var_type: The type of the var.
- _var_data: Additional hooks and imports associated with the Var.
- Returns:
- The var.
- """
- return cls(
- _js_expr="",
- _var_type=_var_type or Color,
- _var_data=_var_data,
- _var_value=value,
- )
- def __hash__(self) -> int:
- """Get the hash of the var.
- Returns:
- The hash of the var.
- """
- return hash(
- (
- self.__class__.__name__,
- self._var_value.color,
- self._var_value.alpha,
- self._var_value.shade,
- )
- )
- @cached_property_no_lock
- def _cached_var_name(self) -> str:
- """The name of the var.
- Returns:
- The name of the var.
- """
- alpha = cast(Union[Var[bool], bool], self._var_value.alpha)
- alpha = (
- ternary_operation(
- alpha,
- LiteralStringVar.create("a"),
- LiteralStringVar.create(""),
- )
- if isinstance(alpha, Var)
- else LiteralStringVar.create("a" if alpha else "")
- )
- shade = self._var_value.shade
- shade = (
- shade.to_string(use_json=False)
- if isinstance(shade, Var)
- else LiteralStringVar.create(str(shade))
- )
- return str(
- ConcatVarOperation.create(
- LiteralStringVar.create("var(--"),
- self._var_value.color,
- LiteralStringVar.create("-"),
- alpha,
- shade,
- LiteralStringVar.create(")"),
- )
- )
- @cached_property_no_lock
- def _cached_get_all_var_data(self) -> VarData | None:
- """Get all the var data.
- Returns:
- The var data.
- """
- return VarData.merge(
- *[
- LiteralVar.create(var)._get_all_var_data()
- for var in (
- self._var_value.color,
- self._var_value.alpha,
- self._var_value.shade,
- )
- ],
- self._var_data,
- )
- def json(self) -> str:
- """Get the JSON representation of the var.
- Returns:
- The JSON representation of the var.
- Raises:
- TypeError: If the color is not a valid color.
- """
- color, alpha, shade = map(
- get_python_literal,
- (self._var_value.color, self._var_value.alpha, self._var_value.shade),
- )
- if color is None or alpha is None or shade is None:
- raise TypeError("Cannot serialize color that contains non-literal vars.")
- if (
- not isinstance(color, str)
- or not isinstance(alpha, bool)
- or not isinstance(shade, int)
- ):
- raise TypeError("Color is not a valid color.")
- return f"var(--{color}-{'a' if alpha else ''}{shade})"
|