diff --git a/utils.py b/utils.py index c428808..0583d49 100644 --- a/utils.py +++ b/utils.py @@ -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()