remove all f-strings from logs

Author: ChrisLovering

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