Add make_formatter

requirements.txt

@@ -266,7 +266,7 @@ polars-lts-cpu==1.6.0
     # via dycw-utilities (pyproject.toml)
 pqdm==0.2.0
     # via dycw-utilities (pyproject.toml)
-protobuf==5.28.1
+protobuf==5.28.2
     # via
     #   streamlit
     #   vegafusion
@@ -427,7 +427,7 @@ urllib3==2.2.3
     # via requests
 userpath==1.9.2
     # via hatch
-uv==0.4.11
+uv==0.4.12
     # via hatch
 vegafusion==1.6.9
     # via dycw-utilities (pyproject.toml)

requirements/streamlit.txt

@@ -63,7 +63,7 @@ pillow==10.4.0
     # via streamlit
 pluggy==1.5.0
     # via pytest
-protobuf==5.28.1
+protobuf==5.28.2
     # via streamlit
 pyarrow==17.0.0
     # via streamlit

src/tests/test_loguru.py

@@ -9,7 +9,7 @@
 from loguru import logger
 from loguru._defaults import LOGURU_FORMAT
 from loguru._recattrs import RecordFile, RecordLevel, RecordProcess, RecordThread
-from pytest import CaptureFixture, mark, param, raises
+from pytest import CaptureFixture, fixture, mark, param, raises
 
 from tests.test_loguru_functions import (
     func_test_log_contextualize,
@@ -36,6 +36,7 @@
     logged_sleep_sync,
     make_except_hook,
     make_filter,
+    make_formatter,
     make_slack_sink,
     make_slack_sink_async,
 )
@@ -44,13 +45,45 @@
 if TYPE_CHECKING:
     from collections.abc import Callable
 
-    from loguru import Record
+    from loguru import Record, RecordException
     from pytest import CaptureFixture
 
     from utilities.iterables import MaybeIterable
     from utilities.types import Duration
 
 
+@fixture
+def record() -> Record:
+    record = {
+        "elapsed": dt.timedelta(seconds=11, microseconds=635587),
+        "exception": None,
+        "extra": {"x": 1, "y": 2},
+        "file": RecordFile(
+            name="1723464958.py",
+            path="/var/folders/z2/t3tvc2yn33j0zdd910j7805r0000gn/T/ipykernel_98745/1723464958.py",
+        ),
+        "function": "<module>",
+        "level": RecordLevel(name="INFO", no=20, icon="ℹ️ "),  # noqa: RUF001
+        "line": 1,
+        "message": "l2",
+        "module": "1723464958",
+        "name": "__main__",
+        "process": RecordProcess(id_=98745, name="MainProcess"),
+        "thread": RecordThread(id_=8420429632, name="MainThread"),
+        "time": dt.datetime(
+            2024,
+            8,
+            31,
+            14,
+            3,
+            52,
+            388537,
+            tzinfo=dt.timezone(dt.timedelta(seconds=32400), "JST"),
+        ),
+    }
+    return cast(Any, record)
+
+
 class TestGetLoggingLevelNameAndNumber:
     @mark.parametrize(
         ("name", "number"),
@@ -275,9 +308,9 @@ def test_main(self) -> None:
 
 
 class TestMakeFilter:
-    def test_main(self) -> None:
+    def test_main(self, *, record: Record) -> None:
         filter_func = make_filter(final_filter=True)
-        assert filter_func(self._record)
+        assert filter_func(record)
 
     @mark.parametrize(
         ("level", "expected"),
@@ -291,9 +324,9 @@ def test_main(self) -> None:
             param(LogLevel.CRITICAL, False),
         ],
     )
-    def test_level(self, *, level: LogLevel, expected: bool) -> None:
+    def test_level(self, *, level: LogLevel, record: Record, expected: bool) -> None:
         filter_func = make_filter(level=level, final_filter=True)
-        result = filter_func(self._record)
+        result = filter_func(record)
         assert result is expected
 
     @mark.parametrize(
@@ -308,9 +341,11 @@ def test_level(self, *, level: LogLevel, expected: bool) -> None:
             param(LogLevel.CRITICAL, False),
         ],
     )
-    def test_min_level(self, *, level: LogLevel, expected: bool) -> None:
+    def test_min_level(
+        self, *, level: LogLevel, record: Record, expected: bool
+    ) -> None:
         filter_func = make_filter(min_level=level, final_filter=True)
-        result = filter_func(self._record)
+        result = filter_func(record)
         assert result is expected
 
     @mark.parametrize(
@@ -325,9 +360,11 @@ def test_min_level(self, *, level: LogLevel, expected: bool) -> None:
             param(LogLevel.CRITICAL, True),
         ],
     )
-    def test_max_level(self, *, level: LogLevel, expected: bool) -> None:
+    def test_max_level(
+        self, *, level: LogLevel, record: Record, expected: bool
+    ) -> None:
         filter_func = make_filter(max_level=level, final_filter=True)
-        result = filter_func(self._record)
+        result = filter_func(record)
         assert result is expected
 
     @mark.parametrize(
@@ -345,12 +382,13 @@ def test_name_exists(
         *,
         name_include: MaybeIterable[str] | None,
         name_exclude: MaybeIterable[str] | None,
+        record: Record,
         expected: bool,
     ) -> None:
         filter_func = make_filter(
             name_include=name_include, name_exclude=name_exclude, final_filter=True
         )
-        result = filter_func(self._record)
+        result = filter_func(record)
         assert result is expected
 
     @mark.parametrize(
@@ -368,12 +406,13 @@ def test_name_does_not_exist(
         *,
         name_include: MaybeIterable[str] | None,
         name_exclude: MaybeIterable[str] | None,
+        record: Record,
     ) -> None:
         filter_func = make_filter(
             name_include=name_include, name_exclude=name_exclude, final_filter=True
         )
-        record: Record = cast(Any, self._record | {"name": None})
-        assert filter_func(record)
+        record2: Record = cast(Any, record | {"name": None})
+        assert filter_func(record2)
 
     @mark.parametrize(
         ("extra_include_all", "extra_exclude_any", "expected"),
@@ -400,14 +439,15 @@ def test_extra_inc_all_exc_any(
         *,
         extra_include_all: MaybeIterable[str] | None,
         extra_exclude_any: MaybeIterable[str] | None,
+        record: Record,
         expected: bool,
     ) -> None:
         filter_func = make_filter(
             extra_include_all=extra_include_all,
             extra_exclude_any=extra_exclude_any,
             final_filter=True,
         )
-        result = filter_func(self._record)
+        result = filter_func(record)
         assert result is expected
 
     @mark.parametrize(
@@ -435,14 +475,15 @@ def test_extra_inc_any_exc_all(
         *,
         extra_include_any: MaybeIterable[str] | None,
         extra_exclude_all: MaybeIterable[str] | None,
+        record: Record,
         expected: bool,
     ) -> None:
         filter_func = make_filter(
             extra_include_any=extra_include_any,
             extra_exclude_all=extra_exclude_all,
             final_filter=True,
         )
-        result = filter_func(self._record)
+        result = filter_func(record)
         assert result is expected
 
     @mark.parametrize(
@@ -465,42 +506,63 @@ def test_final_filter(
         *,
         name: str,
         final_filter: bool | Callable[[], bool] | None,
+        record: Record,
         expected: bool,
     ) -> None:
         filter_func = make_filter(name_include=name, final_filter=final_filter)
-        result = filter_func(self._record)
+        result = filter_func(record)
         assert result is expected
 
-    @property
-    def _record(self) -> Record:
-        record = {
-            "elapsed": dt.timedelta(seconds=11, microseconds=635587),
-            "exception": None,
-            "extra": {"x": 1, "y": 2},
-            "file": RecordFile(
-                name="1723464958.py",
-                path="/var/folders/z2/t3tvc2yn33j0zdd910j7805r0000gn/T/ipykernel_98745/1723464958.py",
-            ),
-            "function": "<module>",
-            "level": RecordLevel(name="INFO", no=20, icon="ℹ️ "),  # noqa: RUF001
-            "line": 1,
-            "message": "l2",
-            "module": "1723464958",
-            "name": "__main__",
-            "process": RecordProcess(id_=98745, name="MainProcess"),
-            "thread": RecordThread(id_=8420429632, name="MainThread"),
-            "time": dt.datetime(
-                2024,
-                8,
-                31,
-                14,
-                3,
-                52,
-                388537,
-                tzinfo=dt.timezone(dt.timedelta(seconds=32400), "JST"),
-            ),
-        }
-        return cast(Any, record)
+
+class TestMakeFormatter:
+    def test_main(self, *, record: Record) -> None:
+        format_ = make_formatter(console_or_file="console")
+        result = format_(record)
+        expected = "{time:YYYY-MM-DD} <level>{time:HH:mm:ss}</level>.{time:SSS}  <level>{function}</level>: <level>{message}</level>  {extra}  ({name}:{line})\n"
+        assert result == expected
+
+    def test_file(self, *, record: Record) -> None:
+        format_ = make_formatter(console_or_file="file")
+        result = format_(record)
+        expected = "{time:YYYY-MM-DD (ddd)} <level>{time:HH:mm:ss}</level>.{time:SSS zz}  <level>{function}</level>: <level>{message}</level>  {extra}  ({name}:{line})\n"
+        assert result == expected
+
+    def test_prefix(self, *, record: Record) -> None:
+        format_ = make_formatter(console_or_file="console", prefix=">")
+        result = format_(record)
+        expected = ">{time:YYYY-MM-DD} <level>{time:HH:mm:ss}</level>.{time:SSS}  <level>{function}</level>: <level>{message}</level>  {extra}  ({name}:{line})\n"
+        assert result == expected
+
+    def test_no_message(self, *, record: Record) -> None:
+        record2: Record = {**record, "message": ""}
+        format_ = make_formatter(console_or_file="console")
+        result = format_(record2)
+        expected = "{time:YYYY-MM-DD} <level>{time:HH:mm:ss}</level>.{time:SSS}  <level>{function}</level>  {extra}  ({name}:{line})\n"
+        assert result == expected
+
+    def test_no_extra(self, *, record: Record) -> None:
+        record2: Record = cast(Any, {k: v for k, v in record.items() if k != "extra"})
+        format_ = make_formatter(console_or_file="console")
+        result = format_(record2)
+        expected = "{time:YYYY-MM-DD} <level>{time:HH:mm:ss}</level>.{time:SSS}  <level>{function}</level>: <level>{message}</level>  ({name}:{line})\n"
+        assert result == expected
+
+    def test_extra_but_only_private(self, *, record: Record) -> None:
+        record2: Record = {**record, "extra": {"_key": "value"}}
+        format_ = make_formatter(console_or_file="console")
+        result = format_(record2)
+        expected = "{time:YYYY-MM-DD} <level>{time:HH:mm:ss}</level>.{time:SSS}  <level>{function}</level>: <level>{message}</level>  ({name}:{line})\n"
+        assert result == expected
+
+    def test_exception(self, *, record: Record) -> None:
+        exception: RecordException = cast(
+            Any, {"type": None, "value": None, "traceback": None}
+        )
+        record2: Record = {**record, "exception": exception}
+        format_ = make_formatter(console_or_file="console")
+        result = format_(record2)
+        expected = "{time:YYYY-MM-DD} <level>{time:HH:mm:ss}</level>.{time:SSS}  <level>{function}</level>: <level>{message}</level>  {extra}  ({name}:{line})\n{exception}\n"
+        assert result == expected
 
 
 class TestMakeSlackSink:

src/utilities/__init__.py

@@ -1,3 +1,3 @@
 from __future__ import annotations
 
-__version__ = "0.57.4"
+__version__ = "0.57.5"

src/utilities/loguru.py

@@ -11,7 +11,7 @@
 from enum import StrEnum, unique
 from logging import Handler, LogRecord
 from sys import __excepthook__, _getframe, stderr
-from typing import TYPE_CHECKING, Any, TextIO, TypedDict, assert_never, cast
+from typing import TYPE_CHECKING, Any, Literal, TextIO, TypedDict, assert_never, cast
 
 from loguru import logger
 from typing_extensions import override
@@ -391,6 +391,52 @@ def filter_func(record: Record, /) -> bool:
     return filter_func
 
 
+def make_formatter(
+    *,
+    console_or_file: Literal["console", "file"],
+    prefix: str | None = None,
+    exception: bool = True,
+) -> FormatFunction:
+    """Make a formatter."""
+
+    def format_record(record: Record, /) -> str:
+        """Format a record."""
+        time_part = "<level>{time:HH:mm:ss}</level>"
+        match console_or_file:
+            case "console":
+                datetime_part = f"{{time:YYYY-MM-DD}} {time_part}.{{time:SSS}}"
+            case "file":
+                datetime_part = f"{{time:YYYY-MM-DD (ddd)}} {time_part}.{{time:SSS zz}}"
+            case _ as never:  # pyright: ignore[reportUnnecessaryComparison]
+                assert_never(never)
+        parts1 = [datetime_part]
+        if record["message"]:
+            parts1.append("<level>{function}</level>: <level>{message}</level>")
+        else:
+            parts1.append("<level>{function}</level>")
+        try:
+            extra = record["extra"]
+        except KeyError:
+            pass
+        else:
+            extra_non_underscore = {
+                k: v
+                for k, v in extra.items()
+                if not (isinstance(k, str) and k.startswith("_"))
+            }
+            if len(extra_non_underscore) >= 1:
+                parts1.append("{extra}")
+        parts1.append("({name}:{line})")
+        parts2 = ["  ".join(parts1), "\n"]
+        if prefix is not None:
+            parts2.insert(0, prefix)
+        if (record["exception"] is not None) and exception:
+            parts2.extend(["{exception}", "\n"])
+        return "".join(parts2)
+
+    return format_record
+
+
 def make_slack_sink(url: str, /) -> Callable[[Message], None]:
     """Make a `slack` sink."""
     from utilities.slack_sdk import SendSlackError, send_slack_sync
@@ -431,6 +477,7 @@ async def sink_async(message: Message, /) -> None:
     "logged_sleep_sync",
     "make_except_hook",
     "make_filter",
+    "make_formatter",
     "make_slack_sink",
     "make_slack_sink_async",
 ]