rethink-public/log_setup
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: rethink-public:main
Branches Tags
rethink-public:main
rethink-public:v0.4.1
rethink-public:v0.4.0
rethink-public:v0.3.2
rethink-public:v0.3.1
rethink-public:v0.3.0
rethink-public:v0.2.0
rethink-public:v0.1.1
rethink-public:v0.1.0
..
compare: rethink-public:v0.3.1
Branches Tags
rethink-public:main
rethink-public:v0.4.1
rethink-public:v0.4.0
rethink-public:v0.3.2
rethink-public:v0.3.1
rethink-public:v0.3.0
rethink-public:v0.2.0
rethink-public:v0.1.1
rethink-public:v0.1.0

No commits in common. "main" and "v0.3.1" have entirely different histories.
main ... v0.3.1

7 changed files with 40 additions and 275 deletions
Show Stats Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1,5 +1,5 @@
# claude
.claude/
CLAUDE.md
# python
__pycache__/

51
README.md
View File

@ -13,13 +13,11 @@ and emit; their records flow into the handlers `log_setup` wired.
## Install
```
log_setup @ git+ssh://git@git.rethinkstudios.io/rethink-public/log_setup.git@v0.4.1
log_setup @ git+ssh://git@git.rethinkstudios.io/rethink-public/log_setup.git@v0.3.1
```
No dependencies — stdlib only.
Drop the `@v0.4.1` suffix from the line above to install the latest unpinned.
## Quick start
```python
@ -50,49 +48,10 @@ emits; the records land in the configured root.
(`run.<timestamp>.log[.gz]`) and starts fresh; prunes to `backup_count`.
- `None` — single file, no rotation.
- **compress=True** (default) gzips each rolled file (`run.log.2026-06-27.gz`).
- **Retention** = `backup_count` (default 14) for every mode — unless tiered retention is
enabled (below).
- **Retention** = `backup_count` (default 14) for every mode.
- **console=True** (off by default) also logs to stdout in the same format — opt in when
you want live terminal output alongside the file.
The `name` you pass is normalized so it produces exactly one `.log`: `name="latest"` and
`name="latest.log"` both yield the live file `latest.log` (never `latest.log.log`).
## Tiered retention (`keep_uncompressed` / `keep_compressed`)
The default is a flat `backup_count`: every rolled file is gzipped on roll and the oldest
are deleted past the count. If instead you want the recent logs **uncompressed** (read them
without `zcat`) and older ones **gzipped**, pass the two tier knobs:
```python
setup_logging(
name="latest",
rotate="on_start", # works for on_start, daily, and size
keep_uncompressed=3, # newest 3 rolled logs kept PLAIN
keep_compressed=7, # next 7 kept GZIPPED; total retained = 10
)
```
Result in `log_dir` (newest → oldest):
```
latest.log <- live "latest" (stable, tail -f)
latest.<t1>.log latest.<t2>.log latest.<t3>.log <- 3 newest: plain
latest.<t4>.log.gz ... latest.<t10>.log.gz <- next 7: gzipped
(anything past 10 deleted)
```
- Each restart (`on_start`) or roll (`daily`/`size`) moves the live file into `log_dir`,
then re-tiers: newest `keep_uncompressed` stay plain, the next `keep_compressed` are
gzipped in place, the rest deleted. Total kept = `keep_uncompressed + keep_compressed`.
- **Opt-in by presence** — pass either knob to enable tiering. Pass **neither** and
rotation behaves exactly as before (`backup_count` + gzip-on-roll), so existing callers
are unaffected.
- In tiered mode `backup_count` and the gzip-on-roll behavior of `compress` are **ignored**
— the tier counts bound retention instead.
- `keep_uncompressed=0` → everything gzipped; `keep_compressed=0` → only the plain tier.
Retention is count-based (not time-based).
## Output format (`output=`)
Two formats, two needs. Default is `"text"`; the live-file name is the same either way
@ -133,9 +92,7 @@ setup_logging(
level="INFO", # root level everything inherits (str name or logging constant)
module_levels=None, # {logger_name: level} per-logger overrides (exact name match)
rotate="daily", # "daily" | "size" | "on_start" | None
backup_count=14, # rotated files to keep (flat retention; ignored if tiered)
keep_uncompressed=None, # tiered: newest N rolled logs kept PLAIN (opt-in)
keep_compressed=None, # tiered: next M rolled logs kept GZIPPED (opt-in)
backup_count=14, # rotated files to keep (older auto-deleted)
max_bytes=10_000_000, # only for rotate="size"
compress=True, # gzip rolled files
console=False, # also log to stdout (off by default; opt in)
@ -220,4 +177,4 @@ color formatting, per-logger filters, remote handlers.
## Versioning
Releases are tagged `vX.Y.Z`. The install line above pins a release; drop the `@vX.Y.Z` suffix to install the latest unpinned. Pin deliberately for reproducible installs.
Tagged `vX.Y.Z`. Pin the tag.

2
pyproject.toml
View File

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "log_setup"
version = "0.4.1"
version = "0.3.1"
description = "stdlib app-entry-point logging setup: live run.log, rotation, gzip, retention, consistent format"
requires-python = ">=3.10"
dependencies = []

2
src/log_setup/__init__.py
View File

@ -19,4 +19,4 @@ from .setup import setup_logging
__all__ = ["setup_logging"]
__version__ = "0.4.1"
__version__ = "0.3.1"

6
src/log_setup/formats.py
View File

@ -46,11 +46,7 @@ class JsonLinesFormatter(logging.Formatter):
if key not in _RESERVED and key not in _OUTPUT_KEYS and not key.startswith("_"):
payload[key] = value
if record.exc_info:
# cache the rendered traceback on the record (as stdlib Formatter does) so a
# second handler/format() of the same record doesn't re-render it
if not record.exc_text:
record.exc_text = self.formatException(record.exc_info)
payload["exc_info"] = record.exc_text
payload["exc_info"] = self.formatException(record.exc_info)
elif record.exc_text:
payload["exc_info"] = record.exc_text
if record.stack_info:

163
src/log_setup/rotation.py
View File

@ -13,42 +13,6 @@ import time
from typing import Callable, Optional, Tuple
def _move(source: str, dest: str) -> None:
"""rename source to dest, falling back to copy+unlink across filesystems
os.replace is atomic but raises OSError(EXDEV) when source and dest are on
different filesystems — exactly the container bind-mount / separate-logs-volume
case this lib targets. fall back to shutil.move (copy+unlink) so the roll still
lands instead of failing every rotation via the handler's silent handleError.
precondition: `dest` is a free, non-directory path (all call sites generate a unique
timestamped/dated dest). os.replace and shutil.move differ on a dest that already
exists as a directory, so this helper is not safe for arbitrary dests — only the
rotation paths that guarantee a fresh file dest.
"""
try:
os.replace(source, dest)
except OSError:
shutil.move(source, dest)
def _gzip_file(source: str, dest: str) -> None:
"""gzip source into dest then remove source (the rolled-file compression idiom)
the source mtime is carried onto dest so a file keeps its position when it crosses
the plain->gz tier boundary — retier ranks by mtime, and a fresh write would
otherwise make a just-compressed file look like the newest one and reshuffle tiers.
"""
mtime = _safe_mtime(source)
with open(source, "rb") as src, gzip.open(dest, "wb") as dst:
shutil.copyfileobj(src, dst)
os.remove(source)
try:
os.utime(dest, (mtime, mtime))
except OSError:
pass
def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]:
"""namer: redirect a rolled filename into log_dir, adding .gz when compressing
@ -65,128 +29,47 @@ def make_namer(log_dir: str, compress: bool) -> Callable[[str], str]:
def make_rotator(
compress: bool, log_dir: Optional[str] = None,
prune_stem: Optional[str] = None, backup_count: int = 0,
keep_uncompressed: Optional[int] = None, keep_compressed: Optional[int] = None,
) -> Callable[[str, str], None]:
"""rotator: move (or gzip) the source live file to the destination rolled path
legacy mode (default): gzip on roll when `compress`, then prune `log_dir` to
`backup_count` newest rolled files. the stdlib handler's own retention
prunes `log_dir` to `backup_count` newest rolled files after each roll when
`log_dir`/`prune_stem` are given. the stdlib handler's own retention
(`getFilesToDelete`) only scans the live file's directory, so it never sees the
rolled files we redirect into `log_dir` — pruning here is what bounds retention for
the daily and size rolling modes.
tiered mode (when `keep_uncompressed`/`keep_compressed` are given): land the rolled
file PLAIN and re-tier `log_dir` — newest `keep_uncompressed` stay uncompressed, the
next `keep_compressed` are gzipped, the rest deleted. `compress`/`backup_count` are
ignored in this mode (the tier counts bound retention instead).
rolled files we redirect into `log_dir` — pruning here is what actually bounds
retention for the daily and size rolling modes.
"""
tiered = keep_uncompressed is not None or keep_compressed is not None
def rotator(source: str, dest: str) -> None:
if not os.path.exists(source):
return
if tiered:
# dest carries the namer's .gz suffix in compress mode; strip it so the
# freshly-rolled file lands plain and retier decides its tier
plain_dest = dest[:-3] if dest.endswith(".gz") else dest
_move(source, plain_dest)
if log_dir is not None and prune_stem is not None:
retier(log_dir, prune_stem, keep_uncompressed or 0, keep_compressed or 0)
return
if compress:
_gzip_file(source, dest)
with open(source, "rb") as src, gzip.open(dest, "wb") as dst:
shutil.copyfileobj(src, dst)
os.remove(source)
else:
_move(source, dest)
os.replace(source, dest)
if log_dir is not None and prune_stem is not None:
prune(log_dir, prune_stem, backup_count)
return rotator
def rotate_on_start(
live_path: str, log_dir: str, compress: bool, clock=time.localtime,
keep_uncompressed: Optional[int] = None, keep_compressed: Optional[int] = None,
) -> None:
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.
tiered mode (when `keep_uncompressed`/`keep_compressed` are given): the rolled file
always lands PLAIN (so it can occupy the newest uncompressed tier) and `retier`
decides compression/deletion across the whole stem — `compress` is ignored for the
just-rolled file.
"""
if not os.path.exists(live_path):
return
tiered = keep_uncompressed is not None or keep_compressed is not None
stem = os.path.splitext(os.path.basename(live_path))[0]
stamp = time.strftime("%Y-%m-%d_%H-%M-%S", clock())
suffix = ".log.gz" if (compress and not tiered) else ".log"
# the stamp is 1-second resolution; two starts in the same second would collide
# and the second clobber the first. disambiguate with a numeric counter so a rapid
# crash-restart loop doesn't lose the earlier rolled file. check BOTH the .log and
# .log.gz forms of each candidate: in tiered mode an earlier same-stamp roll may have
# already been compressed to .log.gz, and reusing its bare stem would create a second
# file for the same logical roll and break the tier counts
def _taken(path: str) -> bool:
base = path[:-3] if path.endswith(".gz") else path
return os.path.exists(base) or os.path.exists(base + ".gz")
dest = os.path.join(log_dir, f"{stem}.{stamp}{suffix}")
counter = 1
while _taken(dest):
dest = os.path.join(log_dir, f"{stem}.{stamp}.{counter}{suffix}")
counter += 1
if compress and not tiered:
_gzip_file(live_path, dest)
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:
_move(live_path, dest)
if tiered:
retier(log_dir, stem, keep_uncompressed or 0, keep_compressed or 0)
def retier(log_dir: str, stem: str, keep_uncompressed: int, keep_compressed: int) -> None:
"""re-tier rolled files for stem: newest plain, next gzipped, rest deleted
newest-first by mtime: the first `keep_uncompressed` stay uncompressed, the next
`keep_compressed` are gzipped in place (a still-plain file in that band is compressed
to <name>.gz and the plain source removed), and everything beyond
keep_uncompressed+keep_compressed is deleted. the live <stem>.log is never touched.
fail-soft per file (skip on OSError) so retention never crashes setup.
`stem` is reduced to its basename: rolled files land in log_dir under the basename
(the namer/rotate_on_start basename them), so a `name` containing a directory (e.g.
"sub/run") must be matched by "run." here or nothing matches and retention silently
never fires (unbounded pileup).
"""
stem = os.path.basename(stem)
try:
names = [
name for name in os.listdir(log_dir)
if name.startswith(f"{stem}.") and name != f"{stem}.log"
]
except OSError:
return
entries = [os.path.join(log_dir, name) for name in names]
files = [(p, _safe_mtime(p)) for p in entries if os.path.isfile(p)]
files.sort(key=lambda pair: pair[1], reverse=True)
keep = keep_uncompressed + keep_compressed
for index, (path, _) in enumerate(files):
if index >= keep:
try:
os.remove(path)
except OSError:
pass
elif index >= keep_uncompressed and not path.endswith(".gz"):
dest = path + ".gz"
if os.path.exists(dest):
continue
try:
_gzip_file(path, dest)
except OSError:
pass
os.replace(live_path, dest)
def prune(log_dir: str, stem: str, backup_count: int) -> None:
@ -194,14 +77,9 @@ def prune(log_dir: str, stem: str, backup_count: int) -> None:
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.
`stem` is reduced to its basename so a `name` containing a directory (e.g. "sub/run")
still matches the basenamed rolled files in log_dir (else nothing matches and old
files pile up forever).
"""
if backup_count <= 0:
return
stem = os.path.basename(stem)
try:
entries = [
os.path.join(log_dir, name)
@ -230,19 +108,14 @@ def _safe_mtime(path: str) -> float:
def attach_rolling(
handler, log_dir: str, compress: bool,
prune_stem: Optional[str] = None, backup_count: int = 0,
keep_uncompressed: Optional[int] = None, keep_compressed: Optional[int] = None,
) -> Tuple[Callable, Callable]:
"""wire the custom namer + rotator onto a rotating handler; return them
pass `prune_stem`/`backup_count` so the rotator prunes `log_dir` after each roll
(the handler's own retention can't see the redirected rolled files). pass
`keep_uncompressed`/`keep_compressed` instead to use tiered retention (newest plain,
next gzipped, rest deleted) — see make_rotator.
(the handler's own retention can't see the redirected rolled files).
"""
namer = make_namer(log_dir, compress)
rotator = make_rotator(
compress, log_dir, prune_stem, backup_count, keep_uncompressed, keep_compressed,
)
rotator = make_rotator(compress, log_dir, prune_stem, backup_count)
handler.namer = namer
handler.rotator = rotator
return namer, rotator

83
src/log_setup/setup.py
View File

@ -12,7 +12,7 @@ import atexit
import logging
import logging.handlers
import os
import queue as _queue
import queue
from typing import Dict, Optional, Union
from .formats import build_formatter
@ -22,15 +22,10 @@ log = logging.getLogger(__name__)
_MARKER = "_log_setup_owned"
_listener = None
_atexit_registered = False
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, bool):
# bool is an int subclass (True==1, below DEBUG) but is never a real level —
# reject it consistently with the per-module path rather than set level 1
return logging.INFO
if isinstance(level, int):
return level
if not isinstance(level, str):
@ -81,14 +76,6 @@ def _clear_owned(root: logging.Logger) -> None:
global _listener
if _listener is not None:
_listener.stop()
# the listener owns the real file/console handlers (only the QueueHandler is
# root-attached + marked); stopping it doesn't close them, so close them here
# to avoid relying on GC finalizers across a re-setup
for wrapped in getattr(_listener, "handlers", ()):
try:
wrapped.close()
except Exception:
log.warning("log_setup: failed to close queued handler %r", wrapped, exc_info=True)
_listener = None
for handler in list(root.handlers):
if getattr(handler, _MARKER, False):
@ -96,9 +83,7 @@ def _clear_owned(root: logging.Logger) -> None:
try:
handler.close()
except Exception:
# a handler failing to close must not abort re-setup, but log it
# rather than swallow silently (consistent with the lib's warn pattern)
log.warning("log_setup: failed to close handler %r during re-setup", handler, exc_info=True)
pass
def _tag(handler: logging.Handler) -> logging.Handler:
@ -107,50 +92,23 @@ def _tag(handler: logging.Handler) -> logging.Handler:
return handler
def _normalize_name(name: str) -> str:
"""strip one trailing '.log' (case-insensitive) so the stem is extension-free
`name` is allowed to be passed with or without the extension — "latest" and
"latest.log" both yield stem "latest" (live file latest.log), never latest.log.log.
only one level is stripped: "app.log.log" -> "app.log" so a legit ".log" inside a
name survives.
"""
if name.lower().endswith(".log"):
return name[:-4]
return name
def _file_handler(
name: str, live_path: str, log_dir: str, rotate: Optional[str],
backup_count: int, max_bytes: int, compress: bool,
keep_uncompressed: Optional[int], keep_compressed: Optional[int],
) -> logging.Handler:
"""build the configured file handler with custom rolling into log_dir"""
tiered = keep_uncompressed is not None or keep_compressed is not None
if rotate == "size":
handler = logging.handlers.RotatingFileHandler(
live_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8",
)
attach_rolling(
handler, log_dir, compress, prune_stem=name, backup_count=backup_count,
keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
)
attach_rolling(handler, log_dir, compress, prune_stem=name, backup_count=backup_count)
elif rotate == "daily":
handler = logging.handlers.TimedRotatingFileHandler(
live_path, when="midnight", backupCount=backup_count, encoding="utf-8",
)
attach_rolling(
handler, log_dir, compress, prune_stem=name, backup_count=backup_count,
keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
)
attach_rolling(handler, log_dir, compress, prune_stem=name, backup_count=backup_count)
else:
if rotate == "on_start":
if tiered:
rotate_on_start(
live_path, log_dir, compress,
keep_uncompressed=keep_uncompressed, keep_compressed=keep_compressed,
)
else:
rotate_on_start(live_path, log_dir, compress)
prune(log_dir, name, backup_count)
handler = logging.FileHandler(live_path, encoding="utf-8")
@ -164,8 +122,6 @@ def setup_logging(
module_levels: Optional[Dict[str, Union[int, str]]] = None,
rotate: Optional[str] = "daily",
backup_count: int = 14,
keep_uncompressed: Optional[int] = None,
keep_compressed: Optional[int] = None,
max_bytes: int = 10_000_000,
compress: bool = True,
console: bool = False,
@ -176,16 +132,7 @@ def setup_logging(
) -> 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`. a
trailing ".log" in `name` is stripped so "latest" and "latest.log" both produce the
live file latest.log (never latest.log.log).
`keep_uncompressed`/`keep_compressed` (default None) enable TIERED retention: when
either is given, rolled files are kept as the newest `keep_uncompressed` uncompressed
+ the next `keep_compressed` gzipped, and the rest are deleted (total retained =
sum). this applies to "on_start", "daily", and "size". `backup_count` and the
gzip-on-roll behavior of `compress` are IGNORED in tiered mode (the tier counts bound
retention). pass NEITHER knob and rotation behaves exactly as before (backup_count +
compress) — existing callers are unaffected.
`name` -> <name>.log live file at cwd; rolled/compressed copies go to `log_dir`.
`level` is the root default every logger inherits. `module_levels` is an optional
map of exact logger name -> level applied after the root is set, the ergonomic way
to quiet noisy dependencies (e.g. {"motor": "WARNING", "aiohttp": "WARNING"}) from
@ -207,7 +154,7 @@ def setup_logging(
unwritable `log_dir` falls back to console-only with a warning even when `console` is
off, so output is never silently lost; an unknown `output` falls back to text.
"""
global _listener, _atexit_registered
global _listener
root = logging.getLogger()
root.setLevel(_level_value(level))
@ -215,8 +162,7 @@ def setup_logging(
_clear_owned(root)
formatter = build_formatter(output, fmt, datefmt)
stem = _normalize_name(name)
live_path = f"{stem}.log"
live_path = f"{name}.log"
handlers = []
@ -228,10 +174,7 @@ def setup_logging(
if file_ok:
try:
fh = _file_handler(
stem, live_path, log_dir, rotate, backup_count, max_bytes, compress,
keep_uncompressed, keep_compressed,
)
fh = _file_handler(name, live_path, log_dir, rotate, backup_count, max_bytes, compress)
fh.setFormatter(formatter)
handlers.append(fh)
except OSError:
@ -243,16 +186,12 @@ def setup_logging(
handlers.append(sh)
if queue:
record_queue: "_queue.Queue" = _make_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()
if not _atexit_registered:
# register once — atexit doesn't dedupe, so repeated queue re-setups would
# otherwise stack identical callbacks (harmless but unbounded)
atexit.register(_stop_listener)
_atexit_registered = True
else:
for handler in handlers:
root.addHandler(_tag(handler))
@ -263,9 +202,9 @@ def setup_logging(
return root
def _make_queue() -> "_queue.Queue":
def _make_queue() -> "queue.Queue":
"""unbounded in-memory queue for the QueueHandler -> QueueListener path"""
return _queue.Queue(-1)
return queue.Queue(-1)
def _stop_listener() -> None: