Allow list of rotation conditions in file rotation logic (#1175)

loguru/__init__.pyi

@@ -233,7 +233,16 @@ class Logger:
         enqueue: bool = ...,
         context: Optional[Union[str, BaseContext]] = ...,
         catch: bool = ...,
-        rotation: Optional[Union[str, int, time, timedelta, RotationFunction]] = ...,
+        rotation: Optional[
+            Union[
+                str,
+                int,
+                time,
+                timedelta,
+                RotationFunction,
+                list[Union[str, int, time, timedelta, RotationFunction]],
+            ]
+        ] = ...,
         retention: Optional[Union[str, int, timedelta, RetentionFunction]] = ...,
         compression: Optional[Union[str, CompressionFunction]] = ...,
         delay: bool = ...,

loguru/_file_sink.py

@@ -153,6 +153,13 @@ def __call__(self, message, file):
                 return True
             return False
 
+    class RotationGroup:
+        def __init__(self, rotations) -> None:
+            self._rotations = rotations
+
+        def __call__(self, message, file) -> bool:
+            return any(rotation(message, file) for rotation in self._rotations)
+
 
 class FileSink:
     def __init__(
@@ -310,6 +317,12 @@ def _make_glob_patterns(path):
     def _make_rotation_function(rotation):
         if rotation is None:
             return None
+        if isinstance(rotation, (list, tuple, set)):
+            if len(rotation) == 0:
+                raise ValueError("Must provide at least one rotation condition")
+            return Rotation.RotationGroup(
+                [FileSink._make_rotation_function(rot) for rot in rotation]
+            )
         if isinstance(rotation, str):
             size = string_parsers.parse_size(rotation)
             if size is not None:

loguru/_logger.py

@@ -314,9 +314,11 @@ def add(
 
         Parameters
         ----------
-        rotation : |str|, |int|, |time|, |timedelta| or |callable|_, optional
-            A condition indicating whenever the current logged file should be closed and a new one
-            started.
+        rotation : |str|, |int|, |time|, |timedelta|, |callable|_, or |list| of any of those
+            types, optional
+            Condition(s) indicating whenever the current logged file should be closed and a
+            new one started. If a list of conditions are provided, the current file is rotated
+            if any condition is true.
         retention : |str|, |int|, |timedelta| or |callable|_, optional
             A directive filtering old files that should be removed during rotation or end of
             program.

tests/test_filesink_rotation.py

@@ -1058,6 +1058,36 @@ def should_rotate(self, message, file):
     )
 
 
+def test_multiple_rotation_conditions(freeze_time, tmp_path):
+    with freeze_time("2020-01-01 20:00:00") as frozen:
+        logger.add(tmp_path / "file.log", rotation=["8 B", "1 min"], format="{message}")
+        logger.info("abcde")
+        frozen.tick()
+
+        logger.info("fghij")
+        frozen.tick()
+
+        logger.info("klm")
+        frozen.move_to("2020-01-01 20:01:01")
+
+        logger.info("no")
+
+    check_dir(
+        tmp_path,
+        files=[
+            ("file.2020-01-01_20-00-00_000000.log", "abcde\n"),
+            ("file.2020-01-01_20-00-01_000000.log", "fghij\n"),
+            ("file.2020-01-01_20-00-02_000000.log", "klm\n"),
+            ("file.log", "no\n"),
+        ],
+    )
+
+
+def test_empty_rotation_condition_list():
+    with pytest.raises(ValueError, match="^Must provide at least one rotation condition$"):
+        logger.add("test.log", rotation=[])
+
+
 @pytest.mark.parametrize(
     "rotation", [object(), os, datetime.date(2017, 11, 11), datetime.datetime.now(), 1j]
 )

tests/typesafety/test_logger.yml

@@ -287,7 +287,7 @@
     main:2: note: Possible overload variants:
     main:2: note:     def add(self, sink: Union[TextIO, Writable, Callable[[Message], None], Handler], *, level: Union[str, int] = ..., format: Union[str, Callable[[Record], str]] = ..., filter: Union[str, Callable[[Record], bool], Dict[Optional[str], Union[str, int, bool]], None] = ..., colorize: Optional[bool] = ..., serialize: bool = ..., backtrace: bool = ..., diagnose: bool = ..., enqueue: bool = ..., context: Union[str, BaseContext, None] = ..., catch: bool = ...) -> int
     main:2: note:     def add(self, sink: Callable[[Message], Awaitable[None]], *, level: Union[str, int] = ..., format: Union[str, Callable[[Record], str]] = ..., filter: Union[str, Callable[[Record], bool], Dict[Optional[str], Union[str, int, bool]], None] = ..., colorize: Optional[bool] = ..., serialize: bool = ..., backtrace: bool = ..., diagnose: bool = ..., enqueue: bool = ..., catch: bool = ..., context: Union[str, BaseContext, None] = ..., loop: Optional[AbstractEventLoop] = ...) -> int
-    main:2: note:     def add(self, sink: Union[str, PathLike[str]], *, level: Union[str, int] = ..., format: Union[str, Callable[[Record], str]] = ..., filter: Union[str, Callable[[Record], bool], Dict[Optional[str], Union[str, int, bool]], None] = ..., colorize: Optional[bool] = ..., serialize: bool = ..., backtrace: bool = ..., diagnose: bool = ..., enqueue: bool = ..., context: Union[str, BaseContext, None] = ..., catch: bool = ..., rotation: Union[str, int, time, timedelta, Callable[[Message, TextIO], bool], None] = ..., retention: Union[str, int, timedelta, Callable[[List[str]], None], None] = ..., compression: Union[str, Callable[[str], None], None] = ..., delay: bool = ..., watch: bool = ..., mode: str = ..., buffering: int = ..., encoding: str = ..., errors: Optional[str] = ..., newline: Optional[str] = ..., closefd: bool = ..., opener: Optional[Callable[[str, int], int]] = ...) -> int
+    main:2: note:     def add(self, sink: Union[str, PathLike[str]], *, level: Union[str, int] = ..., format: Union[str, Callable[[Record], str]] = ..., filter: Union[str, Callable[[Record], bool], Dict[Optional[str], Union[str, int, bool]], None] = ..., colorize: Optional[bool] = ..., serialize: bool = ..., backtrace: bool = ..., diagnose: bool = ..., enqueue: bool = ..., context: Union[str, BaseContext, None] = ..., catch: bool = ..., rotation: Union[str, int, time, timedelta, Callable[[Message, TextIO], bool], List[Union[str, int, time, timedelta, Callable[[Message, TextIO], bool]]], None] = ..., retention: Union[str, int, timedelta, Callable[[List[str]], None], None] = ..., compression: Union[str, Callable[[str], None], None] = ..., delay: bool = ..., watch: bool = ..., mode: str = ..., buffering: int = ..., encoding: str = ..., errors: Optional[str] = ..., newline: Optional[str] = ..., closefd: bool = ..., opener: Optional[Callable[[str, int], int]] = ...) -> int
 
 - case: invalid_logged_object_formatting
   main: |