add package: pyproject + src (setup_logging, rotation namer/rotator, formats)

Signed-off-by: disqualifier <dev@disqualifier.me>

pyproject.toml

@@ -0,0 +1,13 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "log_setup"
+version = "0.1.0"
+description = "stdlib app-entry-point logging setup: live run.log, rotation, gzip, retention, consistent format"
+requires-python = ">=3.10"
+dependencies = []
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/log_setup"]

src/log_setup/__init__.py

@@ -0,0 +1,22 @@
+"""log_setup — app-entry-point logging configuration (sync, stdlib only).
+
+call once at an application's entry point to configure the whole process: a live
+run.log, rotation (daily/size/on_start), gzip of rolled files, retention, console
+output, and a consistent `time | module | level | message` format.
+
+    from log_setup import setup_logging
+
+    setup_logging(name="run", level="INFO")        # daily rotation, logs/ dir, gzip
+    log = logging.getLogger(__name__)
+    log.info("up")                                  # -> run.log + console
+
+reusable libraries do NOT call this — they only `logging.getLogger(__name__)` and
+emit; the application owns this setup. shipping logs to a backend is out of scope
+(that's Promtail's job against the produced files).
+"""
+
+from .setup import setup_logging
+
+__all__ = ["setup_logging"]
+
+__version__ = "0.1.0"

src/log_setup/formats.py

@@ -0,0 +1,15 @@
+"""default log format + datefmt for the app-wide setup.
+
+one format for v0.1.0, used on both console and file. `%(name)s` is the getLogger
+name the emitting module used, so each library/module shows in the line.
+"""
+
+import logging
+
+DEFAULT_FORMAT = "%(asctime)s | %(name)s | %(levelname)s | %(message)s"
+DEFAULT_DATEFMT = "%Y-%m-%d %H:%M:%S"
+
+
+def build_formatter(fmt=None, datefmt=None) -> logging.Formatter:
+    """build a logging.Formatter from overrides, falling back to the defaults"""
+    return logging.Formatter(fmt or DEFAULT_FORMAT, datefmt or DEFAULT_DATEFMT)

src/log_setup/rotation.py

@@ -0,0 +1,102 @@
+"""custom namer/rotator + on-start rotation + retention pruning (stdlib only).
+
+the stdlib rotating handlers roll a file next to the live file; these helpers
+override the namer/rotator so rolled files land in `log_dir` and are gzipped when
+asked, keep the live file at its stable path, and handle the on-start and prune
+paths the handlers don't manage themselves.
+"""
+
+import gzip
+import os
+import shutil
+import time
+from typing import Callable, Tuple
+
+
+def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]:
+    """namer: redirect a rolled filename into log_dir, adding .gz when compressing
+
+    the handler hands us the default rolled path (next to the live file); we keep its
+    basename but place it under log_dir, and append .gz so the gzipped name matches.
+    """
+    def namer(default_name: str) -> str:
+        base = os.path.basename(default_name)
+        target = os.path.join(log_dir, base)
+        return target + ".gz" if compress else target
+    return namer
+
+
+def make_rotator(compress: bool) -> Callable[[str, str], None]:
+    """rotator: move (or gzip) the source live file to the destination rolled path"""
+    def rotator(source: str, dest: str) -> None:
+        if not os.path.exists(source):
+            return
+        if compress:
+            with open(source, "rb") as src, gzip.open(dest, "wb") as dst:
+                shutil.copyfileobj(src, dst)
+            os.remove(source)
+        else:
+            os.replace(source, dest)
+    return rotator
+
+
+def rotate_on_start(live_path: str, log_dir: str, compress: bool, clock=time.localtime) -> None:
+    """move an existing live file into log_dir with a timestamp, gzipped if asked
+
+    no-op if the live file doesn't exist. used by rotate="on_start" before the fresh
+    handler opens a new live file. the timestamp form is run.<%Y-%m-%d_%H-%M-%S>.log.
+    """
+    if not os.path.exists(live_path):
+        return
+    stem = os.path.splitext(os.path.basename(live_path))[0]
+    stamp = time.strftime("%Y-%m-%d_%H-%M-%S", clock())
+    dest = os.path.join(log_dir, f"{stem}.{stamp}.log")
+    if compress:
+        dest += ".gz"
+        with open(live_path, "rb") as src, gzip.open(dest, "wb") as dst:
+            shutil.copyfileobj(src, dst)
+        os.remove(live_path)
+    else:
+        os.replace(live_path, dest)
+
+
+def prune(log_dir: str, stem: str, backup_count: int) -> None:
+    """keep only the newest `backup_count` rolled files for a given stem in log_dir
+
+    matches files beginning with `<stem>.` (e.g. run.*), sorted by mtime, deleting the
+    oldest beyond the count. used for on_start, which the handlers don't auto-prune.
+    """
+    if backup_count <= 0:
+        return
+    try:
+        entries = [
+            os.path.join(log_dir, name)
+            for name in os.listdir(log_dir)
+            if name.startswith(f"{stem}.") and name != f"{stem}.log"
+        ]
+    except OSError:
+        return
+    files = [(p, _safe_mtime(p)) for p in entries if os.path.isfile(p)]
+    files.sort(key=lambda pair: pair[1], reverse=True)
+    for path, _ in files[backup_count:]:
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+
+
+def _safe_mtime(path: str) -> float:
+    """mtime of a path, or 0 if it can't be stat'd"""
+    try:
+        return os.path.getmtime(path)
+    except OSError:
+        return 0.0
+
+
+def attach_rolling(handler, log_dir: str, compress: bool) -> Tuple[Callable, Callable]:
+    """wire the custom namer + rotator onto a rotating handler; return them"""
+    namer = make_namer(log_dir, compress)
+    rotator = make_rotator(compress)
+    handler.namer = namer
+    handler.rotator = rotator
+    return namer, rotator

src/log_setup/setup.py

@@ -0,0 +1,158 @@
+"""app-entry-point logging setup (sync, stdlib only).
+
+`setup_logging` configures the root logger once for the whole process: a live
+run.log at a stable path, rotation (daily/size/on_start/none) into a logs/ dir, gzip
+of rolled files, retention, console output, and a consistent format. it is called by
+the APPLICATION, not by reusable libraries (those stay emit-only). it is idempotent
+(no duplicate handlers on repeat calls), never crashes the app over logging, and can
+route through a background queue so an async event loop doesn't block on file I/O.
+"""
+
+import atexit
+import logging
+import logging.handlers
+import os
+import queue
+from typing import Optional, Union
+
+from .formats import build_formatter
+from .rotation import attach_rolling, prune, rotate_on_start
+
+log = logging.getLogger(__name__)
+
+_MARKER = "_log_setup_owned"
+_listener = None
+
+
+def _level_value(level: Union[int, str]) -> int:
+    """coerce a level name or int to a logging level int (defaults to INFO)"""
+    if isinstance(level, int):
+        return level
+    return logging.getLevelName(str(level).upper()) if isinstance(level, str) else logging.INFO
+
+
+def _clear_owned(root: logging.Logger) -> None:
+    """remove only the handlers this lib previously added; leave app handlers alone"""
+    global _listener
+    if _listener is not None:
+        _listener.stop()
+        _listener = None
+    for handler in list(root.handlers):
+        if getattr(handler, _MARKER, False):
+            root.removeHandler(handler)
+            try:
+                handler.close()
+            except Exception:
+                pass
+
+
+def _tag(handler: logging.Handler) -> logging.Handler:
+    """mark a handler as owned by this lib so idempotent re-setup can clear it"""
+    setattr(handler, _MARKER, True)
+    return handler
+
+
+def _file_handler(
+    name: str, live_path: str, log_dir: str, rotate: Optional[str],
+    backup_count: int, max_bytes: int, compress: bool,
+) -> logging.Handler:
+    """build the configured file handler with custom rolling into log_dir"""
+    if rotate == "size":
+        handler = logging.handlers.RotatingFileHandler(
+            live_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8",
+        )
+        attach_rolling(handler, log_dir, compress)
+    elif rotate == "daily":
+        handler = logging.handlers.TimedRotatingFileHandler(
+            live_path, when="midnight", backupCount=backup_count, encoding="utf-8",
+        )
+        attach_rolling(handler, log_dir, compress)
+    else:
+        if rotate == "on_start":
+            rotate_on_start(live_path, log_dir, compress)
+            prune(log_dir, name, backup_count)
+        handler = logging.FileHandler(live_path, encoding="utf-8")
+    return handler
+
+
+def setup_logging(
+    name: str = "run",
+    log_dir: str = "logs",
+    level: Union[int, str] = "INFO",
+    rotate: Optional[str] = "daily",
+    backup_count: int = 14,
+    max_bytes: int = 10_000_000,
+    compress: bool = True,
+    console: bool = False,
+    queue: bool = False,
+    fmt: Optional[str] = None,
+    datefmt: Optional[str] = None,
+) -> logging.Logger:
+    """configure the root logger for the whole process and return it
+
+    `name` -> <name>.log live file at cwd; rolled/compressed copies go to `log_dir`.
+    `rotate` is "daily" (default), "size", "on_start", or None. `console=True` adds a
+    stdout handler (off by default — the file is the output). `queue=True` routes records
+    through a background QueueListener so file I/O never blocks the caller (the listener
+    is stopped at exit). idempotent: a repeat call clears only the handlers this function
+    added. never raises over logging — an unwritable `log_dir` falls back to console-only
+    with a warning even when `console` is off, so output is never silently lost.
+    """
+    global _listener
+
+    root = logging.getLogger()
+    root.setLevel(_level_value(level))
+    _clear_owned(root)
+
+    formatter = build_formatter(fmt, datefmt)
+    live_path = f"{name}.log"
+
+    handlers = []
+
+    file_ok = True
+    try:
+        os.makedirs(log_dir, exist_ok=True)
+    except OSError:
+        file_ok = False
+
+    if file_ok:
+        try:
+            fh = _file_handler(name, live_path, log_dir, rotate, backup_count, max_bytes, compress)
+            fh.setFormatter(formatter)
+            handlers.append(fh)
+        except OSError:
+            file_ok = False
+
+    if console or not file_ok:
+        sh = logging.StreamHandler()
+        sh.setFormatter(formatter)
+        handlers.append(sh)
+
+    if queue:
+        record_queue: "queue.Queue" = _make_queue()
+        qh = _tag(logging.handlers.QueueHandler(record_queue))
+        root.addHandler(qh)
+        _listener = logging.handlers.QueueListener(record_queue, *handlers, respect_handler_level=True)
+        _listener.start()
+        atexit.register(_stop_listener)
+    else:
+        for handler in handlers:
+            root.addHandler(_tag(handler))
+
+    if not file_ok:
+        log.warning("log_setup: log_dir %r not writable; logging to console only", log_dir)
+
+    return root
+
+
+def _make_queue() -> "queue.Queue":
+    """unbounded in-memory queue for the QueueHandler -> QueueListener path"""
+    return queue.Queue(-1)
+
+
+def _stop_listener() -> None:
+    """stop the background queue listener, flushing remaining records"""
+    global _listener
+    if _listener is not None:
+        _listener.stop()
+        _listener = None