Applied idiomatic Python improvements and micro-optimizations across registries and API layer.

owlbot/builtin_modules/clips/cache.py

@@ -287,14 +287,14 @@ async def _download_segment(
                 return False
             data = await seg_resp.read()
     except Exception:
-        ctx.logger.debug(f"Network error downloading segment {seq}.", exc_info=True)
+        ctx.logger.debug("Network error downloading segment %d.", seq, exc_info=True)
         return False
 
     chunk_path = cache.segment_path(seq)
     await asyncio.to_thread(chunk_path.write_bytes, data)
 
     cache.add_chunk(seq)
-    ctx.logger.debug(f"Cached segment {seq} ({len(data)} bytes).")
+    ctx.logger.debug("Cached segment %d (%d bytes).", seq, len(data))
     return True
 
 
@@ -344,7 +344,7 @@ async def _polling_loop(
         else 2.0
     )
     cache.set_target_duration(poll_interval)
-    ctx.logger.debug(f"Initial poll interval set to {poll_interval:.1f}s.")
+    ctx.logger.debug("Initial poll interval set to %.1fs.", poll_interval)
 
     pending_retries: set[int] = set()
 

owlbot/builtin_modules/clips/commands.py

@@ -127,7 +127,7 @@ async def delclip_command(ctx: CommandContext) -> None:
 
     row = await ctx.storage.fetch_one("SELECT id FROM clips WHERE id = ?", (clip_id,))
     if not row:
-        ctx.logger.debug(f"Clip {clip_id} not found for deletion.")
+        ctx.logger.debug("Clip %s not found for deletion.", clip_id)
         await ctx.owncast_client.send_message(f"Clip {clip_id} not found.")
         return
 

owlbot/builtin_modules/clips/processing.py

@@ -86,7 +86,7 @@ class ProcessingManager:
         async with self._ffprobe_semaphore:
             duration = await self._probe_duration(output_path)
 
-        self._logger.debug(f"Preview generated ({duration:.1f}s).")
+        self._logger.debug("Preview generated (%.1fs).", duration)
         return duration
 
     async def create_clip(
@@ -110,8 +110,11 @@ class ProcessingManager:
         """
         async with self._ffmpeg_semaphore:
             self._logger.debug(
-                f"Cutting clip {start:.1f}-{end:.1f}s "
-                f"from {preview_path} -> {output_path}."
+                "Cutting clip %.1f-%.1fs from %s -> %s.",
+                start,
+                end,
+                preview_path,
+                output_path,
             )
             await self._run_ffprocess(
                 "ffmpeg",
@@ -132,7 +135,7 @@ class ProcessingManager:
         async with self._ffprobe_semaphore:
             duration = await self._probe_duration(output_path)
 
-        self._logger.debug(f"Clip created ({duration:.1f}s).")
+        self._logger.debug("Clip created (%.1fs).", duration)
         return duration
 
     async def generate_thumbnail(
@@ -171,7 +174,7 @@ class ProcessingManager:
                 str(output_path),
             )
 
-        self._logger.debug(f"Thumbnail generated: {output_path}.")
+        self._logger.debug("Thumbnail generated: %s.", output_path)
 
     async def _run_ffprocess(
         self,

owlbot/builtin_modules/clips/routes.py

@@ -122,7 +122,7 @@ def schedule_session_expiry(
         await asyncio.sleep(delay)
         if sessions.get(token) is not session:
             return
-        logger.debug(f"Session expired: token={token[:8]}...")
+        logger.debug("Session expired: token=%s...", token[:8])
         cleanup_session(sessions, token)
 
     session.expiry_task = asyncio.create_task(_expire())
@@ -249,7 +249,7 @@ async def editor_submit(ctx: RouteContext) -> web.Response:
         max_length=max_length,
     )
     if error is not None:
-        ctx.logger.debug(f"Clip submit validation failed: {error}")
+        ctx.logger.debug("Clip submit validation failed: %s", error)
         return _error_page(ctx, 400, error)
 
     # Remove session to prevent double-submission.

owlbot/builtin_modules/custom_commands/placeholder_handlers.py

@@ -462,13 +462,12 @@ type PlaceholderHandler = Callable[
     [str, list[str], "PlaceholderContext"], Awaitable[str]
 ]
 
-HANDLERS: dict[str, PlaceholderHandler] = {}
-
-for _i in range(1, 10):
-    HANDLERS[str(_i)] = _evaluate_arg
-HANDLERS["user"] = _evaluate_user
-HANDLERS["count"] = _evaluate_count
-HANDLERS["getcount"] = _evaluate_getcount
-HANDLERS["rand"] = _evaluate_rand
-HANDLERS["countdown"] = _evaluate_countdown
-HANDLERS["countup"] = _evaluate_countdown
+HANDLERS: dict[str, PlaceholderHandler] = {
+    **{str(i): _evaluate_arg for i in range(1, 10)},
+    "user": _evaluate_user,
+    "count": _evaluate_count,
+    "getcount": _evaluate_getcount,
+    "rand": _evaluate_rand,
+    "countdown": _evaluate_countdown,
+    "countup": _evaluate_countdown,
+}