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

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 

10import itertools 

11import sys 

12from collections.abc import Callable, Iterable 

13from typing import ( 

14 Generic, 

15 Literal, 

16 NoReturn, 

17 ParamSpec, 

18 TypeGuard, 

19 TypeVar, 

20 final, 

21 overload, 

22) 

23 

24from ._typing import TypeVarTuple, Unpack 

25 

26__all__ = ( 

27 "FunctionWrapperIgnoringArgs", 

28 "IndexValueTuple", 

29 "InstanceChecker", 

30 "NoneChecker", 

31 "NotNoneChecker", 

32 "count_required_positional_arguments", 

33 "map_with_additional_args", 

34 "raise_exception", 

35 "wrap_in_tuple", 

36) 

37 

38 

39T = TypeVar("T") 

40U = TypeVar("U") 

41Tvt = TypeVarTuple("Tvt") 

42 

43 

44class FunctionWrapperIgnoringArgs(Generic[T]): 

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

46 

47 _callable: Callable[[], T] 

48 

49 __slots__ = ("_callable",) 

50 

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

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

53 self._callable = fun 

54 

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

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

57 return self._callable() 

58 

59 

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

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

62 

63 __slots__ = () 

64 

65 @property 

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

67 """The index.""" 

68 return self[0] 

69 

70 @property 

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

72 """The value.""" 

73 return self[1] 

74 

75 

76@final 

77class InstanceChecker(Generic[T]): 

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

79 

80 _type: type[T] 

81 __doc__: str | None 

82 

83 __slots__ = ("_type",) 

84 

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

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

87 self._type = type_ 

88 

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

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

91 return isinstance(value, self._type) 

92 

93 

94@final 

95class NoneChecker: 

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

97 

98 __slots__ = () 

99 

100 @overload 

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

102 

103 @overload 

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

105 

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

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

108 return value is None 

109 

110 

111@final 

112class NotNoneChecker: 

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

114 

115 __slots__ = () 

116 

117 @overload 

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

119 

120 @overload 

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

122 

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

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

125 return value is not None 

126 

127 

128_P = ParamSpec("_P") 

129 

130 

131def count_required_positional_arguments( 

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

133) -> int: 

134 """Count the required positional arguments.""" 

135 return len( 

136 [ 

137 param 

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

139 if param.kind 

140 in { 

141 inspect.Parameter.POSITIONAL_ONLY, 

142 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

143 } 

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

145 ] 

146 ) 

147 

148 

149if sys.version_info >= (3, 14): 

150 

151 def map_with_additional_args( 

152 iterable: Iterable[T], 

153 fun: Callable[[T, Unpack[Tvt]], U], 

154 args: tuple[Unpack[Tvt]], 

155 /, 

156 ) -> Iterable[U]: 

157 """Map with constant additional arguments for each value.""" 

158 return map( 

159 fun, 

160 iterable, 

161 *(itertools.repeat(arg) for arg in args), 

162 strict=False, 

163 ) 

164 

165else: 

166 

167 def map_with_additional_args( 

168 iterable: Iterable[T], 

169 fun: Callable[[T, Unpack[Tvt]], U], 

170 args: tuple[Unpack[Tvt]], 

171 /, 

172 ) -> Iterable[U]: 

173 """Map with constant additional arguments for each value.""" 

174 return map( # noqa: B912 

175 fun, iterable, *(itertools.repeat(arg) for arg in args) 

176 ) 

177 

178 

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

180 """Raise the exception.""" 

181 raise exc 

182 

183 

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

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

186 return (value,)