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

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

2 

3import ast 

4import operator as op 

5from dataclasses import dataclass 

6 

7from ._multiprocessing_helpers import mp 

8 

9if mp is not None: 

10 from .externals.loky.process_executor import _ExceptionWithTraceback 

11 

12 

13# supported operators 

14operators = { 

15 ast.Add: op.add, 

16 ast.Sub: op.sub, 

17 ast.Mult: op.mul, 

18 ast.Div: op.truediv, 

19 ast.FloorDiv: op.floordiv, 

20 ast.Mod: op.mod, 

21 ast.Pow: op.pow, 

22 ast.USub: op.neg, 

23} 

24 

25 

26def eval_expr(expr): 

27 """ 

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

29 12 

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

31 64 

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

33 -161.0 

34 """ 

35 try: 

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

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

38 raise ValueError( 

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

40 ) from e 

41 

42 

43def eval_(node): 

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

45 return node.value 

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

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

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

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

50 else: 

51 raise TypeError(node) 

52 

53 

54@dataclass(frozen=True) 

55class _Sentinel: 

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

57 

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