fix: stop using f-strings in logs

pyproject.toml

@@ -82,7 +82,7 @@ dev = [
   "pytest-cov>=7.0.0",
   "pytest-xdist[psutil]>=3.8.0",
   # lint
-  "ruff>=0.14.5",
+  "ruff>=0.16.5",
   # tests with all python versions
   "tox>=4.32.0",
   "tox-uv>=1.29.0",
@@ -161,7 +161,8 @@ lint.select = [
   "PL",  # PyLint checks
   "RUF",  # Specific to Ruff checks
   "FA102",  # Future annotations
-  "UP"  # Pyupgrade
+  "UP",  # Pyupgrade
+  "G",  # flake8-logging-format
 ]
 lint.ignore = [
   "D105",  # Missing docstring in magic method
@@ -172,6 +173,7 @@ lint.ignore = [
   "D100",  # Missing docstring in public module
   "ANN401",  # typing.Any are disallowed in `**kwargs
   "PLR0913",  # Too many arguments for function call
+  "PLR0917",  # Too many positional arguments
   "D106"  # Missing docstring in public nested class
 ]
 lint.mccabe = { max-complexity = 10 }

taskiq/abc/broker.py

@@ -177,8 +177,8 @@ def add_middlewares(self, *middlewares: "TaskiqMiddleware") -> None:
         for middleware in middlewares:
             if not isinstance(middleware, TaskiqMiddleware):
                 logger.warning(
-                    f"Middleware {middleware} is not an instance of TaskiqMiddleware. "
-                    "Skipping...",
+                    "Middleware %s is not an instance of TaskiqMiddleware. Skipping...",
+                    middleware,
                 )
                 continue
             middleware.set_broker(self)
@@ -464,8 +464,8 @@ def with_middlewares(
         for middleware in middlewares:
             if not isinstance(middleware, TaskiqMiddleware):
                 logger.warning(
-                    f"Middleware {middleware} is not an instance of TaskiqMiddleware. "
-                    "Skipping...",
+                    "Middleware %s is not an instance of TaskiqMiddleware. Skipping...",
+                    middleware,
                 )
                 continue
             middleware.set_broker(self)

taskiq/abc/middleware.py

@@ -24,7 +24,7 @@ def set_broker(self, broker: "AsyncBroker") -> None:
 
     def startup(
         self,
-    ) -> Union[None, Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]"]:
+    ) -> Union[Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]", None]:
         """
         Startup method to perform various action during startup.
 
@@ -36,7 +36,7 @@ def startup(
 
     def shutdown(
         self,
-    ) -> Union[None, Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]"]:
+    ) -> Union[Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]", None]:
         """
         Shutdown method to perform various action during shutdown.
 
@@ -68,7 +68,7 @@ def pre_send(
     def post_send(
         self,
         message: "TaskiqMessage",
-    ) -> Union[None, Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]"]:
+    ) -> Union[Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]", None]:
         """
         This hook is executed right after the task is sent.
 
@@ -101,7 +101,7 @@ def post_execute(
         self,
         message: "TaskiqMessage",
         result: "TaskiqResult[Any]",
-    ) -> Union[None, Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]"]:
+    ) -> Union[Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]", None]:
         """
         This hook executes after task is complete.
 
@@ -116,7 +116,7 @@ def post_save(
         self,
         message: "TaskiqMessage",
         result: "TaskiqResult[Any]",
-    ) -> Union[None, Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]"]:
+    ) -> Union[Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]", None]:
         """
         Post save hook.
 
@@ -132,7 +132,7 @@ def on_error(
         message: "TaskiqMessage",
         result: "TaskiqResult[Any]",
         exception: BaseException,
-    ) -> Union[None, Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]"]:
+    ) -> Union[Coroutine[Any, Any, None], "CoroutineType[Any, Any, None]", None]:
         """
         This function is called when exception is found.
 

taskiq/abc/schedule_source.py

@@ -58,7 +58,7 @@ async def delete_schedule(self, schedule_id: str) -> None:
     def pre_send(  # noqa: B027
         self,
         task: "ScheduledTask",
-    ) -> Union[None, "CoroutineType[Any, Any, None]", Coroutine[Any, Any, None]]:
+    ) -> Union["CoroutineType[Any, Any, None]", Coroutine[Any, Any, None], None]:
         """
         操作 to execute before task will be sent to broker.
 
@@ -71,7 +71,7 @@ def pre_send(  # noqa: B027
     def post_send(  # noqa: B027
         self,
         task: "ScheduledTask",
-    ) -> Union[None, "CoroutineType[Any, Any, None]", Coroutine[Any, Any, None]]:
+    ) -> Union["CoroutineType[Any, Any, None]", Coroutine[Any, Any, None], None]:
         """
         操作 to execute after task was sent to broker.
 

taskiq/acks.py

@@ -54,13 +54,13 @@ class AckableMessage(BaseModel):
     """
 
     data: bytes
-    ack: Callable[[], None | Awaitable[None]]
+    ack: Callable[[], Awaitable[None] | None]
 
 
 class AckController:
     """Controls acknowledgement state for a received message."""
 
-    def __init__(self, ack: Callable[[], None | Awaitable[None]] | None) -> None:
+    def __init__(self, ack: Callable[[], Awaitable[None] | None] | None) -> None:
         self._ack = ack
         self.is_acked = False
 

taskiq/cli/scheduler/run.py

@@ -47,13 +47,8 @@ async def get_schedules(source: ScheduleSource) -> list[ScheduledTask]:
     """
     try:
         return await source.get_schedules()
-    except Exception as exc:
-        logger.error(
-            "Cannot update schedules with source: %s\n%s{}",
-            source,
-            exc,
-            exc_info=True,
-        )
+    except Exception:
+        logger.exception("Cannot update schedules with source: %s", source)
         return []
 
 

taskiq/cli/utils.py

@@ -25,15 +25,15 @@ def add_cwd_in_path() -> Generator[None, None, None]:
     if str(cwd) in sys.path:
         yield
     else:
-        logger.debug(f"Inserting {cwd} in sys.path")
+        logger.debug("Inserting %s in sys.path", cwd)
         sys.path.insert(0, str(cwd))
         try:
             yield
         finally:
             try:
                 sys.path.remove(str(cwd))
             except ValueError:
-                logger.warning(f"Cannot remove '{cwd}' from sys.path")
+                logger.warning("Cannot remove '%s' from sys.path", cwd)
 
 
 def import_object(object_spec: str, app_dir: str | None = None) -> Any:
@@ -63,11 +63,11 @@ def import_from_modules(modules: list[str]) -> None:
     """
     for module in modules:
         try:
-            logger.info(f"Importing tasks from module {module}")
+            logger.info("Importing tasks from module %s", module)
             with add_cwd_in_path():
                 import_module(module)
         except ImportError as err:
-            logger.warning(f"Cannot import {module}. Cause:")
+            logger.warning("Cannot import %s. Cause:", module)
             logger.exception(err)
 
 

taskiq/cli/watcher.py

@@ -56,9 +56,11 @@ def dispatch(self, event: FileSystemEvent) -> None:
                 return
         except Exception as exc:
             logger.info(
-                f"Cannot check path `{event.src_path!r}` in gitignore. Cause: {exc}",
+                "Cannot check path `%r` in gitignore. Cause: %s",
+                event.src_path,
+                exc,
             )
             return
 
-        logger.debug(f"File changed. Event: {event}")
+        logger.debug("File changed. Event: %s", event)
         self.callback(**self.callback_kwargs)

taskiq/cli/worker/process_manager.py

@@ -77,7 +77,7 @@ def handle(
         try:
             worker.terminate()
         except ValueError:
-            logger.debug(f"Process {worker.name} is already terminated.")
+            logger.debug("Process %s is already terminated.", worker.name)
         # Waiting worker shutdown.
         worker.join()
         event: EventType = Event()
@@ -88,7 +88,11 @@ def handle(
             daemon=False,
         )
         new_process.start()
-        logger.info(f"Process {new_process.name} restarted with pid {new_process.pid}")
+        logger.info(
+            "Process %s restarted with pid %s",
+            new_process.name,
+            new_process.pid,
+        )
         workers[self.worker_num] = new_process
         _wait_for_worker_startup(new_process, event)
 
@@ -139,7 +143,7 @@ def _signal_handler(signum: int, _frame: Any) -> None:
         if current_process().name.startswith("worker"):
             raise KeyboardInterrupt
 
-        logger.debug(f"Got signal {signum}.")
+        logger.debug("Got signal %s.", signum)
         action_queue.put(action_to_send)
         logger.info("Workers are scheduled for shutdown.")
 
@@ -167,7 +171,7 @@ def __init__(
         if args.reload and observer is not None:
             watch_paths = args.reload_dirs if args.reload_dirs else ["."]
             for path_to_watch in watch_paths:
-                logger.debug(f"关注中 directory: {path_to_watch}")
+                logger.debug("关注中 directory: %s", path_to_watch)
                 observer.schedule(
                     FileWatcher(
                         callback=schedule_workers_reload,
@@ -267,7 +271,7 @@ def start(self) -> int | None:  # noqa: C901
             # We bulk_process all pending events.
             while not self.action_queue.empty():
                 action = self.action_queue.get()
-                logging.debug(f"Got event: {action}")
+                logging.debug("Got event: %s", action)
                 if isinstance(action, ReloadAllAction):
                     action.handle(
                         workers_num=len(self.workers),
@@ -295,7 +299,7 @@ def start(self) -> int | None:  # noqa: C901
 
             for worker_num, worker in enumerate(self.workers):
                 if not worker.is_alive():
-                    logger.info(f"{worker.name} is dead. Scheduling reload.")
+                    logger.info("%s is dead. Scheduling reload.", worker.name)
                     self.action_queue.put(
                         ReloadOneAction(
                             worker_num=worker_num,

taskiq/cli/worker/run.py

@@ -103,7 +103,7 @@ def interrupt_handler(signum: int, _frame: Any) -> None:
         :param _frame: current execution frame.
         :raises KeyboardInterrupt: if termination hasn't begun.
         """
-        logger.debug(f"Got signal {signum}.")
+        logger.debug("Got signal %s.", signum)
         nonlocal shutdown_event
         nonlocal hardkill_counter
         # Soft kill is a signal to start shutdown.

taskiq/kicker.py

@@ -154,7 +154,10 @@ async def kiq(
         :returns: taskiq task.
         """
         logger.debug(
-            f"Kicking {self.task_name} with args={args} and kwargs={kwargs}.",
+            "Kicking %s with args=%s and kwargs=%s.",
+            self.task_name,
+            args,
+            kwargs,
         )
         message = self._prepare_message(*args, **kwargs)
         for middleware in self.broker.middlewares:

taskiq/middlewares/prometheus_middleware.py

@@ -36,7 +36,7 @@ def __init__(
         if not metrics_path.exists():
             metrics_path.mkdir(parents=True)
 
-        logger.debug(f"Setting up multiproc dir to {metrics_path}")
+        logger.debug("Setting up multiproc dir to %s", metrics_path)
 
         os.environ["PROMETHEUS_MULTIPROC_DIR"] = str(metrics_path)
 

taskiq/receiver/receiver.py

@@ -170,7 +170,7 @@ async def callback(  # noqa: C901, PLR0912
                 exc_info=True,
             )
             return
-        logger.debug(f"Received message: {taskiq_msg}")
+        logger.debug("Received message: %s", taskiq_msg)
         task = self.broker.find_task(taskiq_msg.task_name)
         if task is None:
             logger.warning(
@@ -222,11 +222,7 @@ async def callback(  # noqa: C901, PLR0912
                         await maybe_awaitable(middleware.post_save(taskiq_msg, result))
 
         except Exception as exc:
-            logger.exception(
-                "Can't set result in result backend. Cause: %s",
-                exc,
-                exc_info=True,
-            )
+            logger.exception("Can't set result in result backend.")
             if raise_err:
                 raise exc
 
@@ -373,11 +369,7 @@ async def run_task(  # noqa: C901, PLR0912, PLR0915
             )
         except BaseException as exc:
             found_exception = exc
-            logger.error(
-                "Exception found while executing function: %s",
-                exc,
-                exc_info=True,
-            )
+            logger.exception("Exception found while executing function.")
         # Stop the timer.
         execution_time = time() - start_time
         if dep_ctx:
@@ -668,7 +660,8 @@ async def runner(
                     # asyncio.wait will throw an error if there is nothing to wait for
                     if tasks:
                         logger.info(
-                            f"Waiting for {len(tasks)} running tasks to complete...",
+                            "Waiting for %d running tasks to complete...",
+                            len(tasks),
                         )
                         await asyncio.wait(tasks, timeout=self.wait_tasks_timeout)
                         logger.info("No more tasks to wait for. Shutting down.")

taskiq/schedule_sources/label_based.py

@@ -37,8 +37,10 @@ async def startup(self) -> None:
             if task.broker != self.broker:
                 # if task broker doesn't match self, something is probably wrong
                 logger.warning(
-                    f"Broker for {task_name} `{task.broker}` doesn't "
-                    f"match scheduler's broker `{self.broker}`",
+                    "Broker for %s `%s` doesn't match scheduler's broker `%s`",
+                    task_name,
+                    task.broker,
+                    self.broker,
                 )
                 continue
             for schedule in task.labels.get("schedule", []):