File size: 5,726 Bytes
b6068b4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 |
from __future__ import annotations
from abc import ABCMeta, abstractmethod
from contextlib import AbstractContextManager
from types import TracebackType
from typing import (
TYPE_CHECKING,
Any,
AsyncContextManager,
Callable,
ContextManager,
Generator,
Generic,
Iterable,
List,
TypeVar,
Union,
overload,
)
from warnings import warn
if TYPE_CHECKING:
from ._testing import TaskInfo
else:
TaskInfo = object
T = TypeVar("T")
AnyDeprecatedAwaitable = Union[
"DeprecatedAwaitable",
"DeprecatedAwaitableFloat",
"DeprecatedAwaitableList[T]",
TaskInfo,
]
@overload
async def maybe_async(__obj: TaskInfo) -> TaskInfo:
...
@overload
async def maybe_async(__obj: DeprecatedAwaitableFloat) -> float:
...
@overload
async def maybe_async(__obj: DeprecatedAwaitableList[T]) -> list[T]:
...
@overload
async def maybe_async(__obj: DeprecatedAwaitable) -> None:
...
async def maybe_async(
__obj: AnyDeprecatedAwaitable[T],
) -> TaskInfo | float | list[T] | None:
"""
Await on the given object if necessary.
This function is intended to bridge the gap between AnyIO 2.x and 3.x where some functions and
methods were converted from coroutine functions into regular functions.
Do **not** try to use this for any other purpose!
:return: the result of awaiting on the object if coroutine, or the object itself otherwise
.. versionadded:: 2.2
"""
return __obj._unwrap()
class _ContextManagerWrapper:
def __init__(self, cm: ContextManager[T]):
self._cm = cm
async def __aenter__(self) -> T:
return self._cm.__enter__()
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> bool | None:
return self._cm.__exit__(exc_type, exc_val, exc_tb)
def maybe_async_cm(
cm: ContextManager[T] | AsyncContextManager[T],
) -> AsyncContextManager[T]:
"""
Wrap a regular context manager as an async one if necessary.
This function is intended to bridge the gap between AnyIO 2.x and 3.x where some functions and
methods were changed to return regular context managers instead of async ones.
:param cm: a regular or async context manager
:return: an async context manager
.. versionadded:: 2.2
"""
if not isinstance(cm, AbstractContextManager):
raise TypeError("Given object is not an context manager")
return _ContextManagerWrapper(cm)
def _warn_deprecation(
awaitable: AnyDeprecatedAwaitable[Any], stacklevel: int = 1
) -> None:
warn(
f'Awaiting on {awaitable._name}() is deprecated. Use "await '
f"anyio.maybe_async({awaitable._name}(...)) if you have to support both AnyIO 2.x "
f'and 3.x, or just remove the "await" if you are completely migrating to AnyIO 3+.',
DeprecationWarning,
stacklevel=stacklevel + 1,
)
class DeprecatedAwaitable:
def __init__(self, func: Callable[..., DeprecatedAwaitable]):
self._name = f"{func.__module__}.{func.__qualname__}"
def __await__(self) -> Generator[None, None, None]:
_warn_deprecation(self)
if False:
yield
def __reduce__(self) -> tuple[type[None], tuple[()]]:
return type(None), ()
def _unwrap(self) -> None:
return None
class DeprecatedAwaitableFloat(float):
def __new__(
cls, x: float, func: Callable[..., DeprecatedAwaitableFloat]
) -> DeprecatedAwaitableFloat:
return super().__new__(cls, x)
def __init__(self, x: float, func: Callable[..., DeprecatedAwaitableFloat]):
self._name = f"{func.__module__}.{func.__qualname__}"
def __await__(self) -> Generator[None, None, float]:
_warn_deprecation(self)
if False:
yield
return float(self)
def __reduce__(self) -> tuple[type[float], tuple[float]]:
return float, (float(self),)
def _unwrap(self) -> float:
return float(self)
class DeprecatedAwaitableList(List[T]):
def __init__(
self,
iterable: Iterable[T] = (),
*,
func: Callable[..., DeprecatedAwaitableList[T]],
):
super().__init__(iterable)
self._name = f"{func.__module__}.{func.__qualname__}"
def __await__(self) -> Generator[None, None, list[T]]:
_warn_deprecation(self)
if False:
yield
return list(self)
def __reduce__(self) -> tuple[type[list[T]], tuple[list[T]]]:
return list, (list(self),)
def _unwrap(self) -> list[T]:
return list(self)
class DeprecatedAsyncContextManager(Generic[T], metaclass=ABCMeta):
@abstractmethod
def __enter__(self) -> T:
pass
@abstractmethod
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> bool | None:
pass
async def __aenter__(self) -> T:
warn(
f"Using {self.__class__.__name__} as an async context manager has been deprecated. "
f'Use "async with anyio.maybe_async_cm(yourcontextmanager) as foo:" if you have to '
f'support both AnyIO 2.x and 3.x, or just remove the "async" from "async with" if '
f"you are completely migrating to AnyIO 3+.",
DeprecationWarning,
)
return self.__enter__()
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> bool | None:
return self.__exit__(exc_type, exc_val, exc_tb)
|