Refactor record matching to use internal Matcher class (#126)

This will make adding new features such as #25, #26 and #80 easier.

It also opens the door to #31 and #124, if desired in future.

logot/_capture.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from logot._format import format_log
-
 
 class Captured:
     """
@@ -55,6 +53,3 @@ def __eq__(self, other: object) -> bool:
 
     def __repr__(self) -> str:
         return f"Captured({self.levelname!r}, {self.msg!r}, levelno={self.levelno!r})"
-
-    def __str__(self) -> str:
-        return format_log(self.levelname, self.msg)

logot/_level.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from logot._capture import Captured
+from logot._match import Matcher
+from logot._typing import Level
+
+
+class _LevelNameMatcher(Matcher):
+    __slots__ = ("_levelname",)
+
+    def __init__(self, levelname: str) -> None:
+        self._levelname = levelname
+
+    def match(self, captured: Captured) -> bool:
+        return captured.levelname == self._levelname
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, _LevelNameMatcher) and other._levelname == self._levelname
+
+    def __repr__(self) -> str:
+        return repr(self._levelname)
+
+    def __str__(self) -> str:
+        return f"[{self._levelname}]"
+
+
+DEBUG_MATCHER: Matcher = _LevelNameMatcher("DEBUG")
+INFO_MATCHER: Matcher = _LevelNameMatcher("INFO")
+WARNING_MATCHER: Matcher = _LevelNameMatcher("WARNING")
+ERROR_MATCHER: Matcher = _LevelNameMatcher("ERROR")
+CRITICAL_MATCHER: Matcher = _LevelNameMatcher("CRITICAL")
+
+
+class _LevelNoMatcher(Matcher):
+    __slots__ = ("_levelno",)
+
+    def __init__(self, levelno: int) -> None:
+        self._levelno = levelno
+
+    def match(self, captured: Captured) -> bool:
+        return captured.levelno == self._levelno
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, _LevelNoMatcher) and other._levelno == self._levelno
+
+    def __repr__(self) -> str:
+        return repr(self._levelno)
+
+    def __str__(self) -> str:
+        return f"[Level {self._levelno}]"
+
+
+def level_matcher(level: Level) -> Matcher:
+    # Handle `str` level.
+    if isinstance(level, str):
+        return _LevelNameMatcher(level)
+    # Handle `int` level.
+    if isinstance(level, int):
+        return _LevelNoMatcher(level)
+    # Handle invalid level.
+    raise TypeError(f"Invalid level: {level!r}")

logot/_logged.py

@@ -3,10 +3,10 @@
 from abc import ABC, abstractmethod
 
 from logot._capture import Captured
-from logot._format import format_level, format_log
-from logot._msg import compile_msg_matcher
+from logot._level import CRITICAL_MATCHER, DEBUG_MATCHER, ERROR_MATCHER, INFO_MATCHER, WARNING_MATCHER, level_matcher
+from logot._match import Matcher
+from logot._msg import msg_matcher
 from logot._typing import Level
-from logot._validate import validate_level
 
 
 class Logged(ABC):
@@ -78,7 +78,7 @@ def log(level: Level, msg: str) -> Logged:
     :param level: A log level name (e.g. ``"DEBUG"``) or numeric level (e.g. :data:`logging.DEBUG`).
     :param msg: A log :doc:`message pattern </log-message-matching>`.
     """
-    return _RecordLogged(validate_level(level), msg)
+    return _RecordMatcher(level_matcher(level), msg_matcher(msg))
 
 
 def debug(msg: str) -> Logged:
@@ -88,7 +88,7 @@ def debug(msg: str) -> Logged:
 
     :param msg: A log :doc:`message pattern </log-message-matching>`.
     """
-    return _RecordLogged("DEBUG", msg)
+    return _RecordMatcher(DEBUG_MATCHER, msg_matcher(msg))
 
 
 def info(msg: str) -> Logged:
@@ -98,7 +98,7 @@ def info(msg: str) -> Logged:
 
     :param msg: A log :doc:`message pattern </log-message-matching>`.
     """
-    return _RecordLogged("INFO", msg)
+    return _RecordMatcher(INFO_MATCHER, msg_matcher(msg))
 
 
 def warning(msg: str) -> Logged:
@@ -108,7 +108,7 @@ def warning(msg: str) -> Logged:
 
     :param msg: A log :doc:`message pattern </log-message-matching>`.
     """
-    return _RecordLogged("WARNING", msg)
+    return _RecordMatcher(WARNING_MATCHER, msg_matcher(msg))
 
 
 def error(msg: str) -> Logged:
@@ -118,7 +118,7 @@ def error(msg: str) -> Logged:
 
     :param msg: A log :doc:`message pattern </log-message-matching>`.
     """
-    return _RecordLogged("ERROR", msg)
+    return _RecordMatcher(ERROR_MATCHER, msg_matcher(msg))
 
 
 def critical(msg: str) -> Logged:
@@ -128,42 +128,31 @@ def critical(msg: str) -> Logged:
 
     :param msg: A log :doc:`message pattern </log-message-matching>`.
     """
-    return _RecordLogged("CRITICAL", msg)
+    return _RecordMatcher(CRITICAL_MATCHER, msg_matcher(msg))
 
 
-class _RecordLogged(Logged):
-    __slots__ = ("_level", "_msg", "_msg_matcher")
+class _RecordMatcher(Logged):
+    __slots__ = ("_matchers",)
 
-    def __init__(self, level: Level, msg: str) -> None:
-        self._level = level
-        self._msg = msg
-        self._msg_matcher = compile_msg_matcher(msg)
+    def __init__(self, *matchers: Matcher) -> None:
+        self._matchers = matchers
 
     def __eq__(self, other: object) -> bool:
-        return isinstance(other, _RecordLogged) and other._level == self._level and other._msg == self._msg
+        return isinstance(other, _RecordMatcher) and other._matchers == self._matchers
 
     def __repr__(self) -> str:
-        return f"log({self._level!r}, {self._msg!r})"
+        matchers_repr = ", ".join(map(repr, self._matchers))
+        return f"log({matchers_repr})"
 
     def reduce(self, captured: Captured) -> Logged | None:
-        # Match `str` level.
-        if isinstance(self._level, str):
-            if self._level != captured.levelname:
-                return self
-        # Match `int` level.
-        elif isinstance(self._level, int):
-            if self._level != captured.levelno:
-                return self
-        else:  # pragma: no cover
-            raise TypeError(f"Invalid level: {self._level!r}")
-        # Match message.
-        if not self._msg_matcher(captured.msg):
-            return self
-        # We matched!
-        return None
+        # Handle full reduction.
+        if all(matcher.match(captured) for matcher in self._matchers):
+            return None
+        # Handle no reduction.
+        return self
 
     def _str(self, *, indent: str) -> str:
-        return format_log(format_level(self._level), self._msg)
+        return " ".join(map(str, self._matchers))
 
 
 class _ComposedLogged(Logged):

logot/_match.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from logot._capture import Captured
+
+
+class Matcher(ABC):
+    __slots__ = ()
+
+    @abstractmethod
+    def match(self, captured: Captured) -> bool:
+        raise NotImplementedError
+
+    @abstractmethod
+    def __eq__(self, other: object) -> bool:
+        raise NotImplementedError
+
+    @abstractmethod
+    def __repr__(self) -> str:
+        raise NotImplementedError
+
+    @abstractmethod
+    def __str__(self) -> str:
+        raise NotImplementedError

logot/_msg.py

@@ -1,13 +1,9 @@
 from __future__ import annotations
 
 import re
-from typing import Callable
 
-from logot._typing import TypeAlias
-
-# Compiled matcher callable.
-# The returned `object` is truthy on successful match and falsy on failed match.
-MessageMatcher: TypeAlias = Callable[[str], object]
+from logot._capture import Captured
+from logot._match import Matcher
 
 # Regex matching a simplified conversion specifier.
 _RE_CONVERSION = re.compile(r"%(.|$)")
@@ -41,8 +37,38 @@
 }
 
 
-def compile_msg_matcher(pattern: str) -> MessageMatcher:
-    parts: list[str] = _RE_CONVERSION.split(pattern)
+class _MessageMatcher(Matcher):
+    __slots__ = ("_msg",)
+
+    def __init__(self, msg: str) -> None:
+        self._msg = msg
+
+    def match(self, captured: Captured) -> bool:
+        return captured.msg == self._msg
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, _MessageMatcher) and other._msg == self._msg
+
+    def __repr__(self) -> str:
+        return repr(self._msg)
+
+    def __str__(self) -> str:
+        return self._msg
+
+
+class _MessagePatternMatcher(_MessageMatcher):
+    __slots__ = ("_pattern",)
+
+    def __init__(self, msg: str, pattern: re.Pattern[str]) -> None:
+        super().__init__(msg)
+        self._pattern = pattern
+
+    def match(self, captured: Captured) -> bool:
+        return self._pattern.fullmatch(captured.msg) is not None
+
+
+def msg_matcher(msg: str) -> Matcher:
+    parts: list[str] = _RE_CONVERSION.split(msg)
     parts_len = len(parts)
     # If there is more than one part, at least one conversion specifier was found and we might need a regex matcher.
     if parts_len > 1:
@@ -60,8 +86,9 @@ def compile_msg_matcher(pattern: str) -> MessageMatcher:
         # Create regex matcher.
         if is_regex:
             parts[::2] = map(re.escape, parts[::2])
-            return re.compile("".join(parts), re.DOTALL).fullmatch
+            pattern = re.compile("".join(parts), re.DOTALL)
+            return _MessagePatternMatcher(msg, pattern)
         # Recreate the pattern with all escape sequences replaced.
-        pattern = "".join(parts)
+        msg = "".join(parts)
     # Create simple matcher.
-    return pattern.__eq__
+    return _MessageMatcher(msg)

tests/test_captured.py

@@ -18,7 +18,3 @@ def test_eq_fail() -> None:
 
 def test_repr() -> None:
     assert repr(Captured("INFO", "foo bar")) == "Captured('INFO', 'foo bar', levelno=None)"
-
-
-def test_str() -> None:
-    assert str(Captured("INFO", "foo bar")) == "[INFO] foo bar"

tests/test_level.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from typing import cast
+
+import pytest
+
+from logot._level import level_matcher
+from logot._typing import Level
+
+
+def test_eq_pass() -> None:
+    assert level_matcher("INFO") == level_matcher("INFO")
+    assert level_matcher(20) == level_matcher(20)
+
+
+def test_eq_fail() -> None:
+    assert level_matcher("INFO") != level_matcher("WARNING")
+    assert level_matcher(20) != level_matcher(30)
+    assert level_matcher("INFO") != level_matcher(20)
+
+
+def test_repr() -> None:
+    assert repr(level_matcher("INFO")) == "'INFO'"
+    assert repr(level_matcher(20)) == "20"
+
+
+def test_str() -> None:
+    assert str(level_matcher("INFO")) == "[INFO]"
+    assert str(level_matcher(20)) == "[Level 20]"
+
+
+def test_invalid() -> None:
+    with pytest.raises(TypeError) as ex:
+        level_matcher(cast(Level, 1.5))
+    assert str(ex.value) == "Invalid level: 1.5"