Coverage for typed_stream/_impl/_utils.py: 98%

1# Licensed under the EUPL-1.2 or later. 

2# You may obtain a copy of the licence in all the official languages of the 

3# European Union at https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12 

4 

5"""General utility classes and functions.""" 

6 

7from __future__ import annotations 

8 

9import inspect 

10from collections.abc import Callable 

11from typing import ( 

12 Generic, 

13 Literal, 

14 NoReturn, 

15 ParamSpec, 

16 TypeGuard, 

17 TypeVar, 

18 final, 

19 overload, 

20) 

21 

22__all__ = ( 

23 "FunctionWrapperIgnoringArgs", 

24 "IndexValueTuple", 

25 "InstanceChecker", 

26 "NoneChecker", 

27 "NotNoneChecker", 

28 "count_required_positional_arguments", 

29 "raise_exception", 

30 "wrap_in_tuple", 

31) 

32 

33 

34T = TypeVar("T") 

35 

36 

37class FunctionWrapperIgnoringArgs(Generic[T]): 

38 """Wrap a function that takes no arguments.""" 

39 

40 _callable: Callable[[], T] 

41 

42 __slots__ = ("_callable",) 

43 

44 def __init__(self, fun: Callable[[], T], /) -> None: 

45 """Set the callable as attribute of self.""" 

46 self._callable = fun 

47 

48 def __call__(self, *_: object) -> T: 

49 """Call the callable while ignoring all arguments.""" 

50 return self._callable() 

51 

52 

53class IndexValueTuple(tuple[int, T], Generic[T]): 

54 """A tuple to hold index and value.""" 

55 

56 __slots__ = () 

57 

58 @property 

59 def idx(self: tuple[int, object]) -> int: 

60 """The index.""" 

61 return self[0] 

62 

63 @property 

64 def val(self: tuple[int, T]) -> T: 

65 """The value.""" 

66 return self[1] 

67 

68 

69@final 

70class InstanceChecker(Generic[T]): 

71 """Checks whether a value is an instance of a type.""" 

72 

73 _type: type[T] 

74 __doc__: str | None 

75 

76 __slots__ = ("_type",) 

77 

78 def __init__(self, type_: type[T]) -> None: 

79 """Set the type that gets checked.""" 

80 self._type = type_ 

81 

82 def __call__(self, value: object) -> TypeGuard[T]: 

83 """Check whether a value has the correct type.""" 

84 return isinstance(value, self._type) 

85 

86 

87@final 

88class NoneChecker: 

89 """Check whether a value is None.""" 

90 

91 __slots__ = () 

92 

93 @overload 

94 def __call__(self, value: None) -> Literal[True]: ... 

95 

96 @overload 

97 def __call__(self, value: object | None) -> TypeGuard[None]: ... 

98 

99 def __call__(self, value: object | None) -> bool: 

100 """Return True if the value is None.""" 

101 return value is None 

102 

103 

104@final 

105class NotNoneChecker: 

106 """Check whether a value is not None.""" 

107 

108 __slots__ = () 

109 

110 @overload 

111 def __call__(self, value: None) -> Literal[False]: ... 

112 

113 @overload 

114 def __call__(self, value: object) -> bool: ... 

115 

116 def __call__(self, value: object | None) -> bool: 

117 """Return True if the value is not None.""" 

118 return value is not None 

119 

120 

121_P = ParamSpec("_P") 

122 

123 

124def count_required_positional_arguments( 

125 fun: Callable[_P, object], / # noqa: W504 

126) -> int: 

127 """Count the required positional arguments.""" 

128 return len( 

129 [ 

130 param 

131 for param in inspect.signature(fun).parameters.values() 

132 if param.kind 

133 in { 

134 inspect.Parameter.POSITIONAL_ONLY, 

135 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

136 } 

137 if param.default == inspect.Parameter.empty 

138 ] 

139 ) 

140 

141 

142def raise_exception(exc: BaseException, /) -> NoReturn: 

143 """Raise the exception.""" 

144 raise exc 

145 

146 

147def wrap_in_tuple(value: T, /) -> tuple[T]: 

148 """Wrap the value in a single-element tuple.""" 

149 return (value,)