improve typing

unwind/__main__.py

@@ -15,7 +15,7 @@ async def run_load_user_ratings_from_imdb():
     await open_connection_pool()
 
     i = 0
-    async for rating in refresh_user_ratings_from_imdb():
+    async for _ in refresh_user_ratings_from_imdb():
         i += 1
 
     log.info("✨ Imported %s new ratings.", i)

unwind/imdb.py

@@ -34,7 +34,7 @@ log = logging.getLogger(__name__)
 #             p.text-muted.text-small span[name=nv] [data-value]
 
 
-async def refresh_user_ratings_from_imdb(stop_on_dupe=True):
+async def refresh_user_ratings_from_imdb(stop_on_dupe: bool = True):
     with session() as s:
         s.headers["Accept-Language"] = "en-US, en;q=0.5"
 
@@ -149,7 +149,7 @@ def movie_and_rating_from_item(item) -> tuple[Movie, Rating]:
 ForgedRequest = namedtuple("ForgedRequest", "url headers")
 
 
-async def parse_page(url) -> tuple[list[Rating], str | None]:
+async def parse_page(url: str) -> tuple[list[Rating], str | None]:
     ratings = []
 
     soup = soup_from_url(url)
@@ -191,7 +191,7 @@ async def parse_page(url) -> tuple[list[Rating], str | None]:
     return (ratings, next_url if url != next_url else None)
 
 
-async def load_ratings(user_id):
+async def load_ratings(user_id: str):
     next_url = user_ratings_url(user_id)
 
     while next_url:

unwind/models.py

@@ -19,7 +19,7 @@ from typing import (
 
 from .types import ULID
 
-JSON = Union[int, float, str, None, list["JSON"], dict[str, "JSON"]]
+JSON = int | float | str | None | list["JSON"] | dict[str, "JSON"]
 JSONObject = dict[str, JSON]
 
 T = TypeVar("T")

unwind/request.py

@@ -11,7 +11,7 @@ from hashlib import md5
 from pathlib import Path
 from random import random
 from time import sleep, time
-from typing import Callable, cast
+from typing import Callable, ParamSpec, TypeVar, cast
 
 import bs4
 import httpx
@@ -31,6 +31,9 @@ _ASession_T = httpx.AsyncClient
 _Session_T = httpx.Client
 _Response_T = httpx.Response
 
+_T = TypeVar("_T")
+_P = ParamSpec("_P")
+
 
 @contextmanager
 def session():
@@ -88,15 +91,15 @@ async def asession():
 
 def _throttle(
     times: int, per_seconds: float, jitter: Callable[[], float] | None = None
-) -> Callable[[Callable], Callable]:
+) -> Callable[[Callable[_P, _T]], Callable[_P, _T]]:
     calls: deque[float] = deque(maxlen=times)
 
     if jitter is None:
         jitter = lambda: 0.0
 
-    def decorator(func: Callable) -> Callable:
+    def decorator(func: Callable[_P, _T]) -> Callable[_P, _T]:
         @wraps(func)
-        def inner(*args, **kwds):
+        def inner(*args: _P.args, **kwds: _P.kwargs):
             # clean up
             while calls:
                 if calls[0] + per_seconds > time():