Use log formatting over f strings

bot/bot.py

@@ -40,7 +40,7 @@ async def ping_services(self) -> None:
         attempts = 0
         while True:
             try:
-                log.info(f"Attempting site connection: {attempts + 1}/{constants.URLs.connect_max_retries}")
+                log.info("Attempting site connection: %s/%s", attempts + 1, constants.URLs.connect_max_retries)
                 await self.api_client.get("healthcheck")
                 break
 
@@ -55,7 +55,7 @@ async def setup_hook(self) -> None:
         await super().setup_hook()
         await self.load_extensions(exts)
 
-    async def on_error(self, event: str, *args, **kwargs) -> None:
+    async def on_error(self, event: str, /, *args, **kwargs) -> None:
         """Log errors raised in event listeners rather than printing them to stderr."""
         e_val = exception()
 
@@ -76,4 +76,4 @@ async def on_error(self, event: str, *args, **kwargs) -> None:
             scope.set_extra("args", args)
             scope.set_extra("kwargs", kwargs)
 
-            log.exception(f"Unhandled exception in {event}.")
+            log.exception("Unhandled exception in %s.", event)

bot/decorators.py

@@ -238,9 +238,9 @@ async def wrapper(*args, **kwargs) -> t.Any:
 
             if target.top_role >= actor.top_role:
                 log.info(
-                    f"{actor} ({actor.id}) attempted to {cmd} "
-                    f"{target} ({target.id}), who has an equal or higher top role."
-                )
+                    "%s (%s) attempted to %s "
+                    "%s (%s), who has an equal or higher top role.",
+                actor, actor.id, cmd, target, target.id)
                 await ctx.send(
                     f":x: {actor.mention}, you may not {cmd} "
                     "someone with an equal or higher top role."
@@ -266,7 +266,7 @@ def decorator(func: t.Callable) -> t.Callable:
         async def wrapped(*args, **kwargs) -> t.Any:
             """Short-circuit and log if in debug mode."""
             if DEBUG_MODE:
-                log.debug(f"Function {func.__name__} called with args: {args}, kwargs: {kwargs}")
+                log.debug("Function %s called with args: %s, kwargs: %s", func.__name__, args, kwargs)
                 return return_value
             return await func(*args, **kwargs)
         return wrapped

bot/exts/backend/branding/_cog.py

@@ -144,12 +144,12 @@ async def apply_asset(self, asset_type: AssetType, download_url: str) -> bool:
 
         Return a boolean indicating whether the application was successful.
         """
-        log.info(f"Applying '{asset_type.value}' asset to the guild.")
+        log.info("Applying '%s' asset to the guild.", asset_type.value)
 
         try:
             file = await self.repository.fetch_file(download_url)
         except Exception:
-            log.exception(f"Failed to fetch '{asset_type.value}' asset.")
+            log.exception("Failed to fetch '%s' asset.", asset_type.value)
             return False
 
         await self.bot.wait_until_guild_available()
@@ -163,7 +163,7 @@ async def apply_asset(self, asset_type: AssetType, download_url: str) -> bool:
             log.exception("Asset upload to Discord failed.")
             return False
         except TimeoutError:
-            log.error(f"Asset upload to Discord timed out after {timeout} seconds.")
+            log.error("Asset upload to Discord timed out after %s seconds.", timeout)
             return False
         else:
             log.trace("Asset uploaded successfully.")
@@ -182,17 +182,17 @@ async def rotate_assets(self, asset_type: AssetType) -> bool:
 
         Return a boolean indicating whether a new asset was applied successfully.
         """
-        log.debug(f"Rotating {asset_type.value}s.")
+        log.debug("Rotating %ss.", asset_type.value)
 
         state = await self.asset_caches[asset_type].to_dict()
         log.trace(f"Total {asset_type.value}s in rotation: {len(state)}.")
 
         if not state:  # This would only happen if rotation not initiated, but we can handle gracefully.
-            log.warning(f"Attempted {asset_type.value} rotation with an empty cache. This indicates wrong logic.")
+            log.warning("Attempted %s rotation with an empty cache. This indicates wrong logic.", asset_type.value)
             return False
 
         if len(state) == 1 and 1 in state.values():
-            log.debug(f"Aborting {asset_type.value} rotation: only 1 asset is available and has already been applied.")
+            log.debug("Aborting %s rotation: only 1 asset is available and has already been applied.", asset_type.value)
             return False
 
         current_iteration = min(state.values())  # Choose iteration to draw from.
@@ -219,7 +219,7 @@ async def maybe_rotate_assets(self, asset_type: AssetType) -> None:
         is work to be done before the timestamp is read and written, the next read will likely commence slightly
         under 24 hours after the last write.
         """
-        log.debug(f"Checking whether it's time for {asset_type.value}s to rotate.")
+        log.debug("Checking whether it's time for %ss to rotate.", asset_type.value)
 
         last_rotation_timestamp = await self.cache_information.get(f"last_{asset_type.value}_rotation_timestamp")
 
@@ -245,7 +245,7 @@ async def initiate_rotation(self, asset_type: AssetType, available_assets: list[
 
         This function does not upload a new asset!
         """
-        log.debug(f"Initiating new {asset_type.value} rotation.")
+        log.debug("Initiating new %s rotation.", asset_type.value)
 
         await self.asset_caches[asset_type].clear()
 
@@ -265,13 +265,17 @@ async def send_info_embed(self, channel_id: int, *, is_notification: bool) -> No
         We read event information from `cache_information`. The caller is therefore responsible for making
         sure that the cache is up-to-date before calling this function.
         """
-        log.debug(f"Sending event information event to channel: {channel_id} ({is_notification=}).")
+        log.debug(
+            "Sending event information event to channel: %s (is_notification=%r).",
+            channel_id,
+            is_notification,
+        )
 
         await self.bot.wait_until_guild_available()
         channel: discord.TextChannel | None = self.bot.get_channel(channel_id)
 
         if channel is None:
-            log.warning(f"Cannot send event information: channel {channel_id} not found!")
+            log.warning("Cannot send event information: channel %s not found!", channel_id)
             return
 
         log.trace(f"Destination channel: #{channel.name}.")
@@ -304,7 +308,7 @@ async def enter_event(self, event: Event) -> tuple[bool, bool]:
 
         Return a 2-tuple indicating whether the banner, and the icon, were applied successfully.
         """
-        log.info(f"Entering event: '{event.path}'.")
+        log.info("Entering event: '%s'.", event.path)
 
         # Prepare and apply new icon and banner rotations
         await self.initiate_rotation(AssetType.ICON, event.icons)
@@ -571,7 +575,7 @@ async def branding_calendar_group(self, ctx: commands.Context) -> None:
         first_25 = list(available_events.items())[:25]
 
         if len(first_25) != len(available_events):  # Alert core devs that a paginating solution is now necessary.
-            log.warning(f"There are {len(available_events)} events, but the calendar view can only display 25.")
+            log.warning("There are %s events, but the calendar view can only display 25.", len(available_events))
 
         for name, duration in first_25:
             embed.add_field(name=name[:256], value=duration[:1024])

bot/exts/backend/branding/_repository.py

@@ -136,7 +136,7 @@ async def fetch_directory(self, path: str, types: t.Container[str] = ("file", "d
         Passing custom `types` allows getting only files or directories. By default, both are included.
         """
         full_url = f"{BRANDING_URL}/{path}"
-        log.debug(f"Fetching directory from branding repository: '{full_url}'.")
+        log.debug("Fetching directory from branding repository: '%s'.", full_url)
 
         async with self.bot.http_session.get(full_url, params=PARAMS, headers=HEADERS) as response:
             _raise_for_status(response)
@@ -151,7 +151,7 @@ async def fetch_file(self, download_url: str) -> bytes:
 
         Raise an exception if the request does not succeed.
         """
-        log.debug(f"Fetching file from branding repository: '{download_url}'.")
+        log.debug("Fetching file from branding repository: '%s'.", download_url)
 
         async with self.bot.http_session.get(download_url, params=PARAMS, headers=HEADERS) as response:
             _raise_for_status(response)
@@ -245,7 +245,7 @@ async def get_current_event(self) -> tuple[Event, list[Event]]:
         Events are validated in the branding repo. The bot assumes that events are valid.
         """
         utc_now = datetime.now(tz=UTC)
-        log.debug(f"Finding active event for: {utc_now}.")
+        log.debug("Finding active event for: %s.", utc_now)
 
         # Construct an object in the arbitrary year for the purpose of comparison.
         lookup_now = date(year=ARBITRARY_YEAR, month=utc_now.month, day=utc_now.day)

bot/exts/backend/config_verifier.py

@@ -29,7 +29,7 @@ async def cog_load(self) -> None:
         ]
 
         if invalid_channels:
-            log.warning(f"Configured channels do not exist in server: {invalid_channels}.")
+            log.warning("Configured channels do not exist in server: %s.", invalid_channels)
 
 
 async def setup(bot: Bot) -> None:

bot/exts/backend/error_handler.py

@@ -370,7 +370,7 @@ async def handle_check_failure(ctx: Context, e: errors.CheckFailure) -> None:
     async def handle_api_error(ctx: Context, e: ResponseCodeError) -> None:
         """Send an error message in `ctx` for ResponseCodeError and log it."""
         if e.status == 404:
-            log.debug(f"API responded with 404 for command {ctx.command}")
+            log.debug("API responded with 404 for command %s", ctx.command)
             await ctx.send("There does not seem to be anything matching your query.")
             ctx.bot.stats.incr("errors.api_error_404")
         elif e.status == 400:
@@ -382,11 +382,11 @@ async def handle_api_error(ctx: Context, e: ResponseCodeError) -> None:
             await ctx.send("According to the API, your request is malformed.")
             ctx.bot.stats.incr("errors.api_error_400")
         elif 500 <= e.status < 600:
-            log.warning(f"API responded with {e.status} for command {ctx.command}")
+            log.warning("API responded with %s for command %s", e.status, ctx.command)
             await ctx.send("Sorry, there seems to be an internal issue with the API.")
             ctx.bot.stats.incr("errors.api_internal_server_error")
         else:
-            log.warning(f"Unexpected API response for command {ctx.command}: {e.status}")
+            log.warning("Unexpected API response for command %s: %s", ctx.command, e.status)
             await ctx.send(f"Got an unexpected status code from the API (`{e.status}`).")
             ctx.bot.stats.incr(f"errors.api_error_{e.status}")
 
@@ -418,7 +418,7 @@ async def handle_unexpected_error(ctx: Context, e: errors.CommandError) -> None:
                     f"https://discordapp.com/channels/{ctx.guild.id}/{ctx.channel.id}/{ctx.message.id}"
                 )
 
-            log.error(f"Error executing command invoked by {ctx.message.author}: {ctx.message.content}", exc_info=e)
+            log.error("Error executing command invoked by %s: %s", ctx.message.author, ctx.message.content, exc_info=e)
 
 
 async def setup(bot: Bot) -> None:

bot/exts/backend/sync/_syncers.py

@@ -52,7 +52,7 @@ async def sync(cls, guild: Guild, ctx: Context | None = None) -> None:
 
         If `ctx` is given, send a message with the results.
         """
-        log.info(f"Starting {cls.name} syncer.")
+        log.info("Starting %s syncer.", cls.name)
 
         if ctx:
             message = await ctx.send(f"📊 Synchronising {cls.name}s.")
@@ -63,7 +63,7 @@ async def sync(cls, guild: Guild, ctx: Context | None = None) -> None:
         try:
             await cls._sync(diff)
         except ResponseCodeError as e:
-            log.exception(f"{cls.name} syncer failed!")
+            log.exception("%s syncer failed!", cls.name)
 
             # Don't show response text because it's probably some really long HTML.
             results = f"status {e.status}\n```{e.response_json or 'See log output for details'}```"
@@ -73,7 +73,7 @@ async def sync(cls, guild: Guild, ctx: Context | None = None) -> None:
             results = (f"{name} `{len(val)}`" for name, val in diff_dict.items() if val is not None)
             results = ", ".join(results)
 
-            log.info(f"{cls.name} syncer finished: {results}.")
+            log.info("%s syncer finished: %s.", cls.name, results)
             content = f":ok_hand: Synchronisation of {cls.name}s complete: {results}"
 
         if message:

bot/exts/filtering/_filter_lists/antispam.py

@@ -50,7 +50,7 @@ def get_filter_type(self, content: str) -> type[UniqueFilter] | None:
             return antispam_filter_types[content]
         except KeyError:
             if content not in self._already_warned:
-                log.warning(f"An antispam filter named {content} was supplied, but no matching implementation found.")
+                log.warning("An antispam filter named %s was supplied, but no matching implementation found.", content)
                 self._already_warned.add(content)
             return None
 
@@ -124,7 +124,7 @@ async def process_deletion_context() -> None:
                 await asyncio.sleep(ALERT_DELAY)
 
                 if member not in self.message_deletion_queue:
-                    log.error(f"Started processing deletion queue for context `{member}`, but it was not found!")
+                    log.error("Started processing deletion queue for context `%s`, but it was not found!", member)
                     return
 
                 deletion_context = self.message_deletion_queue.pop(member)

bot/exts/filtering/_filter_lists/filter_list.py

@@ -222,7 +222,7 @@ def _create_filter(self, filter_data: dict, defaults: Defaults) -> T | None:
             if filter_type:
                 return filter_type(filter_data, defaults)
             if content not in self._already_warned:
-                log.warning(f"A filter named {content} was supplied, but no matching implementation found.")
+                log.warning("A filter named %s was supplied, but no matching implementation found.", content)
                 self._already_warned.add(content)
             return None
         except TypeError as e:

bot/exts/filtering/_filters/unique/discord_token.py

@@ -188,15 +188,15 @@ def is_valid_timestamp(b64_content: str) -> bool:
             decoded_bytes = base64.urlsafe_b64decode(b64_content)
             timestamp = int.from_bytes(decoded_bytes, byteorder="big")
         except ValueError as e:
-            log.debug(f"Failed to decode token timestamp '{b64_content}': {e}")
+            log.debug("Failed to decode token timestamp '%s': %s", b64_content, e)
             return False
 
         # Seems like newer tokens don't need the epoch added, but add anyway since an upper bound
         # is not checked.
         if timestamp + TOKEN_EPOCH >= DISCORD_EPOCH:
             return True
 
-        log.debug(f"Invalid token timestamp '{b64_content}': smaller than Discord epoch")
+        log.debug("Invalid token timestamp '%s': smaller than Discord epoch", b64_content)
         return False
 
     @staticmethod
@@ -210,8 +210,10 @@ def is_maybe_valid_hmac(b64_content: str) -> bool:
         unique = len(set(b64_content.lower()))
         if unique <= 3:
             log.debug(
-                f"Considering the HMAC {b64_content} a dummy because it has {unique}"
-                " case-insensitively unique characters"
+                "Considering the HMAC %s a dummy because it has %s"
+                " case-insensitively unique characters",
+                b64_content,
+                unique,
             )
             return False
         return True

bot/exts/filtering/_settings.py

@@ -39,7 +39,8 @@ def create_settings(
             validation_data[entry_name] = entry_data
         elif entry_name not in _already_warned:
             log.warning(
-                f"A setting named {entry_name} was loaded from the database, but no matching class."
+                "A setting named %s was loaded from the database, but no matching class.",
+                entry_name,
             )
             _already_warned.add(entry_name)
     if defaults is None:
@@ -80,9 +81,9 @@ def __init__(self, settings_data: dict, *, defaults: Settings | None = None, kee
             except KeyError:
                 if entry_name not in self._already_warned:
                     log.warning(
-                        f"A setting named {entry_name} was loaded from the database, "
-                        f"but no matching {self.entry_type.__name__} class."
-                    )
+                        "A setting named %s was loaded from the database, "
+                        "but no matching %s class.",
+                    entry_name, self.entry_type.__name__)
                     self._already_warned.add(entry_name)
             else:
                 try:

bot/exts/filtering/_settings_types/actions/infraction_and_notification.py

@@ -94,7 +94,7 @@ async def invoke(
         command = bot_module.instance.get_command(command_name)
         if not command:
             await alerts_channel.send(f":warning: Could not apply {command_name} to {user.mention}: command not found.")
-            log.warning(f":warning: Could not apply {command_name} to {user.mention}: command not found.")
+            log.warning(":warning: Could not apply %s to %s: command not found.", command_name, user.mention)
             return None
 
         if isinstance(user, discord.User):  # For example because a message was sent in a DM.
@@ -103,8 +103,10 @@ async def invoke(
                 user = member
             else:
                 log.warning(
-                    f"The user {user} were set to receive an automatic {command_name}, "
-                    "but they were not found in the guild."
+                    "The user %s were set to receive an automatic %s, "
+                    "but they were not found in the guild.",
+                    user,
+                    command_name,
                 )
                 return None
 
@@ -200,14 +202,14 @@ async def action(self, ctx: FilterContext) -> None:
             if self.infraction_channel:
                 channel = bot_module.instance.get_channel(self.infraction_channel)
                 if not channel:
-                    log.info(f"Could not find a channel with ID {self.infraction_channel}, infracting in mod-alerts.")
+                    log.info("Could not find a channel with ID %s, infracting in mod-alerts.", self.infraction_channel)
                     channel = alerts_channel
             elif not ctx.channel:
                 channel = alerts_channel
             else:
                 channel = ctx.channel
             if not channel:  # If somehow it's set to `alerts_channel` and it can't be found.
-                log.error(f"Unable to apply infraction as the context channel {channel} can't be found.")
+                log.error("Unable to apply infraction as the context channel %s can't be found.", channel)
                 return
 
             infraction_action = await self.infraction_type.invoke(

bot/exts/filtering/_settings_types/actions/remove_context.py

@@ -96,13 +96,13 @@ async def _handle_nickname(ctx: FilterContext) -> None:
         """Apply a superstar infraction to remove the user's nickname."""
         alerts_channel = bot.instance.get_channel(Channels.mod_alerts)
         if not alerts_channel:
-            log.error(f"Unable to apply superstar as the context channel {alerts_channel} can't be found.")
+            log.error("Unable to apply superstar as the context channel %s can't be found.", alerts_channel)
             return
         command = bot.instance.get_command("superstar")
         if not command:
             user = ctx.author
             await alerts_channel.send(f":warning: Could not apply superstar to {user.mention}: command not found.")
-            log.warning(f":warning: Could not apply superstar to {user.mention}: command not found.")
+            log.warning(":warning: Could not apply superstar to %s: command not found.", user.mention)
             ctx.action_descriptions.append("failed to superstar")
             return