Coverage for typed_stream/_impl/_types.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"""Helpful types.""" 

6 

7from __future__ import annotations 

8 

9import abc 

10from abc import abstractmethod 

11from collections.abc import Callable, Iterable, Iterator 

12from os import PathLike 

13from typing import Generic, Protocol, TypeAlias, TypeGuard, TypeVar 

14 

15from ._typing import Self, override 

16 

17__all__ = ( 

18 "ClassWithCleanUp", 

19 "Closeable", 

20 "IteratorProxy", 

21 "PathLikeType", 

22 "PrettyRepr", 

23 "StarCallable", 

24 "SupportsAdd", 

25 "SupportsComparison", 

26 "SupportsGreaterThan", 

27 "SupportsLessThan", 

28 "TypeGuardingCallable", 

29) 

30 

31PathLikeType = bytes | PathLike[bytes] | PathLike[str] | str 

32 

33T = TypeVar("T") 

34V = TypeVar("V") 

35T_co = TypeVar("T_co", covariant=True) 

36V_contra = TypeVar("V_contra", contravariant=True) 

37 

38 

39class TypeGuardingCallable(Protocol[T_co, V_contra]): 

40 """A class representing a function that type guards.""" 

41 

42 @abstractmethod 

43 def __call__(self, value: V_contra) -> TypeGuard[T_co]: 

44 """Return True if value isinstance of T_co.""" 

45 

46 

47# pylint: disable=invalid-name 

48SC_IN_contra = TypeVar("SC_IN_contra", contravariant=True) 

49SC_OUT_co = TypeVar("SC_OUT_co", covariant=True) 

50# pylint: enable=invalid-name 

51 

52 

53class StarCallable(Protocol[SC_IN_contra, SC_OUT_co]): 

54 """A class representing a function, that takes many arguments.""" 

55 

56 @abstractmethod 

57 def __call__(self, *args: SC_IN_contra) -> SC_OUT_co: 

58 """Handle the arguments.""" 

59 

60 

61class SupportsLessThan(Protocol): 

62 """A class that supports comparison with less than.""" 

63 

64 @abstractmethod 

65 def __lt__(self: T, other: T) -> bool: 

66 """Compare to another instance of the same type.""" 

67 

68 

69class SupportsGreaterThan(Protocol): 

70 """A class that supports comparison with less than.""" 

71 

72 @abstractmethod 

73 def __gt__(self: T, other: T) -> bool: 

74 """Compare to another instance of the same type.""" 

75 

76 

77SupportsComparison: TypeAlias = SupportsGreaterThan | SupportsLessThan 

78 

79 

80class SupportsAdd(Protocol): 

81 """A class that supports addition.""" 

82 

83 @abstractmethod 

84 def __add__(self: T, other: T) -> T: 

85 """Add another instance of the same type to self.""" 

86 

87 

88class Closeable(abc.ABC): 

89 """Class that can be closed.""" 

90 

91 __slots__ = () 

92 

93 @abc.abstractmethod 

94 def close(self) -> None: 

95 """Run clean-up if not run yet.""" 

96 

97 def __del__(self) -> None: 

98 """Run close.""" 

99 self.close() 

100 

101 def __enter__(self: T) -> T: 

102 """Return self.""" 

103 return self 

104 

105 def __exit__(self, *args: object) -> None: 

106 """Run close.""" 

107 self.close() 

108 

109 

110class PrettyRepr(abc.ABC): 

111 """ABC to inherit from to get a better repr.""" 

112 

113 __slots__ = () 

114 

115 @classmethod 

116 def _module(cls) -> str: 

117 return cls.__module__ 

118 

119 @override 

120 def __repr__(self) -> str: 

121 """Return the string representation of self.""" 

122 args = ",".join( 

123 [("..." if arg is ... else repr(arg)) for arg in self._get_args()] 

124 ) 

125 return ( 

126 f"{self.__class__._module()}.{self.__class__.__qualname__}({args})" 

127 ) 

128 

129 @abc.abstractmethod 

130 def _get_args(self) -> tuple[object, ...]: # pragma: no cover 

131 """Return the args used to initializing self.""" 

132 

133 

134class ClassWithCleanUp(Closeable, PrettyRepr): 

135 """A class that has a cleanup_fun and a close method.""" 

136 

137 cleanup_fun: Callable[[], object | None] | None 

138 

139 __slots__ = ("cleanup_fun",) 

140 

141 def __init__(self, cleanup_fun: Callable[[], object | None]) -> None: 

142 """Initialize this class.""" 

143 self.cleanup_fun = cleanup_fun 

144 

145 @override 

146 def _get_args(self) -> tuple[object, ...]: 

147 """Return the args used to initializing self.""" 

148 return (self.cleanup_fun,) 

149 

150 @override 

151 def close(self) -> None: 

152 """Run clean-up if not run yet.""" 

153 if self.cleanup_fun: 

154 self.cleanup_fun() 

155 self.cleanup_fun = None 

156 

157 

158class IteratorProxy(Iterator[V], Generic[V, T], PrettyRepr, abc.ABC): 

159 """Proxy an iterator.""" 

160 

161 _iterator: Iterator[T] 

162 __slots__ = ("_iterator",) 

163 

164 def __init__(self, iterable: Iterable[T]) -> None: 

165 """Init self.""" 

166 self._iterator = iter(iterable) 

167 

168 @override 

169 def __iter__(self) -> Self: 

170 """Return self.""" 

171 return self 

172 

173 @override 

174 @abc.abstractmethod 

175 def __next__(self) -> V: 

176 """Return the next element.""" 

177 

178 @override 

179 def _get_args(self) -> tuple[object, ...]: 

180 """Return the args used to initializing self.""" 

181 return (self._iterator,)