Switch to 80 maxline now that I'm using 3.12 type aliases

Author: mikeshardmind

.editorconfig

@@ -9,11 +9,11 @@ indent_size = 4
 
 [*.py]
 indent_style = space
-max_line_length = 88
+max_line_length = 90
 
 [*.pyi]
 indent_style = space
-max_line_length = 88
+max_line_length = 90
 
 [*.md]
 indent_style = space

async_utils/_cpython_stuff.py

@@ -21,12 +21,13 @@
 class _HashedSeq(list[Any]):
     """This class guarantees that hash() will be called no more than once
     per element.  This is important because the lru_cache() will hash
-    the key multiple times on a cache miss.
-    """
+    the key multiple times on a cache miss."""
 
     __slots__ = ("hashvalue",)
 
-    def __init__(self, tup: tuple[Any, ...], hash: Callable[[object], int] = hash):  # noqa: A002
+    def __init__(
+        self, tup: tuple[Any, ...], hash: Callable[[object], int] = hash
+    ):  # noqa: A002
         self[:] = tup
         self.hashvalue: int = hash(tup)
 
@@ -47,8 +48,7 @@ def make_key(
     as a nested structure that would take more memory.
     If there is only a single argument and its data type is known to cache
     its hash value, then that argument is returned without a wrapper.  This
-    saves space and improves lookup speed.
-    """
+    saves space and improves lookup speed."""
     # All of code below relies on kwds preserving the order input by the user.
     # Formerly, we sorted() the kwds before looping.  The new way is *much*
     # faster; however, it means that f(x=1, y=2) will now be treated as a

async_utils/_lru.py

@@ -0,0 +1,41 @@
+#   Copyright 2020-present Michael Hall
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+
+from __future__ import annotations
+
+__all__ = ("LRU",)
+
+
+class LRU[K, V]:
+    def __init__(self, maxsize: int, /):
+        self.cache: dict[K, V] = {}
+        self.maxsize = maxsize
+
+    def get[T](self, key: K, default: T, /) -> V | T:
+        try:
+            return self[key]
+        except KeyError:
+            return default
+
+    def __getitem__(self, key: K, /) -> V:
+        val = self.cache[key] = self.cache.pop(key)
+        return val
+
+    def __setitem__(self, key: K, value: V, /):
+        self.cache[key] = value
+        if len(self.cache) > self.maxsize:
+            self.cache.pop(next(iter(self.cache)))
+
+    def remove(self, key: K, /) -> None:
+        self.cache.pop(key, None)

async_utils/bg_loop.py

@@ -47,15 +47,19 @@ async def run(self, coro: _FutureLike[_T]) -> _T:
         return await asyncio.wrap_future(future)
 
 
-def run_forever(loop: asyncio.AbstractEventLoop, use_eager_task_factory: bool, /) -> None:
+def run_forever(
+    loop: asyncio.AbstractEventLoop, use_eager_task_factory: bool, /
+) -> None:
     asyncio.set_event_loop(loop)
     if use_eager_task_factory:
         loop.set_task_factory(asyncio.eager_task_factory)
     try:
         loop.run_forever()
     finally:
         loop.run_until_complete(asyncio.sleep(0))
-        tasks: set[asyncio.Task[Any]] = {t for t in asyncio.all_tasks(loop) if not t.done()}
+        tasks: set[asyncio.Task[Any]] = {
+            t for t in asyncio.all_tasks(loop) if not t.done()
+        }
         for t in tasks:
             t.cancel()
         loop.run_until_complete(asyncio.sleep(0))
@@ -80,7 +84,9 @@ def run_forever(loop: asyncio.AbstractEventLoop, use_eager_task_factory: bool, /
 
 
 @contextmanager
-def threaded_loop(*, use_eager_task_factory: bool = True) -> Generator[LoopWrapper, None, None]:
+def threaded_loop(
+    *, use_eager_task_factory: bool = True
+) -> Generator[LoopWrapper, None, None]:
     """Starts an event loop on a background thread,
     and yields an object with scheduling methods for interacting with
     the loop.
@@ -89,7 +95,9 @@ def threaded_loop(*, use_eager_task_factory: bool = True) -> Generator[LoopWrapp
     loop = asyncio.new_event_loop()
     thread = None
     try:
-        thread = threading.Thread(target=run_forever, args=(loop, use_eager_task_factory))
+        thread = threading.Thread(
+            target=run_forever, args=(loop, use_eager_task_factory)
+        )
         thread.start()
         yield LoopWrapper(loop)
     finally:

async_utils/bg_tasks.py

@@ -33,7 +33,13 @@ def __init__(self, exit_timeout: float | None) -> None:
         self._tasks: set[asyncio.Task[Any]] = set()
         self._exit_timeout: float | None = exit_timeout
 
-    def create_task(self, coro: _CoroutineLike[_T], *, name: str | None = None, context: Context | None = None) -> Any:
+    def create_task(
+        self,
+        coro: _CoroutineLike[_T],
+        *,
+        name: str | None = None,
+        context: Context | None = None,
+    ) -> Any:
         t = asyncio.create_task(coro)
         self._tasks.add(t)
         t.add_done_callback(self._tasks.discard)
@@ -44,7 +50,9 @@ async def __aenter__(self: Self) -> Self:
 
     async def __aexit__(self, *_dont_care: Any):
         while tsks := self._tasks.copy():
-            _done, _pending = await asyncio.wait(tsks, timeout=self._exit_timeout)
+            _done, _pending = await asyncio.wait(
+                tsks, timeout=self._exit_timeout
+            )
             for task in _pending:
                 task.cancel()
             await asyncio.sleep(0)

async_utils/corofunc_cache.py

@@ -20,65 +20,45 @@
 from typing import Any, ParamSpec, TypeVar
 
 from ._cpython_stuff import make_key
+from ._lru import LRU
 
 __all__ = ("corocache", "lrucorocache")
 
 
 P = ParamSpec("P")
-T = TypeVar("T")
+R = TypeVar("R")
 
 
-type CoroFunc[**P, T] = Callable[P, Coroutine[Any, Any, T]]
-type CoroLike[**P, T] = Callable[P, Awaitable[T]]
-
-
-class LRU[K, V]:
-    def __init__(self, maxsize: int, /):
-        self.cache: dict[K, V] = {}
-        self.maxsize = maxsize
-
-    def get(self, key: K, default: T, /) -> V | T:
-        if key not in self.cache:
-            return default
-        self.cache[key] = self.cache.pop(key)
-        return self.cache[key]
-
-    def __getitem__(self, key: K, /) -> V:
-        self.cache[key] = self.cache.pop(key)
-        return self.cache[key]
-
-    def __setitem__(self, key: K, value: V, /):
-        self.cache[key] = value
-        if len(self.cache) > self.maxsize:
-            self.cache.pop(next(iter(self.cache)))
-
-    def remove(self, key: K, /) -> None:
-        self.cache.pop(key, None)
+type CoroFunc[**P, R] = Callable[P, Coroutine[Any, Any, R]]
+type CoroLike[**P, R] = Callable[P, Awaitable[R]]
 
 
 def corocache(
     ttl: float | None = None,
-) -> Callable[[CoroLike[P, T]], CoroFunc[P, T]]:
+) -> Callable[[CoroLike[P, R]], CoroFunc[P, R]]:
     """Decorator to cache coroutine functions.
 
-    This is less powerful than the version in task_cache.py but may work better for
-    some cases where typing of libraries this interacts with is too restrictive.
+    This is less powerful than the version in task_cache.py but may work better
+    for some cases where typing of libraries this interacts with is too restrictive.
 
-    Note: This uses the args and kwargs of the original coroutine function as a cache key.
-    This includes instances (self) when wrapping methods.
-    Consider not wrapping instance methods, but what those methods call when feasible in cases where this may matter.
+    Note: This uses the args and kwargs of the original coroutine function as a
+    cache key. This includes instances (self) when wrapping methods.
+    Consider not wrapping instance methods, but what those methods call when feasible
+    in cases where this may matter.
 
     The ordering of args and kwargs matters."""
 
-    def wrapper(coro: CoroLike[P, T]) -> CoroFunc[P, T]:
-        internal_cache: dict[Hashable, asyncio.Future[T]] = {}
+    def wrapper(coro: CoroLike[P, R]) -> CoroFunc[P, R]:
+        internal_cache: dict[Hashable, asyncio.Future[R]] = {}
 
-        async def wrapped(*args: P.args, **kwargs: P.kwargs) -> T:
+        async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
             key = make_key(args, kwargs)
             try:
                 return await internal_cache[key]
             except KeyError:
-                internal_cache[key] = fut = asyncio.ensure_future(coro(*args, **kwargs))
+                internal_cache[key] = fut = asyncio.ensure_future(
+                    coro(*args, **kwargs)
+                )
                 if ttl is not None:
                     # This results in internal_cache.pop(key, fut) later
                     # while avoiding a late binding issue with a lambda instead
@@ -96,37 +76,45 @@ async def wrapped(*args: P.args, **kwargs: P.kwargs) -> T:
     return wrapper
 
 
-def _lru_evict(ttl: float, cache: LRU[Hashable, Any], key: Hashable, _ignored_fut: object) -> None:
+def _lru_evict(
+    ttl: float, cache: LRU[Hashable, Any], key: Hashable, _ignored_fut: object
+) -> None:
     asyncio.get_running_loop().call_later(ttl, cache.remove, key)
 
 
-def lrucorocache(ttl: float | None = None, maxsize: int = 1024) -> Callable[[CoroLike[P, T]], CoroFunc[P, T]]:
+def lrucorocache(
+    ttl: float | None = None, maxsize: int = 1024
+) -> Callable[[CoroLike[P, R]], CoroFunc[P, R]]:
     """Decorator to cache coroutine functions.
 
-    This is less powerful than the version in task_cache.py but may work better for
-    some cases where typing of libraries this interacts with is too restrictive.
+    This is less powerful than the version in task_cache.py but may work better
+    for some cases where typing of libraries this interacts with is too restrictive.
 
-    Note: This uses the args and kwargs of the original coroutine function as a cache key.
-    This includes instances (self) when wrapping methods.
-    Consider not wrapping instance methods, but what those methods call when feasible in cases where this may matter.
+    Note: This uses the args and kwargs of the original coroutine function as a
+    cache key. This includes instances (self) when wrapping methods.
+    Consider not wrapping instance methods, but what those methods call when feasible
+    in cases where this may matter.
 
     The ordering of args and kwargs matters.
 
-    futs are evicted by LRU and ttl.
-    """
+    cached results are evicted by LRU and ttl."""
 
-    def wrapper(coro: CoroLike[P, T]) -> CoroFunc[P, T]:
-        internal_cache: LRU[Hashable, asyncio.Future[T]] = LRU(maxsize)
+    def wrapper(coro: CoroLike[P, R]) -> CoroFunc[P, R]:
+        internal_cache: LRU[Hashable, asyncio.Future[R]] = LRU(maxsize)
 
         @wraps(coro)
-        async def wrapped(*args: P.args, **kwargs: P.kwargs) -> T:
+        async def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
             key = make_key(args, kwargs)
             try:
                 return await internal_cache[key]
             except KeyError:
-                internal_cache[key] = fut = asyncio.ensure_future(coro(*args, **kwargs))
+                internal_cache[key] = fut = asyncio.ensure_future(
+                    coro(*args, **kwargs)
+                )
                 if ttl is not None:
-                    fut.add_done_callback(partial(_lru_evict, ttl, internal_cache, key))
+                    fut.add_done_callback(
+                        partial(_lru_evict, ttl, internal_cache, key)
+                    )
                 return await fut
 
         return wrapped

async_utils/lockout.py

@@ -43,14 +43,18 @@ class Lockout:
             async with ratelimiter, lockout:
                 response = await some_request(route, **parameters)
                 if response.code == 429:
-                    if reset_after := response.headers.get('X-Ratelimit-Reset-After')
-                        lockout.lock_for(reset_after)
+                    if reset := response.headers.get('X-Ratelimit-Reset-After')
+                        lockout.lock_for(reset)
 
     """
 
     def __repr__(self) -> str:
         res = super().__repr__()
-        extra = "unlocked" if not self._lockouts else f"locked, timestamps={self._lockouts:!r}"
+        extra = (
+            "unlocked"
+            if not self._lockouts
+            else f"locked, timestamps={self._lockouts:!r}"
+        )
         return f"<{res[1:-1]} [{extra}]>"
 
     def __init__(self) -> None:
@@ -105,7 +109,11 @@ def __init__(self) -> None:
 
     def __repr__(self) -> str:
         res = super().__repr__()
-        extra = "unlocked" if not self._lockouts else f"locked, timestamps={self._lockouts:!r}"
+        extra = (
+            "unlocked"
+            if not self._lockouts
+            else f"locked, timestamps={self._lockouts:!r}"
+        )
         return f"<{res[1:-1]} [{extra}]>"
 
     def lockout_for(self, seconds: float, /) -> None:

async_utils/priority_sem.py

@@ -27,7 +27,9 @@
 
 _global_lock = threading.Lock()
 
-_priority: contextvars.ContextVar[int] = contextvars.ContextVar("_priority", default=0)
+_priority: contextvars.ContextVar[int] = contextvars.ContextVar(
+    "_priority", default=0
+)
 
 
 class PriorityWaiter(NamedTuple):
@@ -109,8 +111,11 @@ def __repr__(self):
 
     def locked(self) -> bool:
         # Must do a comparison based on priority then FIFO
-        # in the case of existing waiters, not guaranteed to be immediately available
-        return self._value == 0 or (any(not w.cancelled() for w in (self._waiters or ())))
+        # in the case of existing waiters
+        # not guaranteed to be immediately available
+        return self._value == 0 or (
+            any(not w.cancelled() for w in (self._waiters or ()))
+        )
 
     async def __aenter__(self):
         prio = _priority.get()

async_utils/ratelimiter.py

@@ -34,11 +34,16 @@ def __init__(self, rate_limit: int, period: float, granularity: float):
         self._monotonics: deque[float] = deque()
 
     async def __aenter__(self):
-        # The ordering of these conditions matters to avoid an async context switch between
-        # confirming the ratelimit isn't exhausted and allowing the user code to continue.
-        while (len(self._monotonics) >= self.rate_limit) and await asyncio.sleep(self.granularity, True):
+        # The ordering of these conditions matters to avoid an async context
+        # switch between confirming the ratelimit isn't exhausted and allowing
+        # the user code to continue.
+        while (
+            len(self._monotonics) >= self.rate_limit
+        ) and await asyncio.sleep(self.granularity, True):
             now = time.monotonic()
-            while self._monotonics and (now - self._monotonics[0] > self.period):
+            while self._monotonics and (
+                now - self._monotonics[0] > self.period
+            ):
                 self._monotonics.popleft()
 
         self._monotonics.append(time.monotonic())

async_utils/scheduler.py

@@ -100,14 +100,17 @@ async def __anext__(self) -> T:
                 self.__tqueue.task_done()
         raise StopAsyncIteration
 
-    async def create_task(self, timestamp: float, payload: T, /) -> CancelationToken:
+    async def create_task(
+        self, timestamp: float, payload: T, /
+    ) -> CancelationToken:
         t = _Task(timestamp, payload)
         self.__tasks[t.cancel_token] = t
         await self.__tqueue.put(t)
         return t.cancel_token
 
     async def cancel_task(self, cancel_token: CancelationToken, /) -> bool:
-        """Returns if the task with that CancelationToken. Cancelling an already cancelled task is allowed."""
+        """Returns if the task with that CancelationToken. Cancelling an
+        already cancelled task is allowed and has no additional effect."""
         async with self.__l:
             try:
                 task = self.__tasks[cancel_token]