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

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

2 

3import ast 

4from dataclasses import dataclass 

5import operator as op 

6 

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 """ 

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

30 12 

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

32 64 

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

34 -161.0 

35 """ 

36 try: 

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

38 except (TypeError, SyntaxError, KeyError) as e: 

39 raise ValueError( 

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

41 ) from e 

42 

43 

44def eval_(node): 

45 if isinstance(node, ast.Constant): # <constant> 

46 return node.value 

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

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

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

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

51 else: 

52 raise TypeError(node) 

53 

54 

55@dataclass(frozen=True) 

56class _Sentinel: 

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

58 default_value: object 

59 

60 def __repr__(self): 

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

62 

63 

64class _TracebackCapturingWrapper: 

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

66 

67 def __init__(self, func): 

68 self.func = func 

69 

70 def __call__(self, **kwargs): 

71 try: 

72 return self.func(**kwargs) 

73 except BaseException as e: 

74 return _ExceptionWithTraceback(e) 

75 

76 

77def _retrieve_traceback_capturing_wrapped_call(out): 

78 if isinstance(out, _ExceptionWithTraceback): 

79 rebuild, args = out.__reduce__() 

80 out = rebuild(*args) 

81 if isinstance(out, BaseException): 

82 raise out 

83 return out