fix typing

setup.cfg

@@ -11,6 +11,7 @@ exclude = (?x)(
     | fill_db.py
     | stubs/
   )
+disable_error_code=unused-ignore
 
 [mypy-mmpy_bot.*]
 ignore_missing_imports=True

src/bot/plugins/admin.py

@@ -1,67 +1,79 @@
 import re
 from functools import wraps
-from typing import Any, Callable, Coroutine, TypeVar
+from typing import Any, Generic, Protocol, TypeVar
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from dependency_injector.wiring import Provide, inject
-from mmpy_bot import Message, Plugin, listen_to
 
+from mmpy_bot import Message, Plugin, listen_to
 from src.bot.services.admin import AdminService
 from src.bot.services.matching import MatchingService
 from src.bot.services.notify_service import NotifyService
 from src.core.db.models import Admin
 from src.depends import Container
 
-F = TypeVar("F", bound=Callable[..., Any])
+ReturningT = TypeVar("ReturningT", covariant=True)
 
 
-class BotAdmin(Plugin):
-    @staticmethod
-    def is_admin(fn: F) -> Callable[[Any, Any, AdminService], Coroutine[Any, Any, None]]:
-        @wraps(fn)
-        @inject
-        async def admin(
-            self: Plugin,
-            message: Message,
-            admin_service: AdminService = Provide[Container.admin_service],
-        ) -> None:
-            admin_instance = Admin(username=message.sender_name, user_id=message.user_id)
-            if await admin_service.check_if_admin(message.user_id, admin_instance):
-                await fn(self, message)
-            else:
-                self.driver.reply_to(message, "Недостаточно прав!")
+class Handler(Protocol, Generic[ReturningT]):
+    async def __call__(
+        self_: Any,  # type: ignore ## for pyright
+        self: Plugin,
+        message: Message,
+        *args: Any,
+    ) -> ReturningT:
+        ...
 
-        return admin
 
-    @listen_to("notify_all_users", re.IGNORECASE)
+def is_admin(fn: Handler[ReturningT]) -> Handler[ReturningT | None]:
+    @wraps(fn)
+    @inject
+    async def wrapper(
+        self: Plugin,
+        message: Message,
+        *args: Any,
+        admin_service: AdminService = Provide[Container.admin_service],
+    ) -> ReturningT | None:
+        admin_instance = Admin(username=message.sender_name, user_id=message.user_id)
+        if await admin_service.check_if_admin(message.user_id, admin_instance):
+            return await fn(self, message, *args)
+        else:
+            self.driver.reply_to(message, "Недостаточно прав!")
+        return None
+
+    return wrapper
+
+
+class BotAdmin(Plugin):
     @is_admin
+    @listen_to("notify_all_users", re.IGNORECASE)
     @inject
     async def test_notify_all_users(
         self, message: Message, notify_service: NotifyService = Provide[Container.week_routine_service,]
     ) -> None:
         """Тестирование опроса по пятницам"""
         await notify_service.notify_all_users(plugin=self)
 
-    @listen_to("monday_message", re.IGNORECASE)
     @is_admin
+    @listen_to("monday_message", re.IGNORECASE)
     @inject
     async def test_monday_message(
         self, message: Message, notify_service: NotifyService = Provide[Container.week_routine_service,]
     ) -> None:
         """Тестирование напоминания о встречах"""
         await notify_service.meeting_notifications(plugin=self)
 
-    @listen_to("wednesday_message", re.IGNORECASE)
     @is_admin
+    @listen_to("wednesday_message", re.IGNORECASE)
     @inject
     async def test_wednesday_message(
         self, message: str, notify_service: NotifyService = Provide[Container.week_routine_service,]
     ) -> None:
         """Тестирование опроса по средам"""
         await notify_service.match_review_notifications(plugin=self)
 
-    @listen_to("match", re.IGNORECASE)
     @is_admin
+    @listen_to("match", re.IGNORECASE)
     @inject
     async def test(
         self, message: Message, matching_service: MatchingService = Provide[Container.matching_service]
@@ -73,8 +85,8 @@ async def test(
         except Exception as error:
             self.driver.reply_to(message, str(error))
 
-    @listen_to("close", re.IGNORECASE)
     @is_admin
+    @listen_to("close", re.IGNORECASE)
     @inject
     async def test_closing_meetings(
         self, message: Message, matching_service: MatchingService = Provide[Container.matching_service]
@@ -86,8 +98,8 @@ async def test_closing_meetings(
         except Exception as error:
             self.driver.reply_to(message, str(error))
 
-    @listen_to("stop_jobs", re.IGNORECASE)
     @is_admin
+    @listen_to("stop_jobs", re.IGNORECASE)
     @inject
     def cancel_jobs(self, message: Message, scheduler: AsyncIOScheduler = Provide[Container.scheduler,]) -> None:
         """Остановка всех jobs"""

stubs/mmpy_bot/function.pyi

@@ -1,31 +1,37 @@
 from abc import ABC
-from typing import Callable, TypeVar
+from typing import Any, Callable, Generic, Protocol, TypeVar
 
 from _typeshed import Incomplete
+
 from mmpy_bot.plugins import Plugin as Plugin
 from mmpy_bot.utils import completed_future as completed_future
 from mmpy_bot.webhook_server import NoResponse as NoResponse
 from mmpy_bot.wrappers import Message as Message
 from mmpy_bot.wrappers import WebHookEvent as WebHookEvent
 
 log: Incomplete
-ReturningType = TypeVar("ReturningType")
+ReturningType = TypeVar("ReturningType", covariant=True)
+
+class MessageFunction(Protocol, Generic[ReturningType]):
+    async def __call__(self_, self: Plugin, message: Message, *args) -> ReturningType: ...  # type: ignore
 
 def listen_to(
     regexp: str,
-    regexp_flag: int = ...,
+    regexp_flag: int = 0,
     *,
-    direct_only: bool = ...,
-    needs_mention: bool = ...,
-    allowed_users: Incomplete | None = ...,
-    allowed_channels: Incomplete | None = ...,
-    silence_fail_msg: bool = ...,
-    **metadata: dict[Incomplete, Incomplete]
-) -> Callable[[Callable[..., ReturningType]], ReturningType]: ...
+    direct_only: bool = False,
+    needs_mention: bool = False,
+    allowed_users: Any = None,
+    allowed_channels: Any = None,
+    silence_fail_msg: Any = False,
+    **metadata: dict[str, Any],
+) -> Callable[[Callable[..., ReturningType]], MessageFunction[ReturningType]]:
+    """Wrap the given function in a MessageFunction class so we can register some
+    properties."""
+
 def listen_webhook(
     regexp: str, **metadata: dict[Incomplete, Incomplete]
-) -> Callable[[Callable[..., ReturningType]], ReturningType]: ...
+) -> Callable[[Callable[..., ReturningType]], Callable[..., ReturningType]]: ...
 
 class Function(ABC): ...
-class MessageFunction(Function): ...
 class WebHookFunction(Function): ...

stubs/mmpy_bot/plugins/base.pyi

@@ -3,6 +3,7 @@ from abc import ABC
 from typing import Any, Dict, List, Optional, Sequence, Union
 
 from _typeshed import Incomplete
+
 from mmpy_bot.driver import Driver as Driver
 from mmpy_bot.function import Function as Function
 from mmpy_bot.function import MessageFunction as MessageFunction
@@ -53,7 +54,7 @@ class FunctionInfo:
 
 def get_function_characteristics(function: FunctionInfo) -> tuple[bool, bool, str]: ...
 def generate_plugin_help(
-    listeners: Dict[re.Pattern[Any], List[Union[MessageFunction, WebHookFunction]]]
+    listeners: Dict[re.Pattern[Any], List[Union[MessageFunction[Any], WebHookFunction]]]
 ) -> list[FunctionInfo]: ...
 
 class PluginManager: