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

1from __future__ import annotations 

2 

3from abc import ABCMeta, abstractmethod 

4from contextlib import AbstractContextManager 

5from types import TracebackType 

6from typing import ( 

7 TYPE_CHECKING, 

8 Any, 

9 AsyncContextManager, 

10 Callable, 

11 ContextManager, 

12 Generator, 

13 Generic, 

14 Iterable, 

15 List, 

16 TypeVar, 

17 Union, 

18 overload, 

19) 

20from warnings import warn 

21 

22if TYPE_CHECKING: 

23 from ._testing import TaskInfo 

24else: 

25 TaskInfo = object 

26 

27T = TypeVar("T") 

28AnyDeprecatedAwaitable = Union[ 

29 "DeprecatedAwaitable", 

30 "DeprecatedAwaitableFloat", 

31 "DeprecatedAwaitableList[T]", 

32 TaskInfo, 

33] 

34 

35 

36@overload 

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

38 ... 

39 

40 

41@overload 

42async def maybe_async(__obj: DeprecatedAwaitableFloat) -> float: 

43 ... 

44 

45 

46@overload 

47async def maybe_async(__obj: DeprecatedAwaitableList[T]) -> list[T]: 

48 ... 

49 

50 

51@overload 

52async def maybe_async(__obj: DeprecatedAwaitable) -> None: 

53 ... 

54 

55 

56async def maybe_async( 

57 __obj: AnyDeprecatedAwaitable[T], 

58) -> TaskInfo | float | list[T] | None: 

59 """ 

60 Await on the given object if necessary. 

61 

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

63 methods were converted from coroutine functions into regular functions. 

64 

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

66 

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

68 

69 .. versionadded:: 2.2 

70 

71 """ 

72 return __obj._unwrap() 

73 

74 

75class _ContextManagerWrapper: 

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

77 self._cm = cm 

78 

79 async def __aenter__(self) -> T: 

80 return self._cm.__enter__() 

81 

82 async def __aexit__( 

83 self, 

84 exc_type: type[BaseException] | None, 

85 exc_val: BaseException | None, 

86 exc_tb: TracebackType | None, 

87 ) -> bool | None: 

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

89 

90 

91def maybe_async_cm( 

92 cm: ContextManager[T] | AsyncContextManager[T], 

93) -> AsyncContextManager[T]: 

94 """ 

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

96 

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

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

99 

100 :param cm: a regular or async context manager 

101 :return: an async context manager 

102 

103 .. versionadded:: 2.2 

104 

105 """ 

106 if not isinstance(cm, AbstractContextManager): 

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

108 

109 return _ContextManagerWrapper(cm) 

110 

111 

112def _warn_deprecation( 

113 awaitable: AnyDeprecatedAwaitable[Any], stacklevel: int = 1 

114) -> None: 

115 warn( 

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

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

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

119 DeprecationWarning, 

120 stacklevel=stacklevel + 1, 

121 ) 

122 

123 

124class DeprecatedAwaitable: 

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

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

127 

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

129 _warn_deprecation(self) 

130 if False: 

131 yield 

132 

133 def __reduce__(self) -> tuple[type[None], tuple[()]]: 

134 return type(None), () 

135 

136 def _unwrap(self) -> None: 

137 return None 

138 

139 

140class DeprecatedAwaitableFloat(float): 

141 def __new__( 

142 cls, x: float, func: Callable[..., DeprecatedAwaitableFloat] 

143 ) -> DeprecatedAwaitableFloat: 

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

145 

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

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

148 

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

150 _warn_deprecation(self) 

151 if False: 

152 yield 

153 

154 return float(self) 

155 

156 def __reduce__(self) -> tuple[type[float], tuple[float]]: 

157 return float, (float(self),) 

158 

159 def _unwrap(self) -> float: 

160 return float(self) 

161 

162 

163class DeprecatedAwaitableList(List[T]): 

164 def __init__( 

165 self, 

166 iterable: Iterable[T] = (), 

167 *, 

168 func: Callable[..., DeprecatedAwaitableList[T]], 

169 ): 

170 super().__init__(iterable) 

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

172 

173 def __await__(self) -> Generator[None, None, list[T]]: 

174 _warn_deprecation(self) 

175 if False: 

176 yield 

177 

178 return list(self) 

179 

180 def __reduce__(self) -> tuple[type[list[T]], tuple[list[T]]]: 

181 return list, (list(self),) 

182 

183 def _unwrap(self) -> list[T]: 

184 return list(self) 

185 

186 

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

188 @abstractmethod 

189 def __enter__(self) -> T: 

190 pass 

191 

192 @abstractmethod 

193 def __exit__( 

194 self, 

195 exc_type: type[BaseException] | None, 

196 exc_val: BaseException | None, 

197 exc_tb: TracebackType | None, 

198 ) -> bool | None: 

199 pass 

200 

201 async def __aenter__(self) -> T: 

202 warn( 

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

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

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

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

207 DeprecationWarning, 

208 ) 

209 return self.__enter__() 

210 

211 async def __aexit__( 

212 self, 

213 exc_type: type[BaseException] | None, 

214 exc_val: BaseException | None, 

215 exc_tb: TracebackType | None, 

216 ) -> bool | None: 

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