blob: 384db76f8bc4324b4765dd0e9ad5fa2fbf65650d (
plain)
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
|
from __future__ import annotations
from typing import TYPE_CHECKING, TypeVar
from litestar.types import Empty, EmptyType
__all__ = ("async_next",)
if TYPE_CHECKING:
from typing import Any, AsyncGenerator
T = TypeVar("T")
D = TypeVar("D")
try:
async_next = anext # type: ignore[name-defined]
except NameError:
async def async_next(gen: AsyncGenerator[T, Any], default: D | EmptyType = Empty) -> T | D:
"""Backwards compatibility shim for Python<3.10."""
try:
return await gen.__anext__()
except StopAsyncIteration as exc:
if default is not Empty:
return default
raise exc
|
