Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/joblib/_utils.py: 39%

1# Adapted from https://stackoverflow.com/a/9558001/2536294 

2 

3import ast 

4import functools 

5import operator as op 

6from dataclasses import dataclass 

7 

8from ._multiprocessing_helpers import mp 

9 

10if mp is not None: 

11 from .externals.loky.process_executor import _ExceptionWithTraceback 

12 

13 

14# supported operators 

15operators = { 

16 ast.Add: op.add, 

17 ast.Sub: op.sub, 

18 ast.Mult: op.mul, 

19 ast.Div: op.truediv, 

20 ast.FloorDiv: op.floordiv, 

21 ast.Mod: op.mod, 

22 ast.Pow: op.pow, 

23 ast.USub: op.neg, 

24} 

25 

26# Largest magnitude any sub-expression is allowed to reach. 

27MAX_VALUE = 10**6 

28 

29 

30def _pow_is_too_large(base, exponent): 

31 """Return True if ``base ** exponent`` cannot fit under `MAX_VALUE`. 

32 

33 Only integers are considered. A float power overflows cheaply, whereas a 

34 large integer power is built out in full before `limit` ever sees the 

35 result: ``999999**999999`` costs seconds of CPU and allocates a number with 

36 millions of digits. 

37 """ 

38 if not (isinstance(base, int) and isinstance(exponent, int)): 

39 return False 

40 # For any |base| >= 2, base ** exponent >= 2 ** exponent, so an exponent 

41 # above the bit length of the limit cannot produce a value that fits. 

42 return abs(base) >= 2 and exponent > MAX_VALUE.bit_length() 

43 

44 

45def eval_expr(expr): 

46 """Somewhat safely evaluate an arithmetic expression. 

47 

48 >>> eval_expr('2*6') 

49 12 

50 >>> eval_expr('2**6') 

51 64 

52 >>> eval_expr('1 + 2*3**(4) / (6 + -7)') 

53 -161.0 

54 

55 Raises ValueError if the expression is invalid, too long 

56 or its computation involves too large values. 

57 """ 

58 # Restrict the length of the expression to avoid potential Python crashes 

59 # as per the documentation of ast.parse. 

60 max_length = 30 

61 if len(expr) > max_length: 

62 raise ValueError( 

63 f"Expression {expr[:max_length]!r}... is too long. " 

64 f"Max length is {max_length}, got {len(expr)}." 

65 ) 

66 try: 

67 return eval_(ast.parse(expr, mode="eval").body) 

68 except (TypeError, SyntaxError, OverflowError, KeyError, ZeroDivisionError) as e: 

69 raise ValueError( 

70 f"{expr!r} is not a valid or supported arithmetic expression." 

71 ) from e 

72 

73 

74def limit(max_=None): 

75 """Return decorator that limits allowed returned values.""" 

76 

77 def decorator(func): 

78 @functools.wraps(func) 

79 def wrapper(*args, **kwargs): 

80 ret = func(*args, **kwargs) 

81 try: 

82 mag = abs(ret) 

83 except TypeError: 

84 pass # not applicable 

85 else: 

86 if mag > max_: 

87 raise ValueError( 

88 f"Numeric literal {ret} is too large, max is {max_}." 

89 ) 

90 return ret 

91 

92 return wrapper 

93 

94 return decorator 

95 

96 

97@limit(max_=MAX_VALUE) 

98def eval_(node): 

99 if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): 

100 return node.value 

101 elif isinstance(node, ast.BinOp): # <left> <operator> <right> 

102 left = eval_(node.left) 

103 right = eval_(node.right) 

104 if isinstance(node.op, ast.Pow) and _pow_is_too_large(left, right): 

105 raise ValueError( 

106 f"Numeric literal {left}**{right} is too large, max is {MAX_VALUE}." 

107 ) 

108 return operators[type(node.op)](left, right) 

109 elif isinstance(node, ast.UnaryOp): # <operator> <operand> e.g., -1 

110 return operators[type(node.op)](eval_(node.operand)) 

111 else: 

112 raise TypeError(node) 

113 

114 

115@dataclass(frozen=True) 

116class _Sentinel: 

117 """A sentinel to mark a parameter as not explicitly set""" 

118 

119 default_value: object 

120 

121 def __repr__(self): 

122 return f"default({self.default_value!r})" 

123 

124 

125class _TracebackCapturingWrapper: 

126 """Protect function call and return error with traceback.""" 

127 

128 def __init__(self, func): 

129 self.func = func 

130 

131 def __call__(self, **kwargs): 

132 try: 

133 return self.func(**kwargs) 

134 except BaseException as e: 

135 return _ExceptionWithTraceback(e) 

136 

137 

138def _retrieve_traceback_capturing_wrapped_call(out): 

139 if isinstance(out, _ExceptionWithTraceback): 

140 rebuild, args = out.__reduce__() 

141 out = rebuild(*args) 

142 if isinstance(out, BaseException): 

143 raise out 

144 return out