Coverage for schemas/transformers_validators.py: 36%

1import re 

2from typing import Any, Type, Union 

3 

4from pydantic import ValidationInfo 

5 

6from .overrides import Enum 

7 

8# Used in pydantic Field(pattern=...), which compiles with the Rust regex 

9# engine: end-of-string is \z there (\Z is unsupported and fails at schema build). 

10NAME_PATTERN = r"^[a-zA-Z0-9\-éèàç |&/\\_.#':,@]*\z" 

11 

12# Property/column names are interpolated into ClickHouse SQL as identifiers, so 

13# they must be restricted to a safe character set (no back-ticks, quotes, parens, 

14# etc.) to prevent SQL injection. 

15PROPERTY_NAME_PATTERN = r"^[\w$.\- ]+$" 

16 

17 

18def transform_email(email: str) -> str: 

19 return email.lower().strip() if isinstance(email, str) else email 

20 

21 

22def int_to_string(value: int) -> str: 

23 return str(value) if isinstance(value, int) else value 

24 

25 

26def remove_whitespace(value: str) -> str: 

27 return " ".join(value.split()) if isinstance(value, str) else value 

28 

29 

30def remove_duplicate_values(value: list) -> list: 

31 if value is not None and isinstance(value, list): 

32 if len(value) > 0 and (isinstance(value[0], int) or isinstance(value[0], dict)): 

33 return value 

34 value = list(set(value)) 

35 return value 

36 

37 

38def single_to_list(value: Union[list, Any]) -> list: 

39 if value is not None and not isinstance(value, list): 

40 value = [value] 

41 return value 

42 

43 

44def force_is_event(events_enum: list[Type[Enum]]): 

45 def fn(value: list): 

46 if value is not None and isinstance(value, list): 

47 for v in value: 

48 if v.get("type") is None: 

49 v["isEvent"] = False 

50 continue 

51 r = False 

52 for en in events_enum: 

53 if en.has_value(v["type"]) or en.has_value(v["type"].lower()): 

54 r = True 

55 break 

56 v["isEvent"] = r 

57 return value 

58 

59 return fn 

60 

61 

62def check_alphanumeric(v: str, info: ValidationInfo) -> str: 

63 if isinstance(v, str): 63 ↛ 66line 63 didn't jump to line 66 because the condition on line 63 was always true

64 is_alphanumeric = v.replace(" ", "").isalnum() 

65 assert is_alphanumeric, f"{info.field_name} must be alphanumeric" 

66 return v 

67 

68 

69def check_account_name(v: str, info: ValidationInfo) -> str: 

70 if isinstance(v, str) and len(v) > 0: 

71 pattern = r"^[\w\s\-'\.áéíóúàèìòùâêîôûäëïöüãõñçÁÉÍÓÚÀÈÌÒÙÂÊÎÔÛÄËÏÖÜÃÕÑÇ&@]+\Z" 

72 is_valid = re.match(pattern, v, re.UNICODE) 

73 assert is_valid, ( 

74 f"{info.field_name} contains invalid characters. Only letters, numbers, spaces, hyphens, apostrophes, and periods are allowed" 

75 ) 

76 return v 

77 

78 

79def check_property_name(v: str, info: ValidationInfo = None) -> str: 

80 if isinstance(v, str): 

81 field_name = getattr(info, "field_name", None) or "name" 

82 assert re.match(PROPERTY_NAME_PATTERN, v, re.UNICODE), ( 

83 f"{field_name} contains invalid characters. Only letters, numbers, " 

84 "spaces, and the characters $ . - _ are allowed" 

85 ) 

86 return v 

87 

88 

89def check_regex(v: str) -> str: 

90 assert v is not None, "Regex is null" 

91 assert isinstance(v, str), "Regex value must be a string" 

92 assert len(v) > 0, "Regex is empty" 

93 is_valid = None 

94 try: 

95 re.compile(v) 

96 except re.error as exc: 

97 is_valid = f"Invalid regex: {exc} (at position {exc.pos})" 

98 

99 assert is_valid is None, is_valid 

100 return v