Merge pull request #720 from aobolensk/mypy

Add initial configuration for mypy

.github/workflows/lint.yml

@@ -94,3 +94,28 @@ jobs:
     - name: List files with error (open this section to see the list of files that failed this check)
       run: git diff | grep "diff --git"; exit 1
       if: failure()
+  mypy:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+    - uses: actions/setup-python@v4
+      with:
+        python-version: '3.12'
+        cache: pip
+        cache-dependency-path: |
+          requirements.txt
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install -r requirements.txt
+    - name: Install mypy
+      run: python -m pip install mypy
+    - name: Install types
+      run: |
+        python -m pip install \
+          types-requests \
+          types-psutil \
+          types-python-dateutil \
+          types-PyYAML
+    - name: Run mypy
+      run: python -m mypy .

setup.cfg

@@ -17,3 +17,20 @@ disable =
 skip =
     venv
 multi_line_output = 3
+
+[mypy]
+# TODO: remove disable_error_code
+disable_error_code =
+    union-attr,
+    arg-type,
+    typeddict-item,
+    attr-defined,
+    assignment,
+    func-returns-value,
+    safe-super,
+    misc
+exclude =
+    venv,
+    src/db,
+    tests
+explicit_package_bases = True

src/api/command.py

@@ -1,7 +1,7 @@
 import enum
 from abc import ABC, abstractmethod
 from types import FunctionType
-from typing import TYPE_CHECKING, Any, Dict, List
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 from src import const
 from src.api.execution_context import ExecutionContext
@@ -79,7 +79,7 @@ def store_persistent_state(self, commands_data: Dict[str, Any]):
         commands_data[self.command_name]["times_called"] = self.times_called
         commands_data[self.command_name]["max_execution_time"] = self.max_execution_time
 
-    async def run(self, cmd_line: List[str], execution_ctx: ExecutionContext) -> None:
+    async def run(self, cmd_line: List[str], execution_ctx: ExecutionContext) -> Optional[str]:
         if execution_ctx.platform == const.BotBackend.DISCORD:
             # On Discord platform we are using legacy separate time limit handling for now
             return await self._run_impl(cmd_line, execution_ctx)
@@ -92,12 +92,12 @@ async def run(self, cmd_line: List[str], execution_ctx: ExecutionContext) -> Non
             await Command.send_message(execution_ctx, f"Command '{' '.join(cmd_line)}' took too long to execute")
         return result
 
-    async def _run_impl(self, cmd_line: List[str], execution_ctx: ExecutionContext) -> None:
+    async def _run_impl(self, cmd_line: List[str], execution_ctx: ExecutionContext) -> Optional[str]:
         if execution_ctx.platform != const.BotBackend.DISCORD:
             # On Discord platform we are using legacy separate permission handling for now
             if execution_ctx.permission_level < self.permission_level:
                 await self.send_message(execution_ctx, f"You don't have permission to call command '{cmd_line[0]}'")
-                return
+                return None
         self.times_called += 1
         if not self.postpone_execution:
             cmd_line = (await self.process_variables(execution_ctx, ' '.join(cmd_line), cmd_line)).split(' ')
@@ -122,9 +122,8 @@ async def _run_impl(self, cmd_line: List[str], execution_ctx: ExecutionContext)
         else:
             raise RuntimeError("invalid implementation type")
 
-    @abstractmethod
     async def _exec(self, cmd_line: List[str], execution_ctx: ExecutionContext) -> str:
-        pass
+        raise NotImplementedError("Command executor is not implemented")
 
     @staticmethod
     async def check_args_count(execution_ctx, cmd_line, min=None, max=None):

src/api/execution_context.py

@@ -32,7 +32,7 @@ def message_author(self) -> str:
         pass
 
     @abstractmethod
-    def message_author_id(self) -> int:
+    def message_author_id(self) -> str:
         pass
 
     @abstractmethod

src/api/plugin.py

@@ -48,6 +48,7 @@ async def update_implementation(self) -> None:
         """
         raise NotImplementedError
 
+    @staticmethod
     def plugin_api_version() -> PluginAPIVersion:
         """Plugin API version"""
         return PluginAPIVersion.LATEST

src/autoupdate_impl.py

@@ -70,7 +70,8 @@ def check_updates(context: AutoUpdateContext) -> bool:
     """Function that performs updates check. It is called periodically"""
     secret_config = Util.read_config_file(const.SECRET_CONFIG_PATH)
     if secret_config is None:
-        return log.error("Failed to read secret config file")
+        log.error("Failed to read secret config file")
+        return False
     mail = Mail(secret_config)
     old_sha = context.repo.head.object.hexsha
     try:
@@ -80,20 +81,25 @@ def check_updates(context: AutoUpdateContext) -> bool:
             secret_config.admin_email_list,
             "Autoupdate error",
             get_autoupdate_error_message(f"Failed to fetch updates from remote: {e}"))
-        return log.error(f"Fetch failed: {e}. Skipping this cycle, will try to update on the next one")
+        log.error(f"Fetch failed: {e}. Skipping this cycle, will try to update on the next one")
+        return False
     new_sha = context.repo.remotes.origin.refs['master'].object.name_rev.split()[0]
     log.debug(f"{old_sha} {new_sha}")
     if not FF.is_enabled("WALBOT_TEST_AUTO_UPDATE") and old_sha == new_sha:
-        return log.debug("No new updates")
+        log.debug("No new updates")
+        return False
     bot_cache = importlib.import_module("src.bot_cache").BotCache(True).parse()
     if bot_cache is None:
-        return log.warning("Could not read bot cache. Skipping this cycle, will try to update on the next one")
+        log.warning("Could not read bot cache. Skipping this cycle, will try to update on the next one")
+        return False
     if "do_not_update" not in bot_cache.keys():
-        return log.warning(
+        log.warning(
             "Could not find 'do_not_update' field in bot cache. "
             "Skipping this cycle, will try to update on the next one")
+        return False
     if bot_cache["do_not_update"]:
-        return log.debug("Automatic update is not permitted. Skipping this cycle, will try to update on the next one")
+        log.debug("Automatic update is not permitted. Skipping this cycle, will try to update on the next one")
+        return False
     context.repo.git.reset("--hard")
     try:
         g = git.cmd.Git(const.WALBOT_DIR)

src/backend/discord/commands.py

@@ -12,7 +12,7 @@
 
 
 class DiscordCommandBinding(CommandBinding):
-    def bind(self, cmd_name: str, command: Command):
+    def bind(self, cmd_name: str, command: Command):  # type:ignore
         if command.module_name is None:
             return
         bc.discord.commands.register_command(
@@ -31,9 +31,9 @@ def unbind(self, cmd_name: str):
 class Commands:
     def __init__(self) -> None:
         if not hasattr(self, "data"):
-            self.data = dict()
+            self.data: Dict[str, Any] = dict()
         if not hasattr(self, "aliases"):
-            self.aliases = dict()
+            self.aliases: Dict[str, Any] = dict()
 
     def update(self, reload: bool = False) -> None:
         bc.discord.commands = self

src/backend/discord/config.py

@@ -1,4 +1,7 @@
+from typing import Any, Dict
+
+
 class DiscordConfig:
     def __init__(self) -> None:
-        self.guilds = dict()
-        self.users = dict()
+        self.guilds: Dict[str, Any] = dict()
+        self.users: Dict[str, Any] = dict()

src/backend/discord/context.py

@@ -1,6 +1,6 @@
 import asyncio
 import re
-from typing import Optional
+from typing import Any, Optional
 
 import discord
 
@@ -40,7 +40,7 @@ def disable_pings(self, message: str) -> str:
             if r is None:
                 break
             t = asyncio.create_task(self.message.guild.fetch_member(int(r.group(1))))
-            member = asyncio.run(t)
+            member: Any = asyncio.run(t)
             message = const.DISCORD_USER_ID_REGEX.sub(str(member), message, count=1)
         while True:
             r = const.DISCORD_ROLE_ID_REGEX.search(message)
@@ -57,8 +57,8 @@ def disable_pings(self, message: str) -> str:
     def message_author(self) -> str:
         return self.message.author.mention
 
-    def message_author_id(self) -> int:
-        return self.message.author.id
+    def message_author_id(self) -> str:
+        return str(self.message.author.id)
 
     def channel_name(self) -> str:
         return self.message.channel.mention

src/backend/discord/embed.py

@@ -1,5 +1,5 @@
 import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 
 import discord
 
@@ -10,7 +10,7 @@ class DiscordEmbed:
     """Discord embed constructor"""
 
     def __init__(self) -> None:
-        self._data = dict()
+        self._data: Dict[str, Any] = dict()
 
     def get(self) -> discord.Embed:
         """Get embed result"""

src/backend/discord/instance.py

@@ -4,6 +4,7 @@
 import itertools
 import re
 import sys
+from typing import Optional
 
 import discord
 
@@ -203,8 +204,9 @@ async def _process_regular_message(self, message: discord.Message) -> None:
                 message.content = cmd
                 result = await self._process_command(message, cmd_split, silent=True)
                 message.content = msg_content
+                result = result or ""
                 if not self.config.discord.guilds[message.channel.guild.id].markov_pings:
-                    result = (DiscordExecutionContext(message).disable_pings(result)) or ""
+                    result = (DiscordExecutionContext(message).disable_pings(result))
                 await message.channel.send(message.author.mention + ' ' + result)
         elif channel_id in self.config.discord.guilds[message.channel.guild.id].markov_logging_whitelist:
             # If the message is in a channel that is supposed to log markov chains, doesn't mention the bot then
@@ -232,7 +234,7 @@ async def _process_regular_message(self, message: discord.Message) -> None:
                     except discord.HTTPException:
                         pass
 
-    async def _process_command(self, message: discord.Message, command=None, silent=False) -> None:
+    async def _process_command(self, message: discord.Message, command=None, silent=False) -> Optional[str]:
         if command is None:
             command = message.content.split(' ')
         command = list(filter(None, command))
@@ -246,7 +248,7 @@ async def _process_command(self, message: discord.Message, command=None, silent=
                 await message.channel.send(
                     f"Unknown command '{command[0]}', "
                     f"probably you meant '{self._suggest_similar_command(command[0])}'")
-                return
+                return None
         max_exec_time = self.config.commands.data[command[0]].max_execution_time
         if command[0] in self.config.executor["commands_data"].keys():
             max_exec_time = bc.executor.commands[command[0]].max_execution_time

src/backend/repl/instance.py

@@ -1,6 +1,7 @@
 import asyncio
 import inspect
 import socket
+from typing import Optional
 
 from src import const
 from src.api.bot_instance import BotInstance
@@ -14,7 +15,7 @@
 class ReplBotInstance(BotInstance):
     def __init__(self) -> None:
         self.channel = None
-        self.sock = None
+        self.sock: Optional[socket.socket] = None
 
     async def parse_command(self, commands, message) -> str:
         args = message.split(' ')

src/backend/telegram/command.py

@@ -13,7 +13,7 @@
 
 async def _command_handler(command_name: str, update: Update, context: CallbackContext) -> None:
     await bc.executor.commands[command_name].run(
-        [command_name] + context.args, TelegramExecutionContext(update, context))
+        [command_name] + (context.args or list()), TelegramExecutionContext(update, context))
 
 
 @Mail.send_exception_info_to_admin_emails

src/backend/telegram/config.py

@@ -1,8 +1,9 @@
 import uuid
+from typing import Any, Dict, Set
 
 
 class TelegramConfig:
     def __init__(self) -> None:
-        self.channel_whitelist = set()
+        self.channel_whitelist: Set[str] = set()
         self.passphrase = uuid.uuid4().hex
-        self.users = dict()
+        self.users: Dict[str, Any] = dict()

src/backend/telegram/context.py

@@ -1,4 +1,5 @@
 import os
+from typing import Dict
 
 from telegram import Update
 from telegram.ext import CallbackContext
@@ -13,12 +14,12 @@
 class TelegramExecutionContext(ExecutionContext):
     def __init__(self, update: Update, context: CallbackContext) -> None:
         super().__init__()
-        self.platform = const.BotBackend.TELEGRAM
+        self.platform = str(const.BotBackend.TELEGRAM)
         self.update = update
         self.context = context
         self.user = bc.config.telegram.users[update.message.from_user.id]
         self.permission_level = bc.config.telegram.users[update.message.from_user.id].permission_level
-        self._replace_patterns = dict()
+        self._replace_patterns: Dict[str, str] = dict()
 
     async def send_message(self, message: str, *args, **kwargs) -> None:
         if self.silent:
@@ -45,7 +46,7 @@ async def reply(self, message: str, *args, **kwargs) -> None:
         await self.send_message(message, *args, **kwargs, reply_on_msg=True)
 
     async def send_direct_message(self, user_id: int, message: str, *args, **kwargs) -> None:
-        await send_message(user_id, message)
+        send_message(user_id, message)
 
     def _unescape_ping1(self, message: str) -> str:
         idx = 0
@@ -75,7 +76,7 @@ def message_author(self) -> str:
         return self.update.message.from_user.mention_markdown_v2()
 
     def message_author_id(self) -> str:
-        return self.update.message.from_user.id
+        return str(self.update.message.from_user.id)
 
     def channel_name(self) -> str:
         return self.update.message.chat.title or "<DM>"

src/backend/telegram/instance.py

@@ -83,7 +83,7 @@ def _run(self, args) -> None:
             counter += 1
             if self._is_stopping:
                 log.info("Stopping Telegram instance...")
-                app.stop()
+                loop.run_until_complete(app.stop())
                 log.info("Telegram instance is stopped!")
                 break
             if counter % const.REMINDER_POLLING_INTERVAL == 0:

src/bc.py

@@ -58,15 +58,15 @@ class Telegram:
         def __init__(self) -> None:
             self.bot_username: 'Optional[str]' = None
             self.app: 'Optional[telegram.ext.Application]' = None
-            self.handlers: 'Dict[CommandHandler]' = dict()
+            self.handlers: 'Dict[str, CommandHandler]' = dict()
 
     class Repl:
         def __init__(self) -> None:
             pass
 
     class Backend:
         def __init__(self) -> None:
-            self._backends: Dict[str, bool] = dict(zip(
+            self._backends: Dict[const.BotBackend, bool] = dict(zip(
                 [backend for backend in const.BotBackend][1:], itertools.repeat(False)))
 
         def set_running(self, backend: const.BotBackend, new_state: bool, user_data_msg: str = "") -> None:
@@ -78,7 +78,7 @@ def set_running(self, backend: const.BotBackend, new_state: bool, user_data_msg:
                 log.info(f"Backend controller: {str(backend).title()} instance has stopped!")
             self._backends[backend] = new_state
 
-        def is_running(self, backend: const.BotBackend) -> None:
+        def is_running(self, backend: const.BotBackend) -> bool:
             return self._backends[backend]
 
     def __init__(self):
@@ -98,3 +98,4 @@ def __init__(self):
         self.plugin_manager = PluginManager(self.executor)
         self.message_cache = MessageCache()
         self.be = self.Backend()
+        self.guilds = None  # TODO: move to Discord