This commit is contained in:
DevilXD committed 2022-02-15 20:55:57 +01:00
1 parent 763bc0a536
commit c112003a38
1 file changed
+17 -17
+17 -17
View File
@@ -35,7 +35,7 @@ else:
pass
_V = TypeVar("_V") # value
_T = TypeVar("_T") # type
_D = TypeVar("_D") # default
_P = ParamSpec("_P") # params
logger = logging.getLogger("TwitchDrops")
@@ -50,13 +50,13 @@ def create_nonce(length: int = 30) -> str:
return ''.join(random.choices(NONCE_CHARS, k=length))
def deduplicate(iterable: Iterable[_V]) -> List[_V]:
def deduplicate(iterable: Iterable[_T]) -> List[_T]:
return list(OrderedDict.fromkeys(iterable).keys())
def task_wrapper(
afunc: Callable[_P, Coroutine[Any, Any, _V]]
) -> Callable[_P, Coroutine[Any, Any, _V]]:
afunc: Callable[_P, Coroutine[Any, Any, _T]]
) -> Callable[_P, Coroutine[Any, Any, _T]]:
@wraps(afunc)
async def wrapper(*args: _P.args, **kwargs: _P.kwargs):
try:
@@ -76,13 +76,13 @@ def invalidate_cache(instance, *attrnames):
delattr(instance, name)
class OrderedSet(MutableSet[_V]):
class OrderedSet(MutableSet[_T]):
"""
Implementation of a set that preserves insertion order,
based on OrderedDict with values set to None.
"""
def __init__(self, iterable: Iterable[_V] = [], /):
self._items: OrderedDict[_V, None] = OrderedDict((item, None) for item in iterable)
def __init__(self, iterable: Iterable[_T] = [], /):
self._items: OrderedDict[_T, None] = OrderedDict((item, None) for item in iterable)
def __repr__(self) -> str:
return f"{self.__class__.__name__}([{', '.join(map(repr, self._items))}])"
@@ -90,35 +90,35 @@ class OrderedSet(MutableSet[_V]):
def __contains__(self, item: object, /) -> bool:
return item in self._items
def __iter__(self) -> Iterator[_V]:
def __iter__(self) -> Iterator[_T]:
return iter(self._items)
def __len__(self) -> int:
return len(self._items)
def add(self, item: _V, /) -> None:
def add(self, item: _T, /) -> None:
self._items[item] = None
def discard(self, item: _V, /) -> None:
def discard(self, item: _T, /) -> None:
with suppress(KeyError):
del self._items[item]
def update(self, *others: Iterable[_V]) -> None:
def update(self, *others: Iterable[_T]) -> None:
for it in others:
for item in it:
if item not in self._items:
self._items[item] = None
def difference_update(self, *others: Iterable[_V]) -> None:
def difference_update(self, *others: Iterable[_T]) -> None:
for it in others:
for item in it:
if item in self._items:
del self._items[item]
class AwaitableValue(Generic[_V]):
class AwaitableValue(Generic[_T]):
def __init__(self):
self._value: _V
self._value: _T
self._event = asyncio.Event()
def has_value(self) -> bool:
@@ -127,16 +127,16 @@ class AwaitableValue(Generic[_V]):
def wait(self) -> Coroutine[Any, Any, Literal[True]]:
return cast(Coroutine[Any, Any, Literal[True]], self._event.wait())
def get_with_default(self, default: _D) -> Union[_D, _V]:
def get_with_default(self, default: _D) -> Union[_T, _D]:
if self._event.is_set():
return self._value
return default
async def get(self) -> _V:
async def get(self) -> _T:
await self._event.wait()
return self._value
def set(self, value: _V) -> None:
def set(self, value: _T) -> None:
self._value = value
self._event.set()