Fix types issues

app/bot.py

@@ -1,4 +1,4 @@
-from aiogram import Dispatcher, F, Bot
+from aiogram import Bot, Dispatcher, F
 from aiogram.types import BufferedInputFile, Message
 
 from settings import settings
@@ -29,7 +29,7 @@ async def handle_tiktok_request(message: Message, bot: Bot) -> None:
     ]
 
     async for tiktok in TikTokAPI.download_tiktoks(urls):
-        if tiktok.is_empty:
+        if not tiktok.video:
             continue
 
         video = BufferedInputFile(tiktok.video, filename="video.mp4")

app/settings.py

@@ -12,7 +12,7 @@ class Settings:
 
 
 def parse_env_list(key: str) -> list[int]:
-    return json.loads(os.getenv(key, "[]"))
+    return list(map(int, json.loads(os.getenv(key, "[]"))))
 
 
 def parse_env_bool(key: str, default: str = "false") -> bool:

app/tiktok/api.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import AsyncIterable
+from collections.abc import AsyncIterable
 
 from tiktok.client import AsyncTikTokClient
 from tiktok.data import Tiktok

app/tiktok/client.py

@@ -1,13 +1,13 @@
 import json
 import random
 import string
-from datetime import datetime, UTC
+from datetime import UTC, datetime
 
 import httpx
 from bs4 import BeautifulSoup
 
 from tiktok.data import ItemStruct
-from utils import RetryingError, retries
+from utils import DifferentPageError, NoDataError, NoScriptError, retries
 
 
 class AsyncTikTokClient(httpx.AsyncClient):
@@ -39,15 +39,15 @@ async def get_page_data(self, url: str) -> ItemStruct:
         if script := soup.select_one('script[id="__UNIVERSAL_DATA_FOR_REHYDRATION__"]'):
             script = json.loads(script.text)
         else:
-            raise RetryingError("no script")
+            raise NoScriptError
 
         try:
             data = script["__DEFAULT_SCOPE__"]["webapp.video-detail"]["itemInfo"]["itemStruct"]
         except KeyError as ex:
-            raise RetryingError("no data") from ex
+            raise NoDataError from ex
 
         if data["id"] != page_id:
-            raise RetryingError("tiktok_id is different from page_id")
+            raise DifferentPageError
         return ItemStruct.parse(data)
 
     async def get_video(self, url: str) -> bytes | None:

app/tiktok/data.py

@@ -7,10 +7,6 @@ class Tiktok:
     description: str = ""
     video: bytes | None = None
 
-    @property
-    def is_empty(self) -> bool:
-        return not self.video
-
     @property
     def caption(self) -> str:
         return f"{self.description}\n\n{self.url}"

app/utils.py

@@ -1,22 +1,47 @@
 import asyncio
 import logging
+from collections.abc import Awaitable, Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 
 class RetryingError(Exception):
     pass
 
 
-def retries(times: int):
-    def decorator(func):
+class NoScriptError(RetryingError):
+    def __init__(self) -> None:
+        super().__init__("no script")
+
+
+class NoDataError(RetryingError):
+    def __init__(self) -> None:
+        super().__init__("no data")
+
+
+class DifferentPageError(RetryingError):
+    def __init__(self) -> None:
+        super().__init__("tiktok_id is different from page_id")
+
+
+P = ParamSpec("P")
+T = TypeVar("T")
+
+Wrapper = Callable[P, Awaitable[T | None]]
+Decorator = Callable[[Callable[P, Awaitable[T]]], Wrapper]
+
+
+def retries(times: int) -> Decorator:
+    def decorator(func: Callable[P, Awaitable[T]]) -> Wrapper:
         @wraps(func)
-        async def wrapper(*args, **kwargs):
+        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T | None:
             for _ in range(times):
                 try:
                     return await func(*args, **kwargs)
                 except RetryingError:
                     logging.exception("Retrying")
                     await asyncio.sleep(0.5)
+            return None
 
         return wrapper