fix: replace deprecated getLevelName with IntEnum for log levels (#506)

Author: ruitcatarino

taskiq/cli/common_args.py

@@ -1,11 +1,13 @@
 import enum
+import logging
 
 
-class LogLevel(str, enum.Enum):
+class LogLevel(enum.IntEnum):
     """Different log levels."""
 
-    INFO = "INFO"
-    WARNING = "WARNING"
-    DEBUG = "DEBUG"
-    ERROR = "ERROR"
-    FATAL = "FATAL"
+    INFO = logging.INFO
+    WARNING = logging.WARNING
+    DEBUG = logging.DEBUG
+    ERROR = logging.ERROR
+    CRITICAL = logging.CRITICAL
+    FATAL = logging.FATAL

taskiq/cli/scheduler/args.py

@@ -12,7 +12,7 @@ class SchedulerArgs:
 
     scheduler: Union[str, TaskiqScheduler]
     modules: List[str]
-    log_level: str = LogLevel.INFO.name
+    log_level: LogLevel = LogLevel.INFO
     configure_logging: bool = True
     fs_discover: bool = False
     tasks_pattern: Sequence[str] = ("**/tasks.py",)
@@ -96,4 +96,6 @@ def from_cli(cls, args: Optional[Sequence[str]] = None) -> "SchedulerArgs":
         # This is an argparse limitation.
         if len(namespace.tasks_pattern) > 1:
             namespace.tasks_pattern.pop(0)
+        # Convert log_level string to LogLevel enum
+        namespace.log_level = LogLevel[namespace.log_level]
         return cls(**namespace.__dict__)

taskiq/cli/scheduler/run.py

@@ -2,7 +2,7 @@
 import inspect
 import sys
 from datetime import datetime, timedelta, timezone
-from logging import basicConfig, getLevelName, getLogger
+from logging import basicConfig, getLogger
 from typing import Any, Dict, List, Optional, Set, Tuple
 
 import pytz
@@ -239,14 +239,14 @@ async def run_scheduler(args: SchedulerArgs) -> None:
     """
     if args.configure_logging:
         basicConfig(
-            level=getLevelName(args.log_level),
+            level=args.log_level,
             format=(
                 "[%(asctime)s][%(levelname)-7s]"
                 "[%(module)s:%(funcName)s:%(lineno)d]"
                 " %(message)s"
             ),
         )
-    getLogger("taskiq").setLevel(level=getLevelName(args.log_level))
+    getLogger("taskiq").setLevel(level=args.log_level)
 
     if isinstance(args.scheduler, str):
         scheduler = import_object(args.scheduler)

taskiq/cli/worker/args.py

@@ -261,4 +261,6 @@ def from_cli(
         # This is an argparse limitation.
         if len(namespace.tasks_pattern) > 1:
             namespace.tasks_pattern.pop(0)
+        # Convert log_level string to LogLevel enum
+        namespace.log_level = LogLevel[namespace.log_level]
         return WorkerArgs(**namespace.__dict__)

taskiq/cli/worker/run.py

@@ -88,7 +88,7 @@ def start_listen(args: WorkerArgs) -> None:
     hardkill_counter = 0
     if args.configure_logging and get_start_method() == "spawn":
         logging.basicConfig(
-            level=logging.getLevelName(args.log_level),
+            level=args.log_level,
             format=args.log_format,
         )
 
@@ -190,10 +190,10 @@ def run_worker(args: WorkerArgs) -> Optional[int]:
         set_start_method("spawn")
     if args.configure_logging:
         logging.basicConfig(
-            level=logging.getLevelName(args.log_level),
+            level=args.log_level,
             format=args.log_format,
         )
-    logging.getLogger("taskiq").setLevel(level=logging.getLevelName(args.log_level))
+    logging.getLogger("taskiq").setLevel(level=args.log_level)
     logging.getLogger("watchdog.observers.inotify_buffer").setLevel(level=logging.INFO)
     logger.info("Pid of a main process: %s", str(os.getpid()))
     logger.info("Starting %s worker processes.", args.workers)