refactor: Auto code audit

Tools used:
Sorcery
Ruff
Black

Author: Developer89899

interactions/api/events/discord.py

@@ -534,9 +534,7 @@ class MessageReactionAdd(BaseEvent):
     @property
     def reaction_count(self) -> int:
         """Times the emoji in the event has been used to react"""
-        if self.reaction is None:
-            return 0
-        return self.reaction.count
+        return 0 if self.reaction is None else self.reaction.count
 
 
 @attrs.define(eq=False, order=False, hash=False, kw_only=False)

interactions/api/events/internal.py

@@ -236,9 +236,7 @@ class ExtensionLoad(BaseEvent):
     @property
     def metadata(self) -> "Type[Extension.Metadata] | None":
         """The metadata of the extension, if it has any."""
-        if self.extension.Metadata:
-            return self.extension.Metadata
-        return None
+        return self.extension.Metadata or None
 
 
 @attrs.define(eq=False, order=False, hash=False, kw_only=True)

interactions/api/events/processors/guild_events.py

@@ -33,11 +33,7 @@ async def _on_raw_guild_create(self, event: "RawGatewayEvent") -> None:
             event: raw guild create event
 
         """
-        new_guild: bool = True
-        if self.cache.get_guild(event.data["id"]):
-            # guild already cached, most likely an unavailable guild coming back online
-            new_guild = False
-
+        new_guild = not self.cache.get_guild(event.data["id"])
         guild = self.cache.place_guild_data(event.data)
 
         self._user._guild_ids.add(to_snowflake(event.data.get("id")))

interactions/api/http/http_requests/channels.py

@@ -575,7 +575,7 @@ async def create_tag(
         payload: PAYLOAD_TYPE = {
             "name": name,
             "emoji_id": int(emoji_id) if emoji_id else None,
-            "emoji_name": emoji_name if emoji_name else None,
+            "emoji_name": emoji_name or None,
         }
         payload = dict_filter_none(payload)
 

interactions/api/http/http_requests/members.py

@@ -138,7 +138,7 @@ async def modify_current_member(
             reason: An optional reason for the audit log
 
         """
-        payload: PAYLOAD_TYPE = {"nick": nickname if not isinstance(nickname, Missing) else None}
+        payload: PAYLOAD_TYPE = {"nick": None if isinstance(nickname, Missing) else nickname}
         await self.request(
             Route("PATCH", f"/guilds/{int(guild_id)}/members/@me"),
             payload=payload,

interactions/api/voice/audio.py

@@ -220,10 +220,7 @@ def _max_buffer_size(self) -> int:
     @property
     def audio_complete(self) -> bool:
         """Uses the state of the subprocess to determine if more audio is coming"""
-        if self.process:
-            if self.process.poll() is None:
-                return False
-        return True
+        return not self.process or self.process.poll() is not None
 
     def _create_process(self, *, block: bool = True) -> None:
         before = (

interactions/api/voice/recorder.py

@@ -170,9 +170,7 @@ def run(self) -> None:
         # purge any data that is already in the socket
         readable, _, _ = select.select([sock], [], [], 0)
         log.debug("Purging socket buffer")
-        while readable:
-            if not sock.recv(4096):
-                break
+        while readable and sock.recv(4096):
             readable, _, _ = select.select([sock], [], [], 0)
         log.debug("Socket buffer purged, starting recording")
 

interactions/api/voice/voice_gateway.py

@@ -92,9 +92,7 @@ async def run(self) -> None:
 
             op = msg.get("op")
             data = msg.get("d")
-            seq = msg.get("s")
-
-            if seq:
+            if seq := msg.get("s"):
                 self.sequence = seq
 
             # This may try to reconnect the connection so it is best to wait

interactions/client/auto_shard_client.py

@@ -31,11 +31,7 @@ class AutoShardedClient(Client):
     """
 
     def __init__(self, *args, **kwargs) -> None:
-        if "total_shards" not in kwargs:
-            self.auto_sharding = True
-        else:
-            self.auto_sharding = False
-
+        self.auto_sharding = "total_shards" not in kwargs
         super().__init__(*args, **kwargs)
 
         self._connection_state = None

interactions/client/client.py

@@ -560,9 +560,12 @@ def _sanity_check(self) -> None:
     def _queue_task(self, coro: Listener, event: BaseEvent, *args, **kwargs) -> asyncio.Task:
         async def _async_wrap(_coro: Listener, _event: BaseEvent, *_args, **_kwargs) -> None:
             try:
-                if not isinstance(_event, (events.Error, events.RawGatewayEvent)):
-                    if coro.delay_until_ready and not self.is_ready:
-                        await self.wait_until_ready()
+                if (
+                    not isinstance(_event, (events.Error, events.RawGatewayEvent))
+                    and coro.delay_until_ready
+                    and not self.is_ready
+                ):
+                    await self.wait_until_ready()
 
                 if len(_event.__attrs_attrs__) == 2 and coro.event != "event":
                     # override_name & bot & logging
@@ -1046,11 +1049,7 @@ async def wait_for_modal(
         def predicate(event) -> bool:
             if modal.custom_id != event.ctx.custom_id:
                 return False
-            if not author:
-                return True
-            if author != to_snowflake(event.ctx.author):
-                return False
-            return True
+            return author == to_snowflake(event.ctx.author) if author else True
 
         resp = await self.wait_for("modal_completion", predicate, timeout)
         return resp.ctx
@@ -1085,7 +1084,7 @@ async def wait_for_component(
             asyncio.TimeoutError: if timed out
 
         """
-        if not (messages or components):
+        if not messages and not components:
             raise ValueError("You must specify messages or components (or both)")
 
         message_ids = (
@@ -1105,9 +1104,7 @@ def _check(event: events.Component) -> bool:
             )
             wanted_component = not custom_ids or ctx.custom_id in custom_ids
             if wanted_message and wanted_component:
-                if check is None or check(event):
-                    return True
-                return False
+                return bool(check is None or check(event))
             return False
 
         return await self.wait_for("component", checks=_check, timeout=timeout)
@@ -1242,7 +1239,7 @@ def add_interaction(self, command: InteractionCommand) -> bool:
 
             if group is None or isinstance(command, ContextMenu):
                 self.interaction_tree[scope][command.resolved_name] = command
-            elif group is not None:
+            else:
                 if not (current := self.interaction_tree[scope].get(base)) or isinstance(current, SlashCommand):
                     self.interaction_tree[scope][base] = {}
                 if sub is None:
@@ -1754,9 +1751,7 @@ def get_ext(self, name: str) -> Extension | None:
         Returns:
             A extension, if found
         """
-        if ext := self.get_extensions(name):
-            return ext[0]
-        return None
+        return ext[0] if (ext := self.get_extensions(name)) else None
 
     def __load_module(self, module, module_name, **load_kwargs) -> None:
         """Internal method that handles loading a module."""

interactions/client/const.py

@@ -121,10 +121,10 @@ def get_logger() -> logging.Logger:
 class Singleton(type):
     _instances = {}
 
-    def __call__(cls, *args, **kwargs) -> "Singleton":
-        if cls not in cls._instances:
-            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
-        return cls._instances[cls]
+    def __call__(self, *args, **kwargs) -> "Singleton":
+        if self not in self._instances:
+            self._instances[self] = super(Singleton, self).__call__(*args, **kwargs)
+        return self._instances[self]
 
 
 class Sentinel(metaclass=Singleton):

interactions/client/smart_cache.py

@@ -553,9 +553,7 @@ def get_dm_channel(self, user_id: Optional["Snowflake_Type"]) -> Optional["DM"]:
         """
         user_id = to_optional_snowflake(user_id)
         channel_id = self.dm_channels.get(user_id)
-        if channel_id is None:
-            return None
-        return self.get_channel(channel_id)
+        return None if channel_id is None else self.get_channel(channel_id)
 
     def delete_channel(self, channel_id: "Snowflake_Type") -> None:
         """

interactions/client/utils/attr_converters.py

@@ -55,9 +55,7 @@ def optional(converter: typing.Callable) -> typing.Any:
     """
 
     def optional_converter(val) -> typing.Any:
-        if val is None or val is MISSING:
-            return val
-        return converter(val)
+        return val if val is None or val is MISSING else converter(val)
 
     sig = None
     try:

interactions/client/utils/cache.py

@@ -152,9 +152,7 @@ class _CacheItemsView(ItemsView):
     def __contains__(self, item) -> bool:
         key, value = item
         v = self._mapping.get(key, default=attrs.NOTHING, reset_expiration=False)
-        if v is attrs.NOTHING:
-            return False
-        return v is value or v == value
+        return False if v is attrs.NOTHING else v is value or v == value
 
     def __iter__(self) -> Iterator[Tuple[Any, Any]]:
         for key in self._mapping:

interactions/client/utils/serializer.py

@@ -179,6 +179,6 @@ def get_file_mimetype(file_data: bytes) -> str:
         return "image/png"
     if file_data.startswith(b"\xff\xd8\xff"):
         return "image/jpeg"
-    if file_data[0:4] == b"RIFF" and file_data[8:12] == b"WEBP":
+    if file_data[:4] == b"RIFF" and file_data[8:12] == b"WEBP":
         return "image/webp"
     return "application/octet-stream"

interactions/ext/debug_extension/__init__.py

@@ -70,8 +70,7 @@ async def debug_info(self, ctx: InteractionContext) -> None:
 
         e.add_field("Start Time", f"{uptime.format(TimestampStyles.RelativeTime)}")
 
-        privileged_intents = [i.name for i in self.bot.intents if i in Intents.PRIVILEGED]
-        if privileged_intents:
+        if privileged_intents := [i.name for i in self.bot.intents if i in Intents.PRIVILEGED]:
             e.add_field("Privileged Intents", " | ".join(privileged_intents))
 
         e.add_field("Loaded Exts", ", ".join(self.bot.ext))

interactions/ext/debug_extension/debug_application_cmd.py

@@ -41,16 +41,13 @@ async def app_cmd(self, ctx: InteractionContext) -> None:
         await ctx.defer()
         e = debug_embed("Application-Commands Cache")
 
-        cmds = 0
-        for v in self.bot.interactions.values():
-            cmds += len(v.keys())
-
+        cmds = sum(len(v.keys()) for v in self.bot.interactions.values())
         e.add_field("Local application cmds (incld. Subcommands)", str(cmds))
         e.add_field("Component callbacks", str(len(self.bot._component_callbacks)))
         e.add_field("Prefixed commands", str(len(self.bot.prefixed_commands)))
         e.add_field(
             "Tracked Scopes",
-            str(len(Counter(scope for scope in self.bot._interaction_scopes.values()).keys())),
+            str(len(Counter(iter(self.bot._interaction_scopes.values())).keys())),
         )
 
         await ctx.send(embeds=[e])

interactions/ext/debug_extension/debug_exec.py

@@ -120,7 +120,7 @@ async def handle_exec_result(self, ctx: ModalContext, result: Any, value: Any, b
 
         if hasattr(result, "__iter__"):
             l_result = list(result)
-            if all([isinstance(r, Embed) for r in result]):
+            if all(isinstance(r, Embed) for r in result):
                 paginator = Paginator.create_from_embeds(self.bot, *l_result)
                 return await paginator.send(ctx)
 

interactions/ext/debug_extension/utils.py

@@ -79,9 +79,7 @@ def strf_delta(time_delta: datetime.timedelta, show_seconds: bool = True) -> str
         return f"{days_fmt} and {hours_fmt}"
     if hours >= 1:
         return f"{hours_fmt} and {minutes_fmt}"
-    if show_seconds:
-        return f"{minutes_fmt} and {seconds_fmt}"
-    return f"{minutes_fmt}"
+    return f"{minutes_fmt} and {seconds_fmt}" if show_seconds else f"{minutes_fmt}"
 
 
 def _make_solid_line(
@@ -166,7 +164,6 @@ def make_table(rows: list[list[Any]], labels: Optional[list[Any]] = None, center
     if labels is not None:
         lines.append(_make_data_line(column_widths, labels, data_left, data_middle, data_right, align))
         lines.append(_make_solid_line(column_widths, "├", "┼", "┤"))
-    for row in rows:
-        lines.append(_make_data_line(column_widths, row, data_left, data_middle, data_right, align))
+    lines.extend(_make_data_line(column_widths, row, data_left, data_middle, data_right, align) for row in rows)
     lines.append(_make_solid_line(column_widths, "╰", "┴", "╯"))
     return "\n".join(lines)