Merge pull request #2 from fleetingbytes/develop

Develop

Author: fleetingbytes

.python-version

@@ -1 +1 @@
-3.14
+3.12

pyproject.toml

@@ -30,6 +30,7 @@ lint = [
 ]
 test = [
     "pytest>=8.4.2",
+    "watchdog>=6.0.0",
 ]
 
 [project.urls]

ruff.toml

@@ -0,0 +1,63 @@
+line-length = 108
+
+[format]
+line-ending = "lf"
+docstring-code-format = true
+
+[lint]
+preview = true
+select = [
+    "B",       # flake8-bugbear
+    "C901",    # mcabe
+    "E",       # pycodestyle errors
+    "ERA",     # eradicate (commented-out-code)
+    "F",       # pyflakes
+    "FLY",     # flynt
+    "G",       # flake8-logging-format
+    "I",       # isort
+    "ICN",     # flake8-import-conventions
+    "INT",     # flake8-gettext
+    "ISC",     # flake8-implicit-str-concat
+    "LOG",     # flake8-logging
+    "N",       # pep8-naming
+    "PERF",    # Perflint
+    "PL",      # pylint
+    "Q",       # quotes
+    "RUF",     # ruff-specific rules
+    "SIM",     # flake8-simplify
+    "TC",      # flake8-type-checking
+    "TID251",  # banned-api
+    "TID253",  # banned-module-level-imports
+    "UP",      # pyupgrade
+    "W",       # pycodestyle warnings
+]
+
+[lint.per-file-ignores]
+"src/**/cli/__init__.py" = [
+    "N813",  # camelcase-imported-as-lowercase
+]
+"tests/test_general.py" = [
+    "N813",  # camelcase-imported-as-lowercase
+]
+"tests/test_log.py" = [
+    "N813",  # camelcase-imported-as-lowercase
+]
+"tests/test_decorators.py" = [
+    "N813",    # camelcase-imported-as-lowercase
+    "PLR917",  # too-many-positional-arguments
+]
+"src/readylog/__init__.py" = [
+    "PLR0913",  # too-many-arguments
+    "PLR0914",  # too-many-local-variables
+    "PLR0917",  # too-many-positional-arguments
+    "PLC2701",  # import-private-name
+]
+
+extend-safe-fixes = [
+    "TC001",  # typing-only-first-party-import
+    "TC002",  # typing-only-third-party-import
+    "TC003",  # typing-only-standard-library-import
+]
+
+[lint.mccabe]
+max-complexity = 5

src/readylog/__init__.py

@@ -1,33 +1,21 @@
-"""
-Copy this file into your project
-to manage your logging configuration there.
-Don't forget to change the name of your app and the names of your loggers here!
-
-Use it like this:
-
-    from logging import getLogger
-    from logging.config import dictConfig as configure_logging
-
-    from readylog import logging_configuration
-
-
-    configure_logging(logging_configuration)
-
-    logger = getLogger(__name__)
-"""
-
+from collections.abc import Callable
+from logging import FileHandler, StreamHandler, _checkLevel, getLevelName
 from pathlib import Path
+from sys import modules
 
-from platformdirs import user_log_dir
 
-MY_APP_NAME = "MY_APP_NAME".replace("-", "_")
+def create_dict_config(
+    logfile: Path,
+    app_name: str,
+    console_log_level: str | int = "WARNING",
+    file_log_level: str | int = "DEBUG",
+    console_handler_factory: Callable = StreamHandler,
+    file_handler_factory: Callable = FileHandler,
+) -> dict[str, str]:
+    console_log_level = _checkLevel(console_log_level)
+    file_log_level = _checkLevel(file_log_level)
+    min_level = getLevelName(min(console_log_level, file_log_level))
 
-logging_dir = Path(user_log_dir(MY_APP_NAME))
-logging_dir.mkdir(parents=True, exist_ok=True)
-logfile = logging_dir / "debug.log"
-
-
-def create_dict_config(logfile: Path) -> dict[str, str]:
     custom_file_formatter_conf = {
         "format": "{message:<50s} {levelname:>9s} {asctime}.{msecs:03.0f} {module}({lineno}) {funcName}",
         "style": "{",
@@ -60,30 +48,30 @@ def create_dict_config(logfile: Path) -> dict[str, str]:
     }
 
     custom_console_handler_conf = {
-        "class": "logging.StreamHandler",
-        "level": "DEBUG",
+        "()": console_handler_factory,
+        "level": console_log_level,
         "formatter": "custom_console_formatter",
         "stream": "ext://sys.stderr",
     }
 
     custom_file_handler_conf = {
-        "class": "logging.FileHandler",
-        "level": "DEBUG",
+        "()": file_handler_factory,
+        "level": file_log_level,
         "formatter": "custom_file_formatter",
         "filename": logfile,
         "mode": "w",
         "encoding": "utf-8",
     }
 
     root_console_handler_conf = {
-        "class": "logging.StreamHandler",
+        "()": console_handler_factory,
         "level": "DEBUG",
         "formatter": "root_console_formatter",
         "stream": "ext://sys.stderr",
     }
 
     root_file_handler_conf = {
-        "class": "logging.FileHandler",
+        "()": file_handler_factory,
         "level": "DEBUG",
         "formatter": "root_file_formatter",
         "filename": logfile.with_stem(f"{logfile.stem}_root"),
@@ -101,7 +89,7 @@ def create_dict_config(logfile: Path) -> dict[str, str]:
     custom_logger_conf = {
         "propagate": False,
         "handlers": ["custom_file_handler", "custom_console_handler"],
-        "level": "DEBUG",
+        "level": min_level,
     }
 
     root_logger_conf = {
@@ -110,8 +98,9 @@ def create_dict_config(logfile: Path) -> dict[str, str]:
     }
 
     loggers_dict = {
-        MY_APP_NAME: custom_logger_conf,
+        app_name: custom_logger_conf,
         "__main__": custom_logger_conf,
+        f"{modules[__name__].__spec__.parent}.decorators": custom_logger_conf,
     }
 
     dict_config = {
@@ -125,6 +114,3 @@ def create_dict_config(logfile: Path) -> dict[str, str]:
     }
 
     return dict_config
-
-
-logging_configuration = create_dict_config(logfile)

src/readylog/decorators.py

@@ -0,0 +1,68 @@
+from functools import wraps
+from itertools import product
+from logging import DEBUG, getLevelName, getLevelNamesMapping, getLogger
+from sys import modules
+
+logger = getLogger(__name__)
+
+
+def log_io(level=DEBUG, enter=False, exit=False):
+    """
+    Decorator factory that logs function input arguments and return values
+    at the specified logging level.
+    Usage:
+        from logging import DEBUG, INFO
+
+        debug = @log_io(DEBUG)
+        info = @log_io(INFO)
+
+
+        @debug
+        def my_func(...):
+            ...
+
+
+        @info
+        def more_important_func(...):
+            ...
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            if enter:
+                logger.log(
+                    level,
+                    "Calling %s with args=%s, kwargs=%s",
+                    func.__name__,
+                    args,
+                    kwargs,
+                    stacklevel=2,
+                )
+            result = func(*args, **kwargs)
+            if exit:
+                logger.log(level, "%s returned %r", func.__name__, result, stacklevel=2)
+            return result
+
+        return wrapper
+
+    return decorator
+
+
+level_names = getLevelNamesMapping().keys()
+
+for level, direction in product(level_names, ("_in", "_out", "")):
+    if direction == "_in":
+        enter = True
+        exit = False
+    elif direction == "_out":
+        enter = False
+        exit = True
+    else:
+        enter = True
+        exit = True
+    setattr(
+        modules[__name__],
+        f"{level.lower()}{direction}",
+        log_io(getLevelName(level), enter=enter, exit=exit),
+    )

src/readylog/mixins.py

@@ -0,0 +1,24 @@
+from copy import copy
+from logging import FileHandler, StreamHandler
+
+
+class MultilineMixin:
+    def emit(self, record):
+        s = record.getMessage()
+        if "\n" not in s:
+            super().emit(record)
+        else:
+            lines = s.splitlines()
+            rec = copy(record)
+            rec.args = None
+            for line in lines:
+                rec.msg = line
+                super().emit(rec)
+
+
+class MultilineStreamHandler(MultilineMixin, StreamHandler):
+    pass
+
+
+class MultilineFileHandler(MultilineMixin, FileHandler):
+    pass

tests/__init__.py

@@ -0,0 +1,56 @@
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+from pytest import FixtureRequest, fixture
+from watchdog.events import FileModifiedEvent, FileSystemEventHandler
+from watchdog.observers import Observer
+
+
+@fixture
+def tmp_log_file(tmp_path) -> Path:
+    path = tmp_path / "log" / "debug.log"
+    path.parent.mkdir(parents=True, exist_ok=False)
+    return path
+
+
+@fixture
+def app_name() -> str:
+    return "test_app"
+
+
+@fixture
+def log_file_observer(tmp_log_file, request: FixtureRequest) -> Observer:
+    expected_number_of_log_lines = request.param
+
+    class LogFileChangeHandler(FileSystemEventHandler):
+        @staticmethod
+        def on_modified(event: FileModifiedEvent):
+            if event.src_path.endswith(tmp_log_file.name):
+                pass
+
+    observer = Observer()
+    handler = LogFileChangeHandler()
+    tmp_log_file.touch()
+    observer.schedule(handler, path=tmp_log_file, recursive=False)
+    observer.start()
+    yield observer
+
+    with open(tmp_log_file, encoding="utf-8") as f:
+        lines = f.readlines()
+    number_of_log_lines = len(lines)
+    assert number_of_log_lines == expected_number_of_log_lines
+
+    observer.stop()
+    observer.join()
+
+
+def function_under_test(*args, **kwargs) -> tuple[tuple[Any, ...], dict[Any, Any]]:
+    return args, kwargs
+
+
+@fixture
+def decorated_function(request: FixtureRequest) -> Callable:
+    decorator = request.param
+    decorated_func = decorator(function_under_test)
+    return decorated_func

tests/conftest.py

@@ -0,0 +1,6 @@
+from logging import getLevelNamesMapping
+
+IGNORED_LEVEL = "NOTSET"
+
+USABLE_LEVELS = {k: v for k, v in getLevelNamesMapping().items() if k != IGNORED_LEVEL}
+USABLE_LEVEL_NAMES = tuple(key for key in USABLE_LEVELS)