Coverage for dataexcept/wrapping.py: 96%

1"""Turning a third-party exception into a DataExcept one. 

2 

3The pattern this replaces is everywhere in pipeline code:: 

4 

5 try: 

6 frame = pd.read_csv(path) 

7 except OSError as exc: 

8 raise DataLoadingError(path, exc) from exc 

9 

10It is easy to write and easy to get subtly wrong: forget the ``from exc`` and 

11the traceback stops showing what actually failed; pass the original to the 

12wrong parameter and it is not recorded at all; catch too broadly and a 

13``KeyboardInterrupt`` becomes a data-loading error. 

14 

15:func:`wrap` and :func:`wrapping` do the same thing with the wiring settled. 

16New cause-aware constructors use the canonical keyword ``cause``. Legacy 

17``original`` and ``original_exception`` parameters remain supported, and 

18``__cause__`` is set either way so a traceback always shows both failures. 

19""" 

20 

21from __future__ import annotations 

22 

23import contextlib 

24import inspect 

25from typing import Any, Iterator, Tuple, Type, Union 

26 

27from .base import DataExceptError 

28from .failure_metadata import FailureMetadata 

29 

30__all__ = ["wrap", "wrapping"] 

31 

32_CAUSE_PARAMETERS = ("cause", "original", "original_exception") 

33 

34Catchable = Union[Type[BaseException], Tuple[Type[BaseException], ...]] 

35 

36 

37def _cause_parameter(target: Type[DataExceptError]) -> str | None: 

38 """Return the parameter of *target* that takes a wrapped exception.""" 

39 try: 

40 parameters = inspect.signature(target.__init__).parameters 

41 except (TypeError, ValueError): # pragma: no cover - builtins and C types 

42 return None 

43 for name in _CAUSE_PARAMETERS: 

44 if name in parameters: 

45 return name 

46 return None 

47 

48 

49def _explicit_cause_parameter(kwargs: dict[str, Any]) -> str | None: 

50 """Return an explicitly supplied canonical or legacy cause keyword.""" 

51 for name in _CAUSE_PARAMETERS: 

52 if name in kwargs: 

53 return name 

54 return None 

55 

56 

57def wrap( 

58 original: BaseException, 

59 target: Type[DataExceptError], 

60 /, 

61 *, 

62 failure_metadata: FailureMetadata | None = None, 

63 **kwargs: Any, 

64) -> DataExceptError: 

65 """Build *target* from *original*, recording it as the cause. 

66 

67 Extra keyword arguments are passed through to the target constructor. If 

68 the target accepts a cause parameter, *original* is injected unless the 

69 caller already supplied ``cause``, ``original`` or ``original_exception``. 

70 The resulting exception is always chained to *original* via ``__cause__``. 

71 

72 ``failure_metadata`` optionally overrides the target class's conservative 

73 default when the integration has backend-specific evidence about whether 

74 the failure is transient or retryable. 

75 """ 

76 if failure_metadata is not None and not isinstance( 76 ↛ 79line 76 didn't jump to line 79 because the condition on line 76 was never true

77 failure_metadata, FailureMetadata 

78 ): 

79 raise TypeError("failure_metadata must be FailureMetadata or None") 

80 

81 if _explicit_cause_parameter(kwargs) is None: 

82 parameter = _cause_parameter(target) 

83 if parameter is not None: 

84 kwargs[parameter] = original 

85 

86 exception = target(**kwargs) 

87 exception.__cause__ = original 

88 if failure_metadata is not None: 

89 exception.with_failure_metadata(failure_metadata) 

90 return exception 

91 

92 

93@contextlib.contextmanager 

94def wrapping( 

95 catch: Catchable, 

96 target: Type[DataExceptError], 

97 /, 

98 *, 

99 failure_metadata: FailureMetadata | None = None, 

100 **kwargs: Any, 

101) -> Iterator[None]: 

102 """Translate *catch* raised inside the block into *target*.""" 

103 try: 

104 yield 

105 except catch as exc: 

106 raise wrap( 

107 exc, 

108 target, 

109 failure_metadata=failure_metadata, 

110 **kwargs, 

111 ) from exc