refactor: add and improve type annotations across utils and database

Author: alissonlauffer

eduu/database/localization.py

@@ -1,6 +1,8 @@
 # SPDX-License-Identifier: MIT
 # Copyright (c) 2018-2026 Amano LLC
 
+from __future__ import annotations
+
 from hydrogram.enums import ChatType
 
 from eduu.utils.consts import GROUP_TYPES
@@ -10,7 +12,7 @@
 conn = database.get_conn()
 
 
-async def set_db_lang(chat_id: int, chat_type: ChatType, lang_code: str):
+async def set_db_lang(chat_id: int, chat_type: ChatType, lang_code: str) -> None:
     if chat_type in {ChatType.PRIVATE, ChatType.BOT}:
         await conn.execute(
             "UPDATE users SET chat_lang = ? WHERE user_id = ?", (lang_code, chat_id)
@@ -30,7 +32,7 @@ async def set_db_lang(chat_id: int, chat_type: ChatType, lang_code: str):
         raise TypeError(f"Unknown chat type '{chat_type}'.")
 
 
-async def get_db_lang(chat_id: int, chat_type: ChatType) -> str:
+async def get_db_lang(chat_id: int, chat_type: ChatType) -> str | None:
     if chat_type == ChatType.PRIVATE:
         cursor = await conn.execute("SELECT chat_lang FROM users WHERE user_id = ?", (chat_id,))
         ul = await cursor.fetchone()

eduu/utils/decorators.py

@@ -18,12 +18,12 @@
 from eduu.utils.utils import check_perms
 
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    from collections.abc import Callable, Coroutine
 
 
-def aiowrap(func: Callable) -> Callable:
+def aiowrap[T, **P](func: Callable[P, T]) -> Callable[..., Coroutine[None, None, T]]:
     @wraps(func)
-    async def run(*args, loop=None, executor=None, **kwargs):
+    async def run(*args: P.args, loop=None, executor=None, **kwargs: P.kwargs) -> T:
         if loop is None:
             loop = asyncio.get_event_loop()
         pfunc = partial(func, *args, **kwargs)
@@ -92,10 +92,12 @@ async def wrapper(client: Client, message: CallbackQuery | Message, *args, **kwa
     return decorator
 
 
-def stop_here(func: Callable) -> Callable:
-    async def wrapper(*args, **kwargs):
+def stop_here[T, **P](
+    func: Callable[P, Coroutine[None, None, T]],
+) -> Callable[P, Coroutine[None, None, T]]:
+    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
         try:
-            await func(*args, **kwargs)
+            return await func(*args, **kwargs)
         finally:
             raise StopPropagation
 

eduu/utils/localization.py

@@ -8,12 +8,18 @@
 from collections.abc import Callable
 from functools import partial
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 from hydrogram.enums import ChatType
 from hydrogram.types import CallbackQuery, InlineQuery, Message
 
 from eduu.database.localization import get_db_lang
 
+if TYPE_CHECKING:
+    from collections.abc import Coroutine
+
+    from hydrogram import Client
+
 enabled_locales: list[str] = [
     "en-GB",  # English (United Kingdom)
     "en-US",  # English (United States)
@@ -66,7 +72,8 @@ def cache_locales(locales: list[str]) -> dict[str, dict[str, str]]:
 
         if "_meta_language_name" not in locale_keys or "_meta_language_flag" not in locale_keys:
             logging.warning(
-                "Locale has required keys _meta_language_name or _meta_language_flag missing. This locale will not be loaded."
+                "Locale has required keys _meta_language_name or _meta_language_flag missing."
+                " This locale will not be loaded."
             )
             continue
 
@@ -127,10 +134,17 @@ async def get_lang(message: CallbackQuery | Message | InlineQuery) -> str:
     return lang if lang in enabled_locales else default_language
 
 
-def use_chat_lang(func: Callable):
+def use_chat_lang[T](
+    func: Callable[..., Coroutine[None, None, T]],
+) -> Callable[..., Coroutine[None, None, T]]:
     """Decorator to get the chat language and pass it to the function."""
 
-    async def wrapper(client, message, *args, **kwargs):
+    async def wrapper(
+        client: Client,
+        message: CallbackQuery | Message | InlineQuery,
+        *args,
+        **kwargs,
+    ) -> T:
         lang = await get_lang(message)
 
         lfunc = partial(get_locale_string, lang)

eduu/utils/utils.py

@@ -10,6 +10,7 @@
 from datetime import datetime, timedelta
 from functools import partial
 from string import Formatter
+from typing import TYPE_CHECKING
 
 from curl_cffi.requests import AsyncSession
 from hydrogram import Client, filters
@@ -24,6 +25,12 @@
 
 from config import SUDOERS
 
+if TYPE_CHECKING:
+    from collections.abc import Callable, Coroutine
+    from typing import Any
+
+    from eduu.utils.localization import Strings
+
 BTN_URL_REGEX = re.compile(r"(\[([^\[]+?)\]\(buttonurl:(?:/{0,2})(.+?)(:same)?\))")
 
 SMART_OPEN = "“"
@@ -34,12 +41,14 @@
 http = AsyncSession(timeout=40)
 
 
-def run_async(func, *args, **kwargs):
+def run_async[T, **P](
+    func: Callable[P, Coroutine[Any, Any, T]], *args: P.args, **kwargs: P.kwargs
+) -> T:
     loop = asyncio.get_event_loop()
-    loop.run_until_complete(func(*args, **kwargs))
+    return loop.run_until_complete(func(*args, **kwargs))
 
 
-def pretty_size(size_bytes):
+def pretty_size(size_bytes: int) -> str:
     if size_bytes == 0:
         return "0B"
     size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
@@ -53,7 +62,7 @@ async def check_perms(
     message: CallbackQuery | Message,
     permissions: ChatPrivileges | None = None,
     complain_missing_perms: bool = True,
-    s=None,
+    s: Strings | None = None,
 ) -> bool:
     if isinstance(message, CallbackQuery):
         sender = partial(message.answer, show_alert=True)
@@ -125,7 +134,7 @@ def remove_escapes(text: str) -> str:
     return res
 
 
-def split_quotes(text: str) -> list:
+def split_quotes(text: str) -> list[str]:
     if not any(text.startswith(char) for char in START_CHAR):
         return text.split(None, 1)
     counter = 1  # ignore first char -> is some kind of quote
@@ -198,8 +207,8 @@ def add_command(
         self,
         command: str,
         category: str,
-        aliases: list | None = None,
-    ):
+        aliases: list[str] | None = None,
+    ) -> None:
         description_key = f"cmd_{command}_description"
 
         if self.commands.get(category) is None:
@@ -210,7 +219,7 @@ def add_command(
             "aliases": aliases or [],
         })
 
-    def get_commands_message(self, s, category: str | None = None):
+    def get_commands_message(self, s: Strings, category: str | None = None) -> str:
         # TODO: Add pagination support.
         if category is None:
             cmds_list = []
@@ -238,8 +247,8 @@ def __init__(self):
     def add_command(
         self,
         command: str,
-        aliases: list | None = None,
-    ):
+        aliases: list[str] | None = None,
+    ) -> None:
         description_key = f"inline_cmd_{command.split(maxsplit=1)[0]}_description"
 
         self.commands.append({
@@ -248,7 +257,7 @@ def add_command(
             "aliases": aliases or [],
         })
 
-    def search_commands(self, query: str | None = None):
+    def search_commands(self, query: str | None = None) -> list[dict[str, str | list[str]]]:
         return [
             cmd
             for cmd in sorted(self.commands, key=operator.itemgetter("command"))
@@ -277,7 +286,7 @@ async def get_target_user(c: Client, m: Message) -> User:
     )
 
 
-def get_reason_text(c: Client, m: Message) -> Message:
+def get_reason_text(c: Client, m: Message) -> str | None:
     reply = m.reply_to_message
     spilt_text = m.text.split