Coverage for typed_stream/_impl/functions.py: 100%

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"""Simple helper functions for easy Stream usage.""" 

6 

7from __future__ import annotations 

8 

9import operator 

10import typing 

11from collections.abc import Callable, Sequence 

12from numbers import Number, Real 

13from typing import Concatenate, Generic, Literal, ParamSpec, TypeVar 

14 

15from ._utils import InstanceChecker, NoneChecker, NotNoneChecker 

16 

17__all__ = ( 

18 "is_bool", 

19 "is_complex", 

20 "is_even", 

21 "is_falsy", 

22 "is_float", 

23 "is_int", 

24 "is_negative", 

25 "is_none", 

26 "is_not_none", 

27 "is_number", 

28 "is_odd", 

29 "is_positive", 

30 "is_real_number", 

31 "is_str", 

32 "is_truthy", 

33 "noop", 

34 "one", 

35 "method_partial", 

36) 

37 

38T = TypeVar("T") 

39Seq = TypeVar("Seq", bound=Sequence[object]) 

40 

41is_truthy: Callable[[object], bool] = operator.truth 

42"""Check whether a value is truthy.""" 

43 

44is_falsy: Callable[[object], bool] = operator.not_ 

45"""Check whether a value is falsy.""" 

46 

47 

48def noop(*_: object) -> None: 

49 """Do nothing.""" 

50 

51 

52def one(*_: object) -> Literal[1]: 

53 """Return the smallest positive odd number.""" 

54 return 1 

55 

56 

57def is_even(number: int) -> bool: 

58 """Check whether a number is even.""" 

59 return not number % 2 

60 

61 

62def is_odd(number: int) -> bool: 

63 """Check whether a number is odd.""" 

64 return not not number % 2 # noqa: SIM208 # pylint: disable=unneeded-not 

65 

66 

67def is_positive(number: int | float) -> bool: 

68 """Check whether a number is positive.""" 

69 return number > 0 

70 

71 

72def is_negative(number: int | float) -> bool: 

73 """Check whether a number is negative.""" 

74 return number < 0 

75 

76 

77# fmt: off 

78is_bool: InstanceChecker[bool] = InstanceChecker(bool) 

79"""Check whether a value is an instance of bool.""" 

80is_complex: InstanceChecker[complex] = InstanceChecker(complex) 

81"""Check whether a value is an instance of complex.""" 

82is_float: InstanceChecker[float] = InstanceChecker(float) 

83"""Check whether a value is an instance of float.""" 

84is_int: InstanceChecker[int] = InstanceChecker(int) 

85"""Check whether a value is an instance of int.""" 

86is_str: InstanceChecker[str] = InstanceChecker(str) 

87"""Check whether a value is an instance of str.""" 

88is_number: InstanceChecker[Number] = ( 

89 InstanceChecker(Number) # type: ignore[type-abstract] 

90) 

91"""Check whether a value is an instance of Number.""" 

92is_real_number: InstanceChecker[Real] = ( 

93 InstanceChecker(Real) # type: ignore[type-abstract] 

94) 

95"""Check whether a value is an instance of Real.""" 

96# fmt: on 

97 

98 

99is_not_none: NotNoneChecker = NotNoneChecker() 

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

101is_none: NoneChecker = NoneChecker() 

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

103 

104 

105TArg = TypeVar("TArg") # pylint: disable=invalid-name 

106TRet = TypeVar("TRet") # pylint: disable=invalid-name 

107PApplied = ParamSpec("PApplied") 

108 

109 

110# pylint: disable-next=invalid-name 

111class method_partial(Generic[TArg, TRet, PApplied]): # noqa: N801,D301 

112 r"""Pre-apply arguments to methods. 

113 

114 This is similar to functools.partial, but the returned callable just accepts 

115 one argument which gets provided first positional argument to the wrapped 

116 function. 

117 It has similarities to operator.methodcaller, but it's type-safe. 

118 This is intended to be used with methods that don't support keyword args. 

119 

120 >>> from typed_stream import Stream 

121 >>> from operator import mod 

122 >>> d = "abc\n# comment, please ignore\nxyz".split("\n") 

123 >>> Stream(d).exclude(method_partial(str.startswith, "#")).for_each(print) 

124 abc 

125 xyz 

126 >>> Stream.range(10).exclude(method_partial(int.__mod__, 3)).collect() 

127 (0, 3, 6, 9) 

128 >>> Stream.range(10).exclude(method_partial(mod, 3)).collect() 

129 (0, 3, 6, 9) 

130 """ 

131 

132 _fun: Callable[Concatenate[TArg, PApplied], TRet] 

133 # PApplied.args 

134 _args: typing.Any # type: ignore[explicit-any] 

135 # PApplied.kwargs 

136 _kwargs: typing.Any # type: ignore[explicit-any] 

137 

138 __slots__ = ("_fun", "_args", "_kwargs") 

139 

140 def __init__( 

141 self, 

142 fun: Callable[Concatenate[TArg, PApplied], TRet], 

143 /, 

144 *args: PApplied.args, 

145 **kwargs: PApplied.kwargs, 

146 ) -> None: 

147 """Initialize self.""" 

148 self._fun = fun 

149 self._args = args 

150 self._kwargs = kwargs 

151 

152 def __call__(self, arg: TArg, /) -> TRet: 

153 """Call the wrapped function.""" 

154 return self._fun(arg, *self._args, **self._kwargs)