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 

47ScIn_contra = TypeVar("ScIn_contra", contravariant=True) 

48ScOut_co = TypeVar("ScOut_co", covariant=True) 

49 

50 

51class StarCallable(Protocol[ScIn_contra, ScOut_co]): 

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

53 

54 @abstractmethod 

55 def __call__(self, *args: ScIn_contra) -> ScOut_co: 

56 """Handle the arguments.""" 

57 

58 

59class SupportsLessThan(Protocol): 

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

61 

62 @abstractmethod 

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

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

65 

66 

67class SupportsGreaterThan(Protocol): 

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

69 

70 @abstractmethod 

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

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

73 

74 

75SupportsComparison: TypeAlias = SupportsGreaterThan | SupportsLessThan 

76 

77 

78class SupportsAdd(Protocol): 

79 """A class that supports addition.""" 

80 

81 @abstractmethod 

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

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

84 

85 

86class Closeable(abc.ABC): 

87 """Class that can be closed.""" 

88 

89 __slots__ = () 

90 

91 @abc.abstractmethod 

92 def close(self) -> None: 

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

94 

95 def __del__(self) -> None: 

96 """Run close.""" 

97 self.close() 

98 

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

100 """Return self.""" 

101 return self 

102 

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

104 """Run close.""" 

105 self.close() 

106 

107 

108class PrettyRepr(abc.ABC): 

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

110 

111 __slots__ = () 

112 

113 @classmethod 

114 def _module(cls) -> str: 

115 return cls.__module__ 

116 

117 @override 

118 def __repr__(self) -> str: 

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

120 args = ",".join( 

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

122 ) 

123 return ( 

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

125 ) 

126 

127 @abc.abstractmethod 

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

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

130 

131 

132class ClassWithCleanUp(Closeable, PrettyRepr): 

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

134 

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

136 

137 __slots__ = ("cleanup_fun",) 

138 

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

140 """Initialize this class.""" 

141 self.cleanup_fun = cleanup_fun 

142 

143 @override 

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

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

146 return (self.cleanup_fun,) 

147 

148 @override 

149 def close(self) -> None: 

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

151 if self.cleanup_fun: 

152 self.cleanup_fun() 

153 self.cleanup_fun = None 

154 

155 

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

157 """Proxy an iterator.""" 

158 

159 _iterator: Iterator[T] 

160 __slots__ = ("_iterator",) 

161 

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

163 """Init self.""" 

164 self._iterator = iter(iterable) 

165 

166 @override 

167 def __iter__(self) -> Self: 

168 """Return self.""" 

169 return self 

170 

171 @override 

172 @abc.abstractmethod 

173 def __next__(self) -> V: 

174 """Return the next element.""" 

175 

176 @override 

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

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

179 return (self._iterator,)