refactor: fix typing error with pyright

pyproject.toml

@@ -26,14 +26,14 @@ dev = [
     'asyncpg-stubs ==0.30.0',
     'types-pyyaml ==6.0.12.20241230',
     'google-re2-stubs ==0.1.1',
+    "types-redis>=4.6.0.20241004",
 ]
 
 [tool.uv]
 package = false
 required-environments = [
     "sys_platform == 'linux' and platform_machine == 'x86_64'",
     "sys_platform == 'win32' and platform_machine == 'x86_64'",
-
 ]
 
 [tool.black]
@@ -144,3 +144,17 @@ exclude = ['tests']
 
 [tool.pytest.ini_options]
 python_files = ['tests/**.py', 'test_*.py', '*_test.py']
+
+[tool.pyright]
+root = '.'
+ignore = ['./tests/']
+# include = ['./transmission_rpc/']
+# ignore = ['./tests/', './docs/']
+pythonVersion = '3.10'
+pythonPlatform = 'Linux'
+typeCheckingMode = "strict"
+# reportUnnecessaryComparison = false
+reportUnknownVariableType = false
+reportUnknownMemberType = false
+reportUnnecessaryIsInstance = false
+reportPrivateUsage = false

server/app.py

@@ -170,7 +170,7 @@ def internal_error_handler(req: Request, exc: Exception) -> Response[Any]:
 
 async def startup_fetch_missing_users() -> None:
     logger.info("fetch missing users")
-    s = set()
+    s: set[int | None] = set()
 
     for u1, u2 in await pg.fetch("select from_user_id, wiki_user_id from subject_patch"):
         s.add(u1)

server/config.py

@@ -25,7 +25,7 @@
 
 HEADER_KEY_API = "x-api-token"
 
-SUPER_USERS = {}
+SUPER_USERS: dict[str, dict[str, int]] = {}
 
 for key in os.environ:
     if not key.startswith("SUPER_USER_TOKEN_"):

server/contrib.py

@@ -109,9 +109,8 @@ async def suggest_api(
     res.raise_for_status()
     original_wiki = res.json()
 
-    original = {}
-
-    changed = {}
+    original: dict[str, Any] = {}
+    changed: dict[str, Any] = {}
 
     for key in ["name", "infobox", "summary"]:
         before = original_wiki[key]
@@ -199,9 +198,9 @@ async def suggest_api_from_partial(
     res.raise_for_status()
     original_wiki = res.json()
 
-    original = {}
+    original: dict[str, Any] = {}
 
-    changed = {}
+    changed: dict[str, Any] = {}
 
     for key in ["name", "infobox", "summary", "nsfw"]:
         before = original_wiki[key]
@@ -353,7 +352,7 @@ async def _(
 
             patch = SubjectPatch.from_dict(p)
 
-            changed = {}
+            changed: dict[str, Any] = {}
 
             if patch.from_user_id != request.auth.user_id:
                 raise PermissionDeniedException()
@@ -494,7 +493,7 @@ async def creat_episode_patch(
 
     keys = ["airdate", "name", "name_cn", "duration", "description"]
 
-    changed = {}
+    changed: dict[str, Any] = {}
 
     for key in keys:
         if original_wiki[key] != getattr(data, key):
@@ -505,7 +504,7 @@ async def creat_episode_patch(
 
     reason = data.reason.strip()
     if not reason:
-        reasons = []
+        reasons: list[str] = []
         for key in changed:
             if original_wiki[key]:
                 reasons.append(f"修改{patch_keys[key]}")

server/tmpl.py

@@ -11,6 +11,7 @@
 from jinja2.runtime import Context
 from litestar import Request
 from markupsafe import Markup
+from multidict import MultiDict
 
 from server.config import DEV, PROJECT_PATH, TURNSTILE_SITE_KEY, UTC
 
@@ -21,9 +22,13 @@
     auto_reload=DEV,
 )
 
-engine.globals["TURNSTILE_SITE_KEY"] = TURNSTILE_SITE_KEY
-engine.globals["min"] = min
-engine.globals["max"] = max
+
+engine_globals: dict[str, Any] = engine.globals
+engine_filters: dict[str, Any] = engine.filters
+
+engine_globals["TURNSTILE_SITE_KEY"] = TURNSTILE_SITE_KEY
+engine_globals["min"] = min
+engine_globals["max"] = max
 
 P = typing.ParamSpec("P")
 T = typing.TypeVar("T")
@@ -41,11 +46,15 @@ def add_filter(s: str | Callable[P, T]) -> Any:
     def real_wrapper(name: str, fn: Callable[P, T]) -> Callable[P, T]:
         if name in engine.filters:
             raise ValueError(f"filter '{name}' already exists")
-        engine.filters[name] = fn
+        engine_filters[name] = fn
         return fn
 
     if isinstance(s, str):
-        return lambda fn: real_wrapper(s, fn)
+
+        def inner(fn: Callable[P, T]):
+            return real_wrapper(s, fn)
+
+        return inner
 
     return real_wrapper(s.__name__, s)
 
@@ -54,7 +63,7 @@ def add_global_function(fn: Callable[P, T]) -> Callable[P, T]:
     name = fn.__name__
     if name in engine.globals:
         raise ValueError(f"filter '{name}' already exists")
-    engine.globals[name] = fn
+    engine_globals[name] = fn
     return fn
 
 
@@ -134,7 +143,7 @@ def to_user_local_time(ctx: Context, dt: datetime) -> str:
 @pass_context
 def replace_url_query(ctx: Context, **kwargs: Any) -> str:
     req: Request[None, None, Any] = ctx["request"]
-    q = req.url.query_params.copy()
+    q: MultiDict[Any] = req.url.query_params.copy()
     q.update(kwargs)
     return req.url.path + "?" + urlencode(q)