Change "rotation" option so it uses file creation time (Delgan#58)

CHANGELOG.rst

@@ -2,6 +2,7 @@ Unreleased
 ==========
 
 - Add a new ``logger.patch()`` method which can be used to modify the record dict on-the-fly before its being sent to the handlers
+- Change behavior of ``rotation`` option in file sinks: it is now based on the file creation time rather than the current time, note that propoer support may differ depending on your platform (`#58 <https://github.com/Delgan/loguru/issues/58>`_)
 - Raise errors on unkowns color tags rather than silently ignoring them (`#57 <https://github.com/Delgan/loguru/issues/57>`_)
 - Add the possibility to auto-close color tags by using ``</>`` (eg. ``<yellow>message</>``)
 - Remove colors tags mixing directives (eg. ``<red,blue>``) for simplification

loguru/_file_sink.py

@@ -4,6 +4,7 @@
 import locale
 import numbers
 import os
+import platform
 import shutil
 import string
 
@@ -43,17 +44,21 @@ def __init__(
         self._kwargs = kwargs.copy()
         self._path = str(path)
 
-        self._rotation_function = self._make_rotation_function(rotation)
+        self._rotation_function, self._init_rotation = self._make_rotation_function(rotation)
         self._retention_function = self._make_retention_function(retention)
         self._compression_function = self._make_compression_function(compression)
         self._glob_pattern = self._make_glob_pattern(self._path)
+        self._get_creation_time = self._make_get_creation_time_function()
+        self._set_creation_time = self._make_set_creation_time_function()
 
         self._file = None
         self._file_path = None
 
         if not delay:
             self._initialize_file(rename_existing=False)
             self._initialize_write()
+            if self._init_rotation is not None:
+                self._init_rotation(self._file_path)
         else:
             self.write = self._write_delayed
 
@@ -66,12 +71,16 @@ def _initialize_write(self):
     def _write_delayed(self, message):
         self._initialize_file(rename_existing=False)
         self._initialize_write()
+        if self._init_rotation is not None:
+            self._init_rotation(self._file_path)
         self.write(message)
 
     def _write_rotating(self, message):
         if self._rotation_function(message, self._file):
             self._terminate(teardown=True)
             self._initialize_file(rename_existing=True)
+            if self._set_creation_time is not None:
+                self._set_creation_time(self._file_path, now().replace(tzinfo=None).timestamp())
         self._file.write(message)
 
     def _initialize_file(self, *, rename_existing):
@@ -105,25 +114,65 @@ def _make_glob_pattern(path):
             pattern = root + "*"
         return pattern
 
+    def _make_get_creation_time_function(self):
+        def get_creation_time_windows(filepath):
+            return os.stat(filepath).st_ctime
+
+        def get_creation_time_darwin(filepath):
+            return os.stat(filepath).st_birthtime
+
+        def get_creation_time_linux(filepath):
+            try:
+                return float(os.getxattr(filepath, b"user.loguru_crtime"))
+            except OSError:
+                return os.stat(filepath).st_mtime
+
+        if platform.system().lower() == "windows":
+            return get_creation_time_windows
+        elif hasattr(os.stat_result, "st_birthtime"):
+            return get_creation_time_darwin
+        else:
+            return get_creation_time_linux
+
+    def _make_set_creation_time_function(self):
+        def set_creation_time(filepath, timestamp):
+            try:
+                os.setxattr(filepath, b"user.loguru_crtime", str(timestamp).encode("ascii"))
+            except OSError:
+                pass
+
+        if platform.system().lower() == "windows" or hasattr(os.stat_result, "st_birthtime"):
+            return None
+        else:
+            return set_creation_time
+
     def _make_rotation_function(self, rotation):
         def make_from_size(size_limit):
             def rotation_function(message, file):
                 file.seek(0, 2)
                 return file.tell() + len(message) >= size_limit
 
-            return rotation_function
+            return rotation_function, None
 
         def make_from_time(step_forward, time_init=None):
-            start_time = time_limit = now().replace(tzinfo=None)
-            if time_init is not None:
-                time_limit = time_limit.replace(
-                    hour=time_init.hour,
-                    minute=time_init.minute,
-                    second=time_init.second,
-                    microsecond=time_init.microsecond,
-                )
-            if time_limit <= start_time:
-                time_limit = step_forward(time_limit)
+
+            time_limit = None
+
+            def init_time_rotation(filepath):
+                nonlocal time_limit
+                creation_time = self._get_creation_time(filepath)
+                if self._set_creation_time is not None:
+                    self._set_creation_time(filepath, creation_time)
+                start_time = time_limit = datetime.datetime.fromtimestamp(creation_time)
+                if time_init is not None:
+                    time_limit = time_limit.replace(
+                        hour=time_init.hour,
+                        minute=time_init.minute,
+                        second=time_init.second,
+                        microsecond=time_init.microsecond,
+                    )
+                if time_limit <= start_time:
+                    time_limit = step_forward(time_limit)
 
             def rotation_function(message, file):
                 nonlocal time_limit
@@ -134,10 +183,10 @@ def rotation_function(message, file):
                     return True
                 return False
 
-            return rotation_function
+            return rotation_function, init_time_rotation
 
         if rotation is None:
-            return None
+            return None, None
         elif isinstance(rotation, str):
             size = string_parsers.parse_size(rotation)
             if size is not None:
@@ -179,7 +228,7 @@ def add_interval(t):
 
             return make_from_time(add_interval)
         elif callable(rotation):
-            return rotation
+            return rotation, None
         else:
             raise ValueError(
                 "Cannot infer rotation for objects of type: '%s'" % type(rotation).__name__

tests/test_filesink_rotation.py

@@ -1,9 +1,102 @@
 import pytest
 import datetime
 import os
+import time
+import tempfile
+import pathlib
+import platform
+import builtins
+from collections import namedtuple
+import loguru
 from loguru import logger
 
 
+@pytest.fixture
+def tmpdir_local():
+    # Pytest 'tmpdir' creates directories in /tmp, but /tmp does not support xattr, tests would fail
+    with tempfile.TemporaryDirectory(dir=".") as tempdir:
+        yield pathlib.Path(tempdir)
+
+
+@pytest.fixture
+def monkeypatch_filesystem(monkeypatch):
+    def monkeypatch_filesystem(raising=None, crtime=None, patch_xattr=False):
+        filesystem = {}
+        __open__ = open
+        __stat_result__ = os.stat_result
+        __stat__ = os.stat
+
+        class StatWrapper:
+            def __init__(self, wrapped, timestamp=None):
+                self._wrapped = wrapped
+                self._timestamp = timestamp
+
+            def __getattr__(self, name):
+                if name == raising:
+                    raise AttributeError
+                if name == crtime:
+                    return self._timestamp
+                return getattr(self._wrapped, name)
+
+        def patched_stat(filepath):
+            stat = __stat__(filepath)
+            wrapped = StatWrapper(stat, filesystem.get(os.path.abspath(filepath)))
+            return wrapped
+
+        def patched_open(filepath, *args, **kwargs):
+            if not os.path.exists(filepath):
+                filesystem[os.path.abspath(filepath)] = loguru._datetime.datetime.now().timestamp()
+            return __open__(filepath, *args, **kwargs)
+
+        def patched_setxattr(filepath, attr, val, *arg, **kwargs):
+            filesystem[(os.path.abspath(filepath), attr)] = val
+
+        def patched_getxattr(filepath, attr, *args, **kwargs):
+            try:
+                return filesystem[(os.path.abspath(filepath), attr)]
+            except KeyError:
+                raise OSError
+
+        monkeypatch.setattr(os, "stat_result", StatWrapper(__stat_result__))
+        monkeypatch.setattr(os, "stat", patched_stat)
+        monkeypatch.setattr(builtins, "open", patched_open)
+
+        if patch_xattr:
+            monkeypatch.setattr(os, "setxattr", patched_setxattr)
+            monkeypatch.setattr(os, "getxattr", patched_getxattr)
+
+    return monkeypatch_filesystem
+
+
+@pytest.fixture
+def windows_filesystem(monkeypatch, monkeypatch_filesystem):
+    monkeypatch.setattr(platform, "system", lambda: "Windows")
+    monkeypatch_filesystem(raising="st_birthtime", crtime="st_ctime")
+
+
+@pytest.fixture
+def darwin_filesystem(monkeypatch, monkeypatch_filesystem):
+    monkeypatch.setattr(platform, "system", lambda: "Darwin")
+    monkeypatch_filesystem(crtime="st_birthtime")
+
+
+@pytest.fixture
+def linux_filesystem(monkeypatch, monkeypatch_filesystem):
+    monkeypatch.setattr(platform, "system", lambda: "Linux")
+    monkeypatch_filesystem(raising="st_birthtime", crtime="st_mtime", patch_xattr=True)
+
+
+@pytest.fixture
+def linux_no_xattr_filesystem(monkeypatch, monkeypatch_filesystem):
+    def raising(*args, **kwargs):
+        raise OSError
+
+    monkeypatch.setattr(platform, "system", lambda: "Linux")
+    monkeypatch_filesystem(raising="st_birthtime", crtime="st_mtime")
+    monkeypatch.setattr(os, "setxattr", raising)
+    monkeypatch.setattr(os, "getxattr", raising)
+
+
 def test_renaming(monkeypatch_date, tmpdir):
     i = logger.add(str(tmpdir.join("file.log")), rotation=0, format="{message}")
 
@@ -118,7 +211,7 @@ def test_size_rotation(monkeypatch_date, tmpdir, size):
         ("Yearly ", [100, 24 * 7 * 30, 24 * 300, 24 * 100, 24 * 400]),
     ],
 )
-def test_time_rotation(monkeypatch_date, tmpdir, when, hours):
+def test_time_rotation(monkeypatch_date, darwin_filesystem, tmpdir, when, hours):
     now = datetime.datetime(2017, 6, 18, 12, 0, 0)  # Sunday
 
     monkeypatch_date(
@@ -127,7 +220,6 @@ def test_time_rotation(monkeypatch_date, tmpdir, when, hours):
 
     i = logger.add(str(tmpdir.join("test_{time}.log")), format="{message}", rotation=when, mode="w")
 
-
     for h, m in zip(hours, ["a", "b", "c", "d", "e"]):
         now += datetime.timedelta(hours=h)
         monkeypatch_date(
@@ -136,11 +228,10 @@ def test_time_rotation(monkeypatch_date, tmpdir, when, hours):
         logger.debug(m)
 
     logger.remove(i)
-    assert len(tmpdir.listdir()) == 4
     assert [f.read() for f in sorted(tmpdir.listdir())] == ["a\n", "b\nc\n", "d\n", "e\n"]
 
 
-def test_time_rotation_dst(monkeypatch_date, tmpdir):
+def test_time_rotation_dst(monkeypatch_date, darwin_filesystem, tmpdir):
     monkeypatch_date(2018, 10, 27, 5, 0, 0, 0, "CET", 3600)
     i = logger.add(str(tmpdir.join("test_{time}.log")), format="{message}", rotation="1 day")
     logger.debug("First")
@@ -158,6 +249,101 @@ def test_time_rotation_dst(monkeypatch_date, tmpdir):
     assert tmpdir.join("test_2018-10-29_06-00-00_000000.log").read() == "Third\n"
 
 
+@pytest.mark.parametrize("delay", [False, True])
+def test_time_rotation_reopening_native(tmpdir_local, delay):
+    filepath = str(tmpdir_local / "test.log")
+    i = logger.add(filepath, format="{message}", delay=delay, rotation="1 s")
+    logger.info("1")
+    time.sleep(0.75)
+    logger.info("2")
+    logger.remove(i)
+    i = logger.add(filepath, format="{message}", delay=delay, rotation="1 s")
+    logger.info("3")
+
+    assert len(list(tmpdir_local.iterdir())) == 1
+    assert (tmpdir_local / "test.log").read_text() == "1\n2\n3\n"
+
+    time.sleep(0.5)
+    logger.info("4")
+
+    assert len(list(tmpdir_local.iterdir())) == 2
+    assert (tmpdir_local / "test.log").read_text() == "4\n"
+
+    logger.remove(i)
+    time.sleep(0.5)
+    i = logger.add(filepath, format="{message}", delay=delay, rotation="1 s")
+    logger.info("5")
+
+    assert len(list(tmpdir_local.iterdir())) == 2
+    assert (tmpdir_local / "test.log").read_text() == "4\n5\n"
+
+    time.sleep(0.75)
+    logger.info("6")
+    logger.remove(i)
+
+    assert len(list(tmpdir_local.iterdir())) == 3
+    assert (tmpdir_local / "test.log").read_text() == "6\n"
+
+
+def rotation_reopening(tmpdir, monkeypatch_date, delay):
+    monkeypatch_date(2018, 10, 27, 5, 0, 0, 0)
+    filepath = tmpdir.join("test.log")
+    i = logger.add(str(filepath), format="{message}", delay=delay, rotation="2 h")
+    logger.info("1")
+    monkeypatch_date(2018, 10, 27, 6, 30, 0, 0)
+    logger.info("2")
+    logger.remove(i)
+    i = logger.add(str(filepath), format="{message}", delay=delay, rotation="2 h")
+    logger.info("3")
+
+    assert len(tmpdir.listdir()) == 1
+    assert filepath.read() == "1\n2\n3\n"
+
+    monkeypatch_date(2018, 10, 27, 7, 30, 0, 0)
+    logger.info("4")
+
+    assert len(tmpdir.listdir()) == 2
+    assert filepath.read() == "4\n"
+
+    logger.remove(i)
+    monkeypatch_date(2018, 10, 27, 8, 30, 0, 0)
+
+    i = logger.add(str(filepath), format="{message}", delay=delay, rotation="2 h")
+    logger.info("5")
+
+    assert len(tmpdir.listdir()) == 2
+    assert filepath.read() == "4\n5\n"
+
+    monkeypatch_date(2018, 10, 27, 10, 0, 0, 0)
+    logger.info("6")
+    logger.remove(i)
+
+    assert len(tmpdir.listdir()) == 3
+    assert filepath.read() == "6\n"
+
+
+@pytest.mark.parametrize("delay", [False, True])
+def test_time_rotation_reopening_windows(tmpdir, monkeypatch_date, windows_filesystem, delay):
+    rotation_reopening(tmpdir, monkeypatch_date, delay)
+
+
+@pytest.mark.parametrize("delay", [False, True])
+def test_time_rotation_reopening_darwin(tmpdir, monkeypatch_date, darwin_filesystem, delay):
+    rotation_reopening(tmpdir, monkeypatch_date, delay)
+
+
+@pytest.mark.parametrize("delay", [False, True])
+def test_time_rotation_reopening_linux(tmpdir, monkeypatch_date, linux_filesystem, delay):
+    rotation_reopening(tmpdir, monkeypatch_date, delay)
+
+
+@pytest.mark.parametrize("delay", [False, True])
+def test_time_rotation_reopening_linux_no_xattr(
+    tmpdir, monkeypatch_date, linux_no_xattr_filesystem, delay
+):
+    rotation_reopening(tmpdir, monkeypatch_date, delay)
+
+
 def test_function_rotation(monkeypatch_date, tmpdir):
     monkeypatch_date(2018, 1, 1, 0, 0, 0, 0)
     x = iter([False, True, False])