Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/anyio/_core/_compat.py: 57%

1from abc import ABCMeta, abstractmethod 

2from contextlib import AbstractContextManager 

3from types import TracebackType 

4from typing import ( 

5 TYPE_CHECKING, 

6 Any, 

7 AsyncContextManager, 

8 Callable, 

9 ContextManager, 

10 Generator, 

11 Generic, 

12 Iterable, 

13 List, 

14 Optional, 

15 Tuple, 

16 Type, 

17 TypeVar, 

18 Union, 

19 overload, 

20) 

21from warnings import warn 

22 

23if TYPE_CHECKING: 

24 from ._testing import TaskInfo 

25else: 

26 TaskInfo = object 

27 

28T = TypeVar("T") 

29AnyDeprecatedAwaitable = Union[ 

30 "DeprecatedAwaitable", 

31 "DeprecatedAwaitableFloat", 

32 "DeprecatedAwaitableList[T]", 

33 TaskInfo, 

34] 

35 

36 

37@overload 

38async def maybe_async(__obj: TaskInfo) -> TaskInfo: 

39 ... 

40 

41 

42@overload 

43async def maybe_async(__obj: "DeprecatedAwaitableFloat") -> float: 

44 ... 

45 

46 

47@overload 

48async def maybe_async(__obj: "DeprecatedAwaitableList[T]") -> List[T]: 

49 ... 

50 

51 

52@overload 

53async def maybe_async(__obj: "DeprecatedAwaitable") -> None: 

54 ... 

55 

56 

57async def maybe_async( 

58 __obj: "AnyDeprecatedAwaitable[T]", 

59) -> Union[TaskInfo, float, List[T], None]: 

60 """ 

61 Await on the given object if necessary. 

62 

63 This function is intended to bridge the gap between AnyIO 2.x and 3.x where some functions and 

64 methods were converted from coroutine functions into regular functions. 

65 

66 Do **not** try to use this for any other purpose! 

67 

68 :return: the result of awaiting on the object if coroutine, or the object itself otherwise 

69 

70 .. versionadded:: 2.2 

71 

72 """ 

73 return __obj._unwrap() 

74 

75 

76class _ContextManagerWrapper: 

77 def __init__(self, cm: ContextManager[T]): 

78 self._cm = cm 

79 

80 async def __aenter__(self) -> T: 

81 return self._cm.__enter__() 

82 

83 async def __aexit__( 

84 self, 

85 exc_type: Optional[Type[BaseException]], 

86 exc_val: Optional[BaseException], 

87 exc_tb: Optional[TracebackType], 

88 ) -> Optional[bool]: 

89 return self._cm.__exit__(exc_type, exc_val, exc_tb) 

90 

91 

92def maybe_async_cm( 

93 cm: Union[ContextManager[T], AsyncContextManager[T]] 

94) -> AsyncContextManager[T]: 

95 """ 

96 Wrap a regular context manager as an async one if necessary. 

97 

98 This function is intended to bridge the gap between AnyIO 2.x and 3.x where some functions and 

99 methods were changed to return regular context managers instead of async ones. 

100 

101 :param cm: a regular or async context manager 

102 :return: an async context manager 

103 

104 .. versionadded:: 2.2 

105 

106 """ 

107 if not isinstance(cm, AbstractContextManager): 

108 raise TypeError("Given object is not an context manager") 

109 

110 return _ContextManagerWrapper(cm) 

111 

112 

113def _warn_deprecation( 

114 awaitable: "AnyDeprecatedAwaitable[Any]", stacklevel: int = 1 

115) -> None: 

116 warn( 

117 f'Awaiting on {awaitable._name}() is deprecated. Use "await ' 

118 f"anyio.maybe_async({awaitable._name}(...)) if you have to support both AnyIO 2.x " 

119 f'and 3.x, or just remove the "await" if you are completely migrating to AnyIO 3+.', 

120 DeprecationWarning, 

121 stacklevel=stacklevel + 1, 

122 ) 

123 

124 

125class DeprecatedAwaitable: 

126 def __init__(self, func: Callable[..., "DeprecatedAwaitable"]): 

127 self._name = f"{func.__module__}.{func.__qualname__}" 

128 

129 def __await__(self) -> Generator[None, None, None]: 

130 _warn_deprecation(self) 

131 if False: 

132 yield 

133 

134 def __reduce__(self) -> Tuple[Type[None], Tuple[()]]: 

135 return type(None), () 

136 

137 def _unwrap(self) -> None: 

138 return None 

139 

140 

141class DeprecatedAwaitableFloat(float): 

142 def __new__( 

143 cls, x: float, func: Callable[..., "DeprecatedAwaitableFloat"] 

144 ) -> "DeprecatedAwaitableFloat": 

145 return super().__new__(cls, x) 

146 

147 def __init__(self, x: float, func: Callable[..., "DeprecatedAwaitableFloat"]): 

148 self._name = f"{func.__module__}.{func.__qualname__}" 

149 

150 def __await__(self) -> Generator[None, None, float]: 

151 _warn_deprecation(self) 

152 if False: 

153 yield 

154 

155 return float(self) 

156 

157 def __reduce__(self) -> Tuple[Type[float], Tuple[float]]: 

158 return float, (float(self),) 

159 

160 def _unwrap(self) -> float: 

161 return float(self) 

162 

163 

164class DeprecatedAwaitableList(List[T]): 

165 def __init__( 

166 self, 

167 iterable: Iterable[T] = (), 

168 *, 

169 func: Callable[..., "DeprecatedAwaitableList[T]"], 

170 ): 

171 super().__init__(iterable) 

172 self._name = f"{func.__module__}.{func.__qualname__}" 

173 

174 def __await__(self) -> Generator[None, None, List[T]]: 

175 _warn_deprecation(self) 

176 if False: 

177 yield 

178 

179 return list(self) 

180 

181 def __reduce__(self) -> Tuple[Type[List[T]], Tuple[List[T]]]: 

182 return list, (list(self),) 

183 

184 def _unwrap(self) -> List[T]: 

185 return list(self) 

186 

187 

188class DeprecatedAsyncContextManager(Generic[T], metaclass=ABCMeta): 

189 @abstractmethod 

190 def __enter__(self) -> T: 

191 pass 

192 

193 @abstractmethod 

194 def __exit__( 

195 self, 

196 exc_type: Optional[Type[BaseException]], 

197 exc_val: Optional[BaseException], 

198 exc_tb: Optional[TracebackType], 

199 ) -> Optional[bool]: 

200 pass 

201 

202 async def __aenter__(self) -> T: 

203 warn( 

204 f"Using {self.__class__.__name__} as an async context manager has been deprecated. " 

205 f'Use "async with anyio.maybe_async_cm(yourcontextmanager) as foo:" if you have to ' 

206 f'support both AnyIO 2.x and 3.x, or just remove the "async" from "async with" if ' 

207 f"you are completely migrating to AnyIO 3+.", 

208 DeprecationWarning, 

209 ) 

210 return self.__enter__() 

211 

212 async def __aexit__( 

213 self, 

214 exc_type: Optional[Type[BaseException]], 

215 exc_val: Optional[BaseException], 

216 exc_tb: Optional[TracebackType], 

217 ) -> Optional[bool]: 

218 return self.__exit__(exc_type, exc_val, exc_tb)