Coverage for chalicelib/utils/sql_helper.py: 13%

1from typing import Union 

2from enum import Enum 

3import schemas 

4 

5 

6def get_sql_operator(op: Union[schemas.SearchEventOperator, schemas.ClickEventExtraOperator, schemas.MathOperator]): 

7 return { 

8 schemas.SearchEventOperator.IS: "=", 

9 schemas.SearchEventOperator.ON: "=", 

10 schemas.SearchEventOperator.ON_ANY: "IN", 

11 schemas.SearchEventOperator.IS_NOT: "!=", 

12 schemas.SearchEventOperator.NOT_ON: "!=", 

13 schemas.SearchEventOperator.CONTAINS: "ILIKE", 

14 schemas.SearchEventOperator.NOT_CONTAINS: "NOT ILIKE", 

15 schemas.SearchEventOperator.STARTS_WITH: "ILIKE", 

16 schemas.SearchEventOperator.ENDS_WITH: "ILIKE", 

17 # this is not used as an operator, it is used in order to maintain a valid value for conditions 

18 schemas.SearchEventOperator.PATTERN: "regex", 

19 

20 # Selector operators: 

21 schemas.ClickEventExtraOperator.IS: "=", 

22 schemas.ClickEventExtraOperator.IS_NOT: "!=", 

23 schemas.ClickEventExtraOperator.CONTAINS: "ILIKE", 

24 schemas.ClickEventExtraOperator.NOT_CONTAINS: "NOT ILIKE", 

25 schemas.ClickEventExtraOperator.STARTS_WITH: "ILIKE", 

26 schemas.ClickEventExtraOperator.ENDS_WITH: "ILIKE", 

27 

28 schemas.MathOperator.GREATER: ">", 

29 schemas.MathOperator.GREATER_EQ: ">=", 

30 schemas.MathOperator.LESS: "<", 

31 schemas.MathOperator.LESS_EQ: "<=", 

32 }.get(op, "=") 

33 

34 

35def is_negation_operator(op: schemas.SearchEventOperator): 

36 return op in [schemas.SearchEventOperator.IS_NOT, 

37 schemas.SearchEventOperator.NOT_ON, 

38 schemas.SearchEventOperator.NOT_CONTAINS, 

39 schemas.ClickEventExtraOperator.IS_NOT, 

40 schemas.ClickEventExtraOperator.NOT_CONTAINS] 

41 

42 

43def reverse_sql_operator(op): 

44 return "=" if op == "!=" else "!=" if op == "=" else "ILIKE" if op == "NOT ILIKE" else "NOT ILIKE" 

45 

46 

47def multi_conditions(condition, values, value_key="value", is_not=False) -> str: 

48 query = [] 

49 for i in range(len(values)): 

50 k = f"{value_key}_{i}" 

51 query.append(condition.replace(value_key, k)) 

52 return "(" + (" AND " if is_not else " OR ").join(query) + ")" 

53 

54 

55def multi_values(values, value_key="value", data_type: schemas.PropertyType | None = None): 

56 query_values = {} 

57 if values is not None and isinstance(values, list): 

58 for i in range(len(values)): 

59 k = f"{value_key}_{i}" 

60 query_values[k] = values[i].value if isinstance(values[i], Enum) else values[i] 

61 if data_type: 

62 if data_type == schemas.PropertyType.STRING: 

63 query_values[k] = str(query_values[k]) 

64 

65 return query_values 

66 

67 

68def isAny_opreator(op: schemas.SearchEventOperator): 

69 return op in [schemas.SearchEventOperator.ON_ANY, schemas.SearchEventOperator.IS_ANY] 

70 

71 

72def isUndefined_operator(op: schemas.SearchEventOperator): 

73 return op in [schemas.SearchEventOperator.IS_UNDEFINED] 

74 

75 

76def single_value(values): 

77 if values is not None and isinstance(values, list): 

78 for i, v in enumerate(values): 

79 if isinstance(v, Enum): 

80 values[i] = v.value 

81 return values 

82 

83 

84def coordinate_conditions(condition_x, condition_y, values, value_key="value", is_not=False): 

85 query = [] 

86 if len(values) == 2: 

87 # if 2 values are provided, it means x=v[0] and y=v[1] 

88 for i in range(len(values)): 

89 k = f"{value_key}_{i}" 

90 if i == 0: 

91 query.append(f"{condition_x}=%({k})s") 

92 elif i == 1: 

93 query.append(f"{condition_y}=%({k})s") 

94 

95 elif len(values) == 4: 

96 # if 4 values are provided, it means v[0]<=x<=v[1] and v[2]<=y<=v[3] 

97 for i in range(len(values)): 

98 k = f"{value_key}_{i}" 

99 if i == 0: 

100 query.append(f"{condition_x}>=%({k})s") 

101 elif i == 1: 

102 query.append(f"{condition_x}<=%({k})s") 

103 elif i == 2: 

104 query.append(f"{condition_y}>=%({k})s") 

105 elif i == 3: 

106 query.append(f"{condition_y}<=%({k})s") 

107 

108 return "(" + (" AND " if is_not else " OR ").join(query) + ")"