Update aiohappyeyeballs

dynamo/_evt_policy.py

@@ -3,7 +3,7 @@
 
 
 def get_event_loop_policy() -> asyncio.AbstractEventLoopPolicy:
-    if sys.platform in ("win32", "cygwin", "cli"):
+    if sys.platform in {"win32", "cygwin", "cli"}:
         try:
             import winloop
         except ImportError:

dynamo/_typing.py → dynamo/_types.py

@@ -1,7 +1,6 @@
 from collections.abc import Callable, Coroutine, Mapping
-from typing import Annotated, Any, ParamSpec, TypeVar
+from typing import Any, ParamSpec, TypeVar
 
-import numpy as np
 from discord import app_commands
 from discord.ext import commands
 from discord.ui import View
@@ -15,7 +14,6 @@
 S = TypeVar("S", bound=object)
 S_co = TypeVar("S_co", bound=object, covariant=True)
 
-ArrayRGB = Annotated[np.ndarray[Any, Any], tuple[int, int, int]]
 AC = TypeVar("AC", bound=Callable[..., Coroutine[Any, Any, Any]])
 CogT = TypeVar("CogT", bound=commands.Cog)
 CommandT = TypeVar("CommandT", bound=commands.Command[Any, ..., Any])

dynamo/core/__init__.py

@@ -1,6 +1,6 @@
 from dynamo.core.base_cog import DynamoCog
 from dynamo.core.bot import Dynamo
-from dynamo.core.logging import setup_logging
+from dynamo.core.logging_context import setup_logging
 
 __all__ = (
     "Dynamo",

dynamo/core/bot.py

@@ -236,11 +236,11 @@ async def on_ready(self) -> None:
 
         log.info("Ready: %s (ID: %s)", self.user, self.user.id)
 
-    async def get_context(
+    async def get_context[ContextT: commands.Context[Any]](
         self,
         origin: discord.Message | discord.Interaction,
         /,
         *,
-        cls: type[Context] = Context,
-    ) -> Context:
+        cls: type[ContextT] = Context,
+    ) -> ContextT:
         return await super().get_context(origin, cls=cls)

dynamo/extensions/cogs/errors.py

@@ -1,11 +1,12 @@
-from typing import Any, Callable, Coroutine
+from collections.abc import Callable, Coroutine
+from typing import Any
 
 import discord
 from discord import Interaction, app_commands
 from discord.ext import commands
 from rapidfuzz import fuzz
 
-from dynamo._typing import NotFoundWithHelp, app_command_error_messages, command_error_messages
+from dynamo._types import NotFoundWithHelp, app_command_error_messages, command_error_messages
 from dynamo.core import Dynamo, DynamoCog
 from dynamo.utils.context import Context
 
@@ -73,17 +74,17 @@ async def on_command_error(self, ctx: Context, error: commands.CommandError) ->
 
         error_message = self.get_command_error_message(error)
 
-        if isinstance(error, (commands.CommandNotFound, NotFoundWithHelp)):
+        if isinstance(error, commands.CommandNotFound | NotFoundWithHelp):
             invoked = ctx.invoked_with
             trigger: str = invoked if invoked and isinstance(error, commands.CommandNotFound) else error.args[0]
 
             matches = [
-                f"**{command.qualified_name}** - {command.short_doc or 'No description provided'}"
+                f"**{command.qualified_name}** - {command.short_doc or "No description provided"}"
                 for command in self.bot.commands
                 if fuzz.ratio(trigger, command.name) > 70
             ]
             matches_string = (
-                f"\n\nDid you mean \N{RIGHT-POINTING MAGNIFYING GLASS}\n>>> {'\n'.join(matches)}" if matches else ""
+                f"\n\nDid you mean \N{RIGHT-POINTING MAGNIFYING GLASS}\n>>> {"\n".join(matches)}" if matches else ""
             )
 
             error_message = error_message.format(trigger, matches_string)
@@ -116,7 +117,7 @@ async def on_app_command_error(self, interaction: Interaction, error: app_comman
             ]
             msg = f"Command not found: '{command_name}'"
             if matches:
-                msg += f"\n\nDid you mean \N{RIGHT-POINTING MAGNIFYING GLASS}\n>>> {'\n'.join(matches)}"
+                msg += f"\n\nDid you mean \N{RIGHT-POINTING MAGNIFYING GLASS}\n>>> {"\n".join(matches)}"
 
             await interaction.response.send_message(
                 ephemeral=True,

dynamo/extensions/cogs/events.py

@@ -2,7 +2,6 @@
 
 import contextlib
 from collections.abc import AsyncGenerator
-from copy import copy
 from typing import Any, cast
 
 import discord
@@ -52,18 +51,13 @@ def __init__(
         self.author_id: int = author_id
         self.add_item(dropdown(events))
 
-    @property
-    def __children(self) -> list[EventsDropdown[EventsView]]:
-        return getattr(self, "_children", []).copy()
-
     async def interaction_check(self, interaction: discord.Interaction) -> bool:
         return bool(interaction.user and interaction.user.id == self.author_id)
 
     async def on_timeout(self) -> None:
-        for item in self.__children:
-            new_item = copy(item)
-            new_item.disabled = True
-            self.add_item(item)
+        for i in self.children:
+            item = cast(EventsDropdown[EventsView], i)
+            item.disabled = True
         await self.message.edit(view=self)
 
 
@@ -95,7 +89,7 @@ async def get_interested(event: discord.ScheduledEvent) -> str:
     # https://peps.python.org/pep-0533/
     async with contextlib.aclosing(cast(AsyncGenerator[discord.User], event.users())) as gen:
         users: list[discord.User] = [u async for u in gen]
-    return f"`[{event.name}]({event.url}) {' '.join(u.mention for u in users) or "No users found"}`"
+    return f"`[{event.name}]({event.url}) {" ".join(u.mention for u in users) or "No users found"}`"
 
 
 class Events(DynamoCog):
@@ -157,7 +151,7 @@ async def event(self, ctx: Context, event: int | None = None) -> None:
         if ctx.guild is None:
             return
 
-        message = await ctx.send(f"{self.bot.app_emojis.get('loading2', '⏳')}\tFetching events...")
+        message = await ctx.send(f"{self.bot.app_emojis.get("loading2", "⏳")}\tFetching events...")
 
         event_check: str | list[discord.ScheduledEvent] = await self.event_check(ctx.guild, event)
         if isinstance(event_check, str):

dynamo/extensions/cogs/general.py

@@ -6,7 +6,6 @@
 
 from dynamo.core import Dynamo, DynamoCog
 from dynamo.utils import spotify
-from dynamo.utils.cache import async_cache
 from dynamo.utils.context import Context
 from dynamo.utils.converter import SeedConverter
 from dynamo.utils.format import human_join
@@ -19,7 +18,6 @@ class General(DynamoCog):
     def __init__(self, bot: Dynamo) -> None:
         super().__init__(bot)
 
-    @async_cache
     async def generate_identicon(
         self, seed: discord.Member | str | int, guild: discord.Guild | None
     ) -> tuple[discord.Embed, discord.File]:
@@ -43,7 +41,7 @@ async def generate_identicon(
         if isinstance(seed_to_use, str) and (parsed := urlparse(seed_to_use)).scheme and parsed.netloc:
             seed_to_use = (parsed.netloc + parsed.path).replace("/", "-")
 
-        display_name = seed_to_use if (isinstance(seed_to_use, (str, int))) else seed_to_use.display_name
+        display_name = seed_to_use if (isinstance(seed_to_use, str | int)) else seed_to_use.display_name
 
         seed_to_use = derive_seed(display_name)
         fg, bg = get_colors(seed=seed_to_use)
@@ -121,7 +119,7 @@ async def spotify(
         embed = discord.Embed(
             title=f"{spotify_emoji} Now Playing",
             description=f"{user.mention} is listening to **{track}** by"
-            f" **{human_join(activity.artists, conjunction='and')}**",
+            f" **{human_join(activity.artists, conjunction="and")}**",
             color=activity.color,
         )
         embed.set_footer(text=f"Requested by {ctx.author!s}", icon_url=ctx.author.display_avatar.url)

dynamo/extensions/cogs/help.py

@@ -1,11 +1,12 @@
 import logging
+from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Any, Mapping, Never, TypedDict
+from typing import Any, Never, TypedDict
 
 import discord
 from discord.ext import commands
 
-from dynamo._typing import CogT, CommandT, NotFoundWithHelp
+from dynamo._types import CogT, CommandT, NotFoundWithHelp
 from dynamo.core import Dynamo, DynamoCog
 from dynamo.utils.format import code_block, human_join
 
@@ -58,7 +59,7 @@ async def send_bot_help(self, mapping: Mapping[CogT | None, list[CommandT]]) ->
 
         await self.send(embed=embed)
 
-    def command_not_found(self, string: str) -> Never:
+    def command_not_found(self, string: str) -> Never:  # noqa: PLR6301
         log.debug("Command not found: %s", string)
         raise NotFoundWithHelp(string)
 
@@ -75,7 +76,7 @@ async def send_command_help(self, command: commands.Command[CogT, ..., Any]) ->
         description = command.help or "No help found..."
         embed = HelpEmbed(title=command.qualified_name, description=code_block(description))
 
-        embed.add_field(name="Aliases", value=f"`{human_join(command.aliases) or 'N/A'}`")
+        embed.add_field(name="Aliases", value=f"`{human_join(command.aliases) or "N/A"}`")
 
         cog = command.cog
         if cog and cog.qualified_name not in self.blacklisted:
@@ -100,18 +101,18 @@ async def send_help_embed(
         embed = HelpEmbed(title=title, description=description or "No help found...")
 
         if aliases:
-            embed.add_field(name="Aliases", value=f"`{human_join(aliases) or 'None'}`")
+            embed.add_field(name="Aliases", value=f"`{human_join(aliases) or "None"}`")
 
         if category:
             embed.add_field(name="Category", value=category)
 
         if filtered_commands := await self.filter_commands(commands):
             sub_commands = [
-                f"**{command.name}** - {command.help or 'No help found...'}" for command in filtered_commands
+                f"**{command.name}** - {command.help or "No help found..."}" for command in filtered_commands
             ]
             subcommand_field_title = "Commands" if title.endswith("Category") else "Subcommands"
             if sub_commands:
-                embed.add_field(name=subcommand_field_title, value=f">>> {'\n'.join(sub_commands)}", inline=False)
+                embed.add_field(name=subcommand_field_title, value=f">>> {"\n".join(sub_commands)}", inline=False)
 
         await self.send(embed=embed)
 

dynamo/extensions/cogs/info.py

@@ -9,7 +9,7 @@
 
 from dynamo.core import Dynamo, DynamoCog
 from dynamo.utils.context import Context
-from dynamo.utils.time import format_relative, human_timedelta
+from dynamo.utils.time_utils import format_relative, human_timedelta
 
 PYTHON = "https://s3.dualstack.us-east-2.amazonaws.com/pythondotorg-assets/media/community/logos/python-logo-only.png"
 
@@ -22,7 +22,8 @@ def __init__(self, bot: Dynamo) -> None:
 
         self.process = psutil.Process()
 
-    def format_commit(self, commit: pygit2.Commit) -> str:
+    @staticmethod
+    def format_commit(commit: pygit2.Commit) -> str:
         """A formatted commit message [`hash`](url) message (offset)"""
         short, *_ = commit.message.partition("\n")
         commit_tz = datetime.timezone(datetime.timedelta(minutes=commit.commit_time_offset))
@@ -36,7 +37,8 @@ def get_latest_commits(self, count: int = 3) -> str:
         commits = list(itertools.islice(repo.walk(repo.head.target, pygit2.enums.SortMode.TOPOLOGICAL), count))
         return "\n".join(self.format_commit(c) for c in commits)
 
-    def embed_from_user(self, user: discord.Member | discord.User) -> discord.Embed:
+    @staticmethod
+    def embed_from_user(user: discord.Member | discord.User) -> discord.Embed:
         embed = discord.Embed(color=user.color)
         embed.set_footer(text=f"ID: {user.id}")
         avatar = user.display_avatar.with_static_format("png")

dynamo/launcher.py

@@ -24,15 +24,14 @@
 
 def run_bot() -> None:
     loop = asyncio.new_event_loop()
-    loop.set_task_factory(asyncio.eager_task_factory)
     policy = get_event_loop_policy()
     asyncio.set_event_loop_policy(policy)
     asyncio.set_event_loop(loop)
+    loop.set_task_factory(asyncio.eager_task_factory)
 
     # https://github.com/aio-libs/aiohttp/issues/8599
     # https://github.com/mikeshardmind/salamander-reloaded
     connector = aiohttp.TCPConnector(
-        happy_eyeballs_delay=None,
         family=socket.AddressFamily.AF_INET,
         ttl_dns_cache=60,
         loop=loop,
@@ -52,8 +51,8 @@ async def entrypoint() -> None:
                 await bot.close()
 
     try:
-        loop.add_signal_handler(signal.SIGINT, lambda: loop.stop())
-        loop.add_signal_handler(signal.SIGTERM, lambda: loop.stop())
+        loop.add_signal_handler(signal.SIGINT, loop.stop)
+        loop.add_signal_handler(signal.SIGTERM, loop.stop)
     except NotImplementedError:
         pass
 
@@ -99,13 +98,11 @@ async def limited_finalization() -> None:
         for task in tasks:
             try:
                 if (exc := task.exception()) is not None:
-                    loop.call_exception_handler(
-                        {
-                            "message": "Unhandled exception in task during shutdown.",
-                            "exception": exc,
-                            "task": task,
-                        }
-                    )
+                    loop.call_exception_handler({
+                        "message": "Unhandled exception in task during shutdown.",
+                        "exception": exc,
+                        "task": task,
+                    })
             except (asyncio.InvalidStateError, asyncio.CancelledError):
                 pass