rabbithole of type checker appeasement

Author: Borketh

fred/cogs/crashes.py

@@ -78,8 +78,7 @@ async def make_sml_version_message(self, game_version: int = 0, sml: str = "", *
                 if latest_compatible_sml != sml_versions[0]:
                     msg += "\nAlso, your game itself may need an update!"
                 return CrashResponse("Outdated SML!", msg, inline=True)
-            else:
-                return None
+        return None
 
     # fmt: off
     _QUERY_TEMPLATE: Final[str] = """
@@ -406,8 +405,7 @@ async def process_message(self, message: Message) -> bool:
                 embed = createembed.crashes(filtered_responses)
                 embed.set_author(
                     name=f"Automated responses for {message.author.global_name or message.author.display_name} ({message.author.id})",
-                    icon_url=message.author.avatar and message.author.avatar.url,
-                    # defaults to None if no avatar, like mircea
+                    icon_url=message.author.display_avatar.url,
                 )
                 await self.bot.reply_to_msg(message, embed=embed, propagate_reply=False, files=resp_files)
             return True
@@ -519,19 +517,19 @@ def update_from_fg_log(self, log_file: IO[bytes]):
 
         info = self._get_fg_log_details(log_file)
 
-        if sml_version := info.get("sml"):
+        if sml_version := info.get("sml", ""):
             if self.sml_version and self.sml_version != sml_version:
                 self.mismatches.append(f"SML Version ({sml_version})")
             else:
                 self.sml_version = sml_version
 
-        if game_version := info.get("game_version"):
+        if game_version := info.get("game_version", ""):
             if self.game_version and int(float(self.game_version)) != int(float(game_version)):
                 self.mismatches.append(f"Game Version ({game_version})")
             else:
                 self.game_version = game_version
 
-        if path := info.get("path"):
+        if path := info.get("path", ""):
             if self.game_path:
                 p1 = path.replace("\\", "/")
                 p2 = self.game_path.replace("\\", "/")
@@ -543,13 +541,13 @@ def update_from_fg_log(self, log_file: IO[bytes]):
             else:
                 self.game_path = path
 
-        if launcher := info.get("launcher"):
+        if launcher := info.get("launcher", ""):
             if self.game_launcher_id and self.game_launcher_id.lower() != launcher.lower():
                 self.mismatches.append(f"Launcher ID: {launcher}")
             else:
                 self.game_launcher_id = launcher
 
-        if cli := info.get("cli"):
+        if cli := info.get("cli", ""):
             if self.game_command_line:
                 self.game_command_line += cli
 
@@ -591,10 +589,10 @@ def _get_fg_log_details(log_file: IO[bytes]):
                 info |= match.groupdict()
                 break
 
-        if cl := info.get("game_version"):
+        if cl := info.get("game_version", ""):
             info["game_version"] = int(cl)
 
-        if cli := info.get("cli"):
+        if cli := info.get("cli", ""):
             info["cli"] = filter_epic_commandline(cli)
 
         return info

fred/cogs/levelling.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
+import math
 from datetime import *
 
-import math
 from nextcord import DMChannel, Message, Guild
 
 from .. import config

fred/cogs/welcome.py

@@ -1,4 +1,4 @@
-from nextcord import Member
+from nextcord import Member, User
 
 from .. import config
 from ..libraries import common
@@ -12,7 +12,7 @@ async def on_member_join(self, member: Member):
 
         await self.send_welcome_message(member)
 
-    async def send_welcome_message(self, member: Member):
+    async def send_welcome_message(self, member: Member | User):
         if welcome := config.Misc.fetch("welcome_message"):
             self.logger.info("Sending the welcome message to a new member", extra=common.user_info(member))
             await self.bot.send_safe_direct_message(member, welcome)

fred/config.py

@@ -3,7 +3,7 @@
 import os
 import pathlib
 from numbers import Number
-from typing import Optional, Any
+from typing import Optional, Any, TypedDict
 
 import nextcord
 from sqlobject import SQLObject, IntCol, BoolCol, JSONCol, BigIntCol, StringCol, FloatCol, sqlhub
@@ -84,7 +84,7 @@ def fetch(user_id: int) -> Users:
         return Users.selectBy(user_id=user_id).getOne(None)
 
     @staticmethod
-    def create_if_missing(user: nextcord.User) -> Users:
+    def create_if_missing(user: nextcord.User | nextcord.Member) -> Users:
         return Users.selectBy(user_id=user.id).getOne(False) or Users(user_id=user.id)
 
 
@@ -112,26 +112,34 @@ def check(channel_id: int) -> bool:
         return bool(MediaOnlyChannels.selectBy(channel_id=channel_id).getOne(False))
 
 
-type CommandsOrCrashesDict = dict[str, str | StringCol]
+class CommandsDict(TypedDict):
+    name: str
+    content: str
+    attachment: str | None
+
+
+type CrashesDict = dict[str, str]
 
 
 class Commands(SQLObject):
     name = StringCol()
     content = StringCol()
     attachment = StringCol(default=None)
 
-    def as_dict(self) -> CommandsOrCrashesDict:
-        return dict(name=self.name, content=self.content, attachment=self.attachment)
+    def as_dict(self) -> CommandsDict:
+        return dict(
+            name=str(self.name), content=str(self.content), attachment=str(self.attachment) if self.attachment else None
+        )
 
     @staticmethod
-    def fetch(name: str) -> Optional[CommandsOrCrashesDict]:
+    def fetch(name: str) -> Optional[CommandsDict]:
         query: Optional[Commands]
         if query := Commands.selectBy(name=name.lower()).getOne(None):
             return query.as_dict()
         return None
 
     @classmethod
-    def fetch_by(cls, col: str, val: str) -> Optional[CommandsOrCrashesDict]:
+    def fetch_by(cls, col: str, val: str) -> Optional[CommandsDict]:
         # used by the search command to get a specific value if possible
         col, val = col.lower(), val.lower()
         if not isinstance(getattr(cls, col, None), property):
@@ -143,7 +151,7 @@ def fetch_by(cls, col: str, val: str) -> Optional[CommandsOrCrashesDict]:
         return query.as_dict()
 
     @staticmethod
-    def fetch_all() -> list[CommandsOrCrashesDict]:
+    def fetch_all() -> list[CommandsDict]:
         query = Commands.selectBy()
         return [cmd.as_dict() for cmd in query.lazyIter()]
 
@@ -153,18 +161,18 @@ class Crashes(SQLObject):
     crash = StringCol()
     response = StringCol()
 
-    def as_dict(self) -> CommandsOrCrashesDict:
-        return dict(name=self.name, response=self.response, crash=self.crash)
+    def as_dict(self) -> CrashesDict:
+        return dict(name=str(self.name), response=str(self.response), crash=str(self.crash))
 
     @staticmethod
-    def fetch(name: str) -> Optional[CommandsOrCrashesDict]:
+    def fetch(name: str) -> Optional[CrashesDict]:
         query: Optional[Crashes]
         if (query := Crashes.selectBy(name=name.lower()).getOne(None)) is not None:
             return query.as_dict()
         return None
 
     @classmethod
-    def fetch_by(cls, col: str, val: str) -> Optional[CommandsOrCrashesDict]:
+    def fetch_by(cls, col: str, val: str) -> Optional[CrashesDict]:
         col, val = col.lower(), val.lower()
         if not isinstance(getattr(cls, col), property):
             raise KeyError("This is not a valid column!")
@@ -175,7 +183,7 @@ def fetch_by(cls, col: str, val: str) -> Optional[CommandsOrCrashesDict]:
         return query.as_dict()
 
     @staticmethod
-    def fetch_all() -> list[CommandsOrCrashesDict]:
+    def fetch_all() -> list[CrashesDict]:
         query = Crashes.selectBy()
         return [crash.as_dict() for crash in query.lazyIter()]
 

fred/fred.py

@@ -53,10 +53,10 @@ def __init__(self, *args, **kwargs):
         self.setup_cogs()
         FredHelpEmbed.setup()
         self.owo = False
-        self.web_session: aiohttp.ClientSession = ...
+        self.web_session: aiohttp.ClientSession = ...  # type: ignore
         self.loop = asyncio.get_event_loop()
         self.executor = ThreadPoolExecutor()
-        self.error_channel = int(chan) if (chan := config.Misc.fetch("error_channel")) else 748229790825185311
+        self.error_channel: int = int(config.Misc.fetch("error_channel")) or 748229790825185311  # type: ignore
 
     async def start(self, *args, **kwargs):
         async with aiohttp.ClientSession() as session:
@@ -146,7 +146,10 @@ async def on_error(self, event_method: str, *args, **kwargs):
         # error_embed = nextcord.Embed(colour=nextcord.Colour.red(), title=error_meta, description=full_error)
         # error_embed.set_author(name=fred_str)
 
-        await self.get_channel(self.error_channel).send(f"**{fred_str}**\n{error_meta}\n```py\n{full_error}```")
+        error_channel = self.get_partial_messageable(self.error_channel)
+        if error_channel is None:
+            raise RuntimeError("Error channel unreachable! Value: " + str(self.error_channel))
+        await error_channel.send(f"**{fred_str}**\n{error_meta}\n```py\n{full_error}```")
 
     async def githook_send(self, data: dict):
         self.logger.info("Handling GitHub payload", extra={"data": data})
@@ -156,7 +159,7 @@ async def githook_send(self, data: dict):
             self.logger.info("Non-supported Payload received")
         else:
             self.logger.info("GitHub payload was supported, sending an embed")
-            channel = self.get_channel(config.Misc.fetch("githook_channel"))
+            channel = self.get_partial_messageable(config.Misc.fetch("githook_channel"))  # type: ignore
             await channel.send(content=None, embed=embed)
 
     async def _send_safe_direct_message_internal(
@@ -189,21 +192,19 @@ async def _send_safe_direct_message_internal(
             return False
 
         try:
+            dm_channel = user.dm_channel or await user.create_dm()
 
-            if not user.dm_channel:
-                await user.create_dm()
-
-            if not embed:
+            if not embed and content is not None:
                 embed = createembed.DM(content)
                 content = None
 
-            await self.safe_send(user.dm_channel, content, embed=embed, **kwargs)
+            await self.safe_send(dm_channel, content, embed=embed, **kwargs)
             return True
         except Exception:  # noqa
             self.logger.error(f"DMs: Failed to DM, reason: \n{traceback.format_exc()}")
             return False
 
-    async def send_safe_direct_message(self, user: nextcord.User, content=None, **kwargs) -> bool:
+    async def send_safe_direct_message(self, user: nextcord.User | nextcord.Member, content=None, **kwargs) -> bool:
         user_meta = config.Users.create_if_missing(user)
         try:
             return await self._send_safe_direct_message_internal(user, content, user_meta=user_meta, **kwargs)
@@ -213,18 +214,18 @@ async def send_safe_direct_message(self, user: nextcord.User, content=None, **kw
 
     @staticmethod
     async def safe_send(
-        chan: nextcord.TextChannel | nextcord.DMChannel,
+        to: nextcord.PartialMessageable | nextcord.abc.Messageable,
         content: Optional[str],
         *,
         files: Optional[list[nextcord.File]] = None,
         **kwargs,
     ) -> nextcord.Message:
         if content is not None and len(content) > 2000:
-            files = files or []
+            files: list[nextcord.File] = files or []
             files.append(text2file(content, filename="long-message.txt"))
             content = "Message too long, converted to text file!"
 
-        return await chan.send(content, files=files, **kwargs)
+        return await to.send(content=content, files=files, **kwargs)
 
     async def reply_to_msg(
         self,
@@ -268,7 +269,8 @@ def check(message2: nextcord.Message):
             return (message2.author == message.author) and (message2.channel == message.channel)
 
         try:
-            response: nextcord.Message = await self.wait_for("message", timeout=120.0, check=check)
+            maybe_response = self.wait_for("message", timeout=120.0, check=check)
+            response: nextcord.Message = await maybe_response
         except asyncio.TimeoutError:
             await self.reply_to_msg(message, "Timed out and aborted after 120 seconds.")
             raise asyncio.TimeoutError
@@ -317,7 +319,7 @@ async def on_message(self, message: nextcord.Message):
         if not removed:
             before, space, after = message.content.partition(" ")
             # if the prefix is the only thing before the space then this isn't a command
-            if before.startswith(self.command_prefix) and len(before) > 1:
+            if before.startswith(str(self.command_prefix)) and len(before) > 1:
                 self.logger.info("Processing commands")
                 await self.process_commands(message)
             else:

fred/fred_commands/_command_utils.py

@@ -2,7 +2,7 @@
 
 from regex import ENHANCEMATCH, escape, search as re_search
 
-from ..config import Commands, Crashes, Misc
+from ..config import Commands, Crashes, Misc, CommandsDict, CrashesDict
 from ..libraries.common import new_logger
 
 logger = new_logger("[Command/Crash Search]")
@@ -16,8 +16,12 @@ def search(
     if column not in dir(table):
         raise KeyError(f"`{column}` is not a column in the {table.__name__} table!")
 
+    if column == "attachment":
+        raise KeyError(f"You cannot search for attachments!")
+
     if not force_fuzzy and (exact_match := table.fetch_by(column, pattern)):
-        return exact_match[column], True
+        res: str = exact_match.get(column) or ""
+        return res, True
 
     if len(pattern) < 2:
         raise KeyError("Search pattern must be at least 2 characters long for fuzzy searching!")
@@ -28,7 +32,8 @@ def search(
 
     scored_results: list[tuple[int, str]] = []
     for item in table.fetch_all():
-        value = item.get(column)
+        item: CommandsDict | CrashesDict
+        value = item.get(column) or ""
 
         # Filter non matching strings
         if not isinstance(value, str):

fred/fred_commands/crashes.py

@@ -65,24 +65,24 @@ async def modify_crash(
             await ctx.send(f"Could not find a crash with name '{name}'. Aborting")
             return
 
+        checked_crash = new_crash or str(crash.crash)
+        checked_response = new_response or str(crash.response)
+
         try:
-            if change_crash := await self.bot.reply_yes_or_no(ctx.message, "Do you want to change the crash to match?"):
-                new_crash, _ = await self.bot.reply_question(
+            if await self.bot.reply_yes_or_no(ctx.message, "Do you want to change the crash to match?"):
+                checked_crash, _ = await self.bot.reply_question(
                     ctx.message, "What is the regular expression to match in the logs?"
                 )
 
-            if change_response := await self.bot.reply_yes_or_no(ctx.message, "Do you want to change the response?"):
-                new_response, _ = await self.bot.reply_question(
+            if await self.bot.reply_yes_or_no(ctx.message, "Do you want to change the response?"):
+                checked_response, _ = await self.bot.reply_question(
                     ctx.message,
                     f"What response do you want it to provide? Responding with `{self.bot.command_prefix}command_name`"
                     "will use the response of that command.",
                 )
         except ValueError:
             return
 
-        checked_crash = new_crash if change_crash else crash.crash
-        checked_response = new_response if change_response else crash.response
-
         issue = validate_crash(checked_crash, checked_response)
         if issue:
             await self.bot.reply_to_msg(ctx.message, issue)

fred/fred_commands/dbcommands.py

@@ -236,7 +236,7 @@ async def rename_alias(self, ctx: commands.Context, name: str.lower, *, new_name
         """Usage: `rename alias (name) (new name)`
         Purpose: Renames an alias.
         Notes: If response is not supplied you will be prompted for one with a timeout"""
-        await self.rename_command(ctx, name, new_name)
+        await self.rename_command(ctx, name, new_name=new_name)
 
     @BaseCmds.search.command(name="commands")
     async def search_commands(self, ctx: commands.Context, pattern: str, *, flags: SearchFlags) -> None:

fred/fred_commands/help.py

@@ -157,7 +157,7 @@ def get_shift(index: int, field_number: int) -> int:
     @lru_cache
     def get_field_indices(index: int, field_number: int) -> str:
         start = 1 + FredHelpEmbed.get_shift(index, field_number)
-        end = field_size + FredHelpEmbed.get_shift(index, field_number)
+        end = page_size + FredHelpEmbed.get_shift(index, field_number)
         return f"{start}-{end}"
 
     @classmethod