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

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 

27def eval_expr(expr): 

28 """Somewhat safely evaluate an arithmetic expression. 

29 

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

31 12 

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

33 64 

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

35 -161.0 

36 

37 Raises ValueError if the expression is invalid, too long 

38 or its computation involves too large values. 

39 """ 

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

41 # as per the documentation of ast.parse. 

42 max_length = 30 

43 if len(expr) > max_length: 

44 raise ValueError( 

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

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

47 ) 

48 try: 

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

50 except (TypeError, SyntaxError, OverflowError, KeyError) as e: 

51 raise ValueError( 

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

53 ) from e 

54 

55 

56def limit(max_=None): 

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

58 

59 def decorator(func): 

60 @functools.wraps(func) 

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

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

63 try: 

64 mag = abs(ret) 

65 except TypeError: 

66 pass # not applicable 

67 else: 

68 if mag > max_: 

69 raise ValueError( 

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

71 ) 

72 return ret 

73 

74 return wrapper 

75 

76 return decorator 

77 

78 

79@limit(max_=10**6) 

80def eval_(node): 

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

82 return node.value 

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

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

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

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

87 else: 

88 raise TypeError(node) 

89 

90 

91@dataclass(frozen=True) 

92class _Sentinel: 

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

94 

95 default_value: object 

96 

97 def __repr__(self): 

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

99 

100 

101class _TracebackCapturingWrapper: 

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

103 

104 def __init__(self, func): 

105 self.func = func 

106 

107 def __call__(self, **kwargs): 

108 try: 

109 return self.func(**kwargs) 

110 except BaseException as e: 

111 return _ExceptionWithTraceback(e) 

112 

113 

114def _retrieve_traceback_capturing_wrapped_call(out): 

115 if isinstance(out, _ExceptionWithTraceback): 

116 rebuild, args = out.__reduce__() 

117 out = rebuild(*args) 

118 if isinstance(out, BaseException): 

119 raise out 

120 return out