From 609202c1e9a351a429913563af542478f3ba5ca8 Mon Sep 17 00:00:00 2001 From: anon Date: Thu, 18 Jun 2026 22:18:39 +0200 Subject: [PATCH 1/8] feat(logging): shared logger skeleton with Rule extension One `scverse` parent logger + single handler (rich auto-detected, else plain), package loggers as children, central control via `config`. Sole extension point is `Rule` (keep/rewrite); ships `Elapsed`/`Deep` universal rules on by default. Opt-in `get_logger(name, timed=True)` adds scanpy-style `time=`/`deep=`/`.hint()` and a datetime return. Co-Authored-By: Claude Opus 4.8 (1M context) --- pyproject.toml | 1 + src/scverse_misc/logging.py | 253 ++++++++++++++++++++++++++++++++++++ 2 files changed, 254 insertions(+) create mode 100644 src/scverse_misc/logging.py diff --git a/pyproject.toml b/pyproject.toml index 5469e9b..5437af1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ dependencies = [ "session-info2", "typing-extensions; python_version<'3.13'", ] +optional-dependencies.rich = [ "rich" ] optional-dependencies.settings = [ "pydantic-settings", "python-dotenv" ] optional-dependencies.sphinx = [ "pydocstring-rs>=0.1.13", "sphinx>=9" ] # https://docs.pypi.org/project_metadata/#project-urls diff --git a/src/scverse_misc/logging.py b/src/scverse_misc/logging.py new file mode 100644 index 0000000..30c8d8c --- /dev/null +++ b/src/scverse_misc/logging.py @@ -0,0 +1,253 @@ +"""Shared logger for scverse packages. + +Skeleton: one ``scverse`` parent logger owning a single handler (rich if +installed, else plain), with package loggers as children. Control via +:data:`config`. The only extension point is :class:`Rule` — subclass it to +filter or rewrite output; register any number with :meth:`config.add_rule`. + +Two **universal rules** ship enabled by default and are no-ops until a record +carries the matching attribute: + +- :class:`Elapsed` renders ``record.time_passed`` (a ``timedelta``): substitutes + ``{time_passed}`` in the message, else appends ``(H:MM:SS)``. +- :class:`Deep` appends ``record.deep`` as ``": detail"``. + +Because rules run on the handler, they render identically under rich and plain. + +scanpy's ``time=`` / ``deep=`` keywords and the ``-> datetime`` return are a +call-site concern a rule cannot provide (a rule runs after the call returns), +so they live in an **opt-in** logger: ``get_logger("scanpy", timed=True)``:: + + log = get_logger("scanpy", timed=True) + t = log.info("normalizing") # returns a datetime + log.info("finished ({time_passed})", time=t) # Elapsed rule renders it + log.info("done", time=t, deep="42 cells dropped") +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta + +__all__ = ["Rule", "Elapsed", "Deep", "config", "get_logger"] + +_ROOT = "scverse" +HINT = (logging.INFO + logging.DEBUG) // 2 # 15; used by the timed logger +logging.addLevelName(HINT, "HINT") + + +class Rule(logging.Filter): + """A logging rule — subclass and override what you need; both default to no-ops. + + - :meth:`keep` ``(record) -> bool`` — return ``False`` to drop the record. + - :meth:`rewrite` ``(message, record) -> str`` — return the new text + (``message`` is the fully formatted string). + + Rules run in registration order on the shared handler, for every package. + """ + + def keep(self, record: logging.LogRecord) -> bool: + return True + + def rewrite(self, message: str, record: logging.LogRecord) -> str: + return message + + def filter(self, record: logging.LogRecord) -> bool: # stdlib hook; don't override + if not self.keep(record): + return False + message = record.getMessage() # always a str, %-args already expanded + new = self.rewrite(message, record) + if new != message: + record.msg, record.args = new, () + return True + + +class Elapsed(Rule): + """Render ``record.time_passed`` (a ``timedelta``). Universal, enabled by default.""" + + def rewrite(self, message: str, record: logging.LogRecord) -> str: + td = getattr(record, "time_passed", None) + if not td: # None or zero -> show nothing (matches scanpy) + return message + td = timedelta(seconds=int(td.total_seconds())) # strip sub-second noise + if "{time_passed}" in message: + return message.replace("{time_passed}", str(td)) + return f"{message} ({td})" + + +class Deep(Rule): + """Append ``record.deep`` as detail. Universal, enabled by default.""" + + def rewrite(self, message: str, record: logging.LogRecord) -> str: + deep = getattr(record, "deep", None) + return message if deep is None else f"{message}: {deep}" + + +def _make_handler(use_rich: bool) -> logging.Handler: + if use_rich: + from rich.logging import RichHandler + + return RichHandler(show_path=False, show_time=False) # rich renders the level itself + handler = logging.StreamHandler() + handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s")) + return handler + + +def _rich_available() -> bool: + from importlib.util import find_spec + + return find_spec("rich") is not None + + +class _Config: + """Central logging configuration; the singleton instance is :data:`config`.""" + + def __init__(self) -> None: + self._parent = logging.getLogger(_ROOT) + self._parent.setLevel(logging.WARNING) + self._parent.propagate = False # one handler here; don't double-log via root + self._rules: list = [Elapsed(), Deep()] # universal defaults; order matters + self._install(_make_handler(_rich_available())) + + def _install(self, handler: logging.Handler) -> None: + for r in self._rules: + handler.addFilter(r) + self._parent.addHandler(handler) + + @property + def verbosity(self): + """Central level for all scverse loggers. Set with a name (``"info"``) or int.""" + return logging.getLevelName(self._parent.level) + + @verbosity.setter + def verbosity(self, level) -> None: + self._parent.setLevel(level.upper() if isinstance(level, str) else level) + + def use_rich(self, enabled: bool = True) -> None: + """Force the rich (``True``) or plain (``False``) handler.""" + for h in list(self._parent.handlers): + self._parent.removeHandler(h) + self._install(_make_handler(enabled)) + + def add_rule(self, rule: Rule) -> None: + self._rules.append(rule) + for h in self._parent.handlers: + h.addFilter(rule) + + def remove_rule(self, rule: Rule) -> None: + if rule in self._rules: + self._rules.remove(rule) + for h in self._parent.handlers: + h.removeFilter(rule) + + +config = _Config() + + +class _TimedLogger: + """Opt-in scanpy-style wrapper: ``time=``/``deep=`` keywords + a ``datetime`` return. + + Sets ``time_passed``/``deep`` on the record (rendered by the :class:`Elapsed` + and :class:`Deep` rules) and returns the current time so callers can thread it. + Everything else delegates to the underlying real logger. + """ + + def __init__(self, logger: logging.Logger) -> None: + self._logger = logger + + def __getattr__(self, name): + if name == "_logger": + raise AttributeError(name) + return getattr(self._logger, name) + + def _emit(self, level, msg, *args, time=None, deep=None, **kw) -> datetime: + now = datetime.now() + if self._logger.isEnabledFor(level): + extra = dict(kw.pop("extra", None) or {}) + if time is not None: + extra["time_passed"] = now - time # thread the returned value (both naive) + if deep is not None and self._logger.getEffectiveLevel() < level: + extra["deep"] = deep + kw["stacklevel"] = kw.get("stacklevel", 1) + 2 # skip _emit + the level method + self._logger.log(level, msg, *args, extra=extra, **kw) + return now + + def debug(self, msg, *a, **k) -> datetime: + return self._emit(logging.DEBUG, msg, *a, **k) + + def hint(self, msg, *a, **k) -> datetime: + return self._emit(HINT, msg, *a, **k) + + def info(self, msg, *a, **k) -> datetime: + return self._emit(logging.INFO, msg, *a, **k) + + def warning(self, msg, *a, **k) -> datetime: + return self._emit(logging.WARNING, msg, *a, **k) + + def error(self, msg, *a, **k) -> datetime: + return self._emit(logging.ERROR, msg, *a, **k) + + def critical(self, msg, *a, **k) -> datetime: + return self._emit(logging.CRITICAL, msg, *a, **k) + + +def get_logger(name: str, *, timed: bool = False): + """Return the ``scverse.`` logger a package should use. + + ``timed=False`` (default) returns a plain :class:`logging.Logger`. + ``timed=True`` returns a :class:`_TimedLogger` with scanpy-style ``time=`` / + ``deep=`` keywords and a ``datetime`` return. + """ + logger = logging.getLogger(name if name.startswith(f"{_ROOT}.") else f"{_ROOT}.{name}") + return _TimedLogger(logger) if timed else logger + + +if __name__ == "__main__": + # ponytail: one runnable check. `python -m scverse_misc.logging`. + import io + + # capture the shared handler's output via a plain formatter we control + config.use_rich(False) + buf = io.StringIO() + config._parent.handlers[0].setStream(buf) # type: ignore[attr-defined] + config.verbosity = "debug" + + plain = get_logger("selftest") + assert plain.name == "scverse.selftest" and plain.parent.name == "scverse" + assert isinstance(plain, logging.Logger) # real logger: critical/getChild/etc. work + + log = get_logger("selftest", timed=True) + t = log.info("start") + assert isinstance(t, datetime) + + # {time_passed} substitution + append, via the Elapsed rule (works rich or plain) + log.info("finished ({time_passed})", time=t - timedelta(seconds=5)) + log.info("done", time=t - timedelta(seconds=2)) + # deep, via the Deep rule (shown because verbosity debug < info level) + log.info("normalized", deep="3 cells dropped") + out = buf.getvalue() + assert "finished (0:00:05)" in out, out + assert "done (0:00:02)" in out, out + assert "normalized: 3 cells dropped" in out, out + + # falsy deep is preserved (not dropped by a truthiness check) + buf.truncate(0); buf.seek(0) + log.info("count", deep=0) + assert "count: 0" in buf.getvalue(), buf.getvalue() + + # a user Rule composes with the defaults; central verbosity gates output + class Tag(Rule): + def rewrite(self, message, record): + return f"[{record.name.rsplit('.', 1)[-1]}] {message}" + + tag = Tag() + config.add_rule(tag) + buf.truncate(0); buf.seek(0) + log.warning("hi") + assert "[selftest] hi" in buf.getvalue(), buf.getvalue() + config.remove_rule(tag) + + config.verbosity = "warning" + assert not plain.isEnabledFor(logging.INFO) and plain.isEnabledFor(logging.WARNING) + + print("ok") From 4a04ddc58a80aafc89e0850475817180ff00b670 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Thu, 18 Jun 2026 20:19:16 +0000 Subject: [PATCH 2/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- src/scverse_misc/logging.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/scverse_misc/logging.py b/src/scverse_misc/logging.py index 30c8d8c..5d24fbb 100644 --- a/src/scverse_misc/logging.py +++ b/src/scverse_misc/logging.py @@ -19,8 +19,8 @@ so they live in an **opt-in** logger: ``get_logger("scanpy", timed=True)``:: log = get_logger("scanpy", timed=True) - t = log.info("normalizing") # returns a datetime - log.info("finished ({time_passed})", time=t) # Elapsed rule renders it + t = log.info("normalizing") # returns a datetime + log.info("finished ({time_passed})", time=t) # Elapsed rule renders it log.info("done", time=t, deep="42 cells dropped") """ @@ -231,7 +231,8 @@ def get_logger(name: str, *, timed: bool = False): assert "normalized: 3 cells dropped" in out, out # falsy deep is preserved (not dropped by a truthiness check) - buf.truncate(0); buf.seek(0) + buf.truncate(0) + buf.seek(0) log.info("count", deep=0) assert "count: 0" in buf.getvalue(), buf.getvalue() @@ -242,7 +243,8 @@ def rewrite(self, message, record): tag = Tag() config.add_rule(tag) - buf.truncate(0); buf.seek(0) + buf.truncate(0) + buf.seek(0) log.warning("hi") assert "[selftest] hi" in buf.getvalue(), buf.getvalue() config.remove_rule(tag) From 26063a5a1ac99d19023320da85642218e10ba109 Mon Sep 17 00:00:00 2001 From: anon Date: Thu, 18 Jun 2026 22:46:13 +0200 Subject: [PATCH 3/8] test(logging): cover logger module + fix lint/type errors The logging module shipped its checks inside `if __name__ == "__main__"`, which CI never runs, so the 255 new lines had 0% coverage; the same block plus the module body also failed ruff (missing annotations/docstrings, E702) and would fail strict mypy. - annotate the public surface (`_emit`, level methods, `verbosity`, `get_logger` via @overload so `timed=True` narrows to `_TimedLogger`); add the missing `Rule` docstrings; type `_rules: list[Rule]`; `__getattr__ -> Any` for transparent delegation - trim `_emit` to the documented `time=`/`deep=` surface (the old `**kwargs` passthrough never passed strict mypy) - remove the `__main__` self-check, promote it to tests/test_logging.py (18 tests; logging.py coverage 0% -> 99%) Verified locally: ruff check + format, mypy --strict, pytest all pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/scverse_misc/logging.py | 112 ++++++++--------------- tests/test_logging.py | 171 ++++++++++++++++++++++++++++++++++++ 2 files changed, 209 insertions(+), 74 deletions(-) create mode 100644 tests/test_logging.py diff --git a/src/scverse_misc/logging.py b/src/scverse_misc/logging.py index 5d24fbb..b6f5c3d 100644 --- a/src/scverse_misc/logging.py +++ b/src/scverse_misc/logging.py @@ -28,6 +28,7 @@ import logging from datetime import datetime, timedelta +from typing import Any, Literal, overload __all__ = ["Rule", "Elapsed", "Deep", "config", "get_logger"] @@ -47,12 +48,15 @@ class Rule(logging.Filter): """ def keep(self, record: logging.LogRecord) -> bool: + """Return ``False`` to drop the record (default: keep everything).""" return True def rewrite(self, message: str, record: logging.LogRecord) -> str: + """Return the new message text (default: unchanged).""" return message def filter(self, record: logging.LogRecord) -> bool: # stdlib hook; don't override + """Stdlib hook: apply :meth:`keep` then :meth:`rewrite`. Don't override.""" if not self.keep(record): return False message = record.getMessage() # always a str, %-args already expanded @@ -66,6 +70,7 @@ class Elapsed(Rule): """Render ``record.time_passed`` (a ``timedelta``). Universal, enabled by default.""" def rewrite(self, message: str, record: logging.LogRecord) -> str: + """Substitute ``{time_passed}`` if present, else append ``(H:MM:SS)``.""" td = getattr(record, "time_passed", None) if not td: # None or zero -> show nothing (matches scanpy) return message @@ -79,6 +84,7 @@ class Deep(Rule): """Append ``record.deep`` as detail. Universal, enabled by default.""" def rewrite(self, message: str, record: logging.LogRecord) -> str: + """Append ``record.deep`` as ``": detail"`` when present.""" deep = getattr(record, "deep", None) return message if deep is None else f"{message}: {deep}" @@ -106,7 +112,7 @@ def __init__(self) -> None: self._parent = logging.getLogger(_ROOT) self._parent.setLevel(logging.WARNING) self._parent.propagate = False # one handler here; don't double-log via root - self._rules: list = [Elapsed(), Deep()] # universal defaults; order matters + self._rules: list[Rule] = [Elapsed(), Deep()] # universal defaults; order matters self._install(_make_handler(_rich_available())) def _install(self, handler: logging.Handler) -> None: @@ -115,12 +121,12 @@ def _install(self, handler: logging.Handler) -> None: self._parent.addHandler(handler) @property - def verbosity(self): + def verbosity(self) -> str | int: """Central level for all scverse loggers. Set with a name (``"info"``) or int.""" return logging.getLevelName(self._parent.level) @verbosity.setter - def verbosity(self, level) -> None: + def verbosity(self, level: str | int) -> None: self._parent.setLevel(level.upper() if isinstance(level, str) else level) def use_rich(self, enabled: bool = True) -> None: @@ -155,43 +161,54 @@ class _TimedLogger: def __init__(self, logger: logging.Logger) -> None: self._logger = logger - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: # noqa: ANN401 # transparent delegation to the real logger if name == "_logger": raise AttributeError(name) return getattr(self._logger, name) - def _emit(self, level, msg, *args, time=None, deep=None, **kw) -> datetime: + def _emit( + self, + level: int, + msg: object, + *args: object, + time: datetime | None = None, + deep: object = None, + ) -> datetime: now = datetime.now() if self._logger.isEnabledFor(level): - extra = dict(kw.pop("extra", None) or {}) + extra: dict[str, object] = {} if time is not None: extra["time_passed"] = now - time # thread the returned value (both naive) if deep is not None and self._logger.getEffectiveLevel() < level: extra["deep"] = deep - kw["stacklevel"] = kw.get("stacklevel", 1) + 2 # skip _emit + the level method - self._logger.log(level, msg, *args, extra=extra, **kw) + # stacklevel=3: skip _emit + the level method so the call-site is the caller + self._logger.log(level, msg, *args, extra=extra, stacklevel=3) return now - def debug(self, msg, *a, **k) -> datetime: - return self._emit(logging.DEBUG, msg, *a, **k) + def debug(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + return self._emit(logging.DEBUG, msg, *a, time=time, deep=deep) - def hint(self, msg, *a, **k) -> datetime: - return self._emit(HINT, msg, *a, **k) + def hint(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + return self._emit(HINT, msg, *a, time=time, deep=deep) - def info(self, msg, *a, **k) -> datetime: - return self._emit(logging.INFO, msg, *a, **k) + def info(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + return self._emit(logging.INFO, msg, *a, time=time, deep=deep) - def warning(self, msg, *a, **k) -> datetime: - return self._emit(logging.WARNING, msg, *a, **k) + def warning(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + return self._emit(logging.WARNING, msg, *a, time=time, deep=deep) - def error(self, msg, *a, **k) -> datetime: - return self._emit(logging.ERROR, msg, *a, **k) + def error(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + return self._emit(logging.ERROR, msg, *a, time=time, deep=deep) - def critical(self, msg, *a, **k) -> datetime: - return self._emit(logging.CRITICAL, msg, *a, **k) + def critical(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + return self._emit(logging.CRITICAL, msg, *a, time=time, deep=deep) -def get_logger(name: str, *, timed: bool = False): +@overload +def get_logger(name: str, *, timed: Literal[False] = False) -> logging.Logger: ... +@overload +def get_logger(name: str, *, timed: Literal[True]) -> _TimedLogger: ... +def get_logger(name: str, *, timed: bool = False) -> logging.Logger | _TimedLogger: """Return the ``scverse.`` logger a package should use. ``timed=False`` (default) returns a plain :class:`logging.Logger`. @@ -200,56 +217,3 @@ def get_logger(name: str, *, timed: bool = False): """ logger = logging.getLogger(name if name.startswith(f"{_ROOT}.") else f"{_ROOT}.{name}") return _TimedLogger(logger) if timed else logger - - -if __name__ == "__main__": - # ponytail: one runnable check. `python -m scverse_misc.logging`. - import io - - # capture the shared handler's output via a plain formatter we control - config.use_rich(False) - buf = io.StringIO() - config._parent.handlers[0].setStream(buf) # type: ignore[attr-defined] - config.verbosity = "debug" - - plain = get_logger("selftest") - assert plain.name == "scverse.selftest" and plain.parent.name == "scverse" - assert isinstance(plain, logging.Logger) # real logger: critical/getChild/etc. work - - log = get_logger("selftest", timed=True) - t = log.info("start") - assert isinstance(t, datetime) - - # {time_passed} substitution + append, via the Elapsed rule (works rich or plain) - log.info("finished ({time_passed})", time=t - timedelta(seconds=5)) - log.info("done", time=t - timedelta(seconds=2)) - # deep, via the Deep rule (shown because verbosity debug < info level) - log.info("normalized", deep="3 cells dropped") - out = buf.getvalue() - assert "finished (0:00:05)" in out, out - assert "done (0:00:02)" in out, out - assert "normalized: 3 cells dropped" in out, out - - # falsy deep is preserved (not dropped by a truthiness check) - buf.truncate(0) - buf.seek(0) - log.info("count", deep=0) - assert "count: 0" in buf.getvalue(), buf.getvalue() - - # a user Rule composes with the defaults; central verbosity gates output - class Tag(Rule): - def rewrite(self, message, record): - return f"[{record.name.rsplit('.', 1)[-1]}] {message}" - - tag = Tag() - config.add_rule(tag) - buf.truncate(0) - buf.seek(0) - log.warning("hi") - assert "[selftest] hi" in buf.getvalue(), buf.getvalue() - config.remove_rule(tag) - - config.verbosity = "warning" - assert not plain.isEnabledFor(logging.INFO) and plain.isEnabledFor(logging.WARNING) - - print("ok") diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..5d207e7 --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import io +import logging +from datetime import datetime, timedelta +from typing import TYPE_CHECKING + +import pytest + +from scverse_misc import logging as mod +from scverse_misc.logging import Deep, Elapsed, Rule, config, get_logger + +if TYPE_CHECKING: + from collections.abc import Generator + + +@pytest.fixture +def sink() -> Generator[io.StringIO, None, None]: + """Force the plain handler, capture its output, and restore global state after.""" + old_level = config._parent.level + old_rules = list(config._rules) + config.use_rich(False) + handler = config._parent.handlers[0] + assert isinstance(handler, logging.StreamHandler) + buf = io.StringIO() + handler.setStream(buf) + config.verbosity = "debug" + try: + yield buf + finally: + # drop any rules a test added, then restore level + a clean plain handler + for rule in list(config._rules): + if rule not in old_rules: + config.remove_rule(rule) + config._parent.setLevel(old_level) + + +def test_get_logger_plain_naming() -> None: + plain = get_logger("selftest") + assert isinstance(plain, logging.Logger) + assert plain.name == "scverse.selftest" + assert plain.parent is not None and plain.parent.name == "scverse" + + +def test_get_logger_does_not_double_prefix() -> None: + assert get_logger("scverse.already").name == "scverse.already" + + +def test_timed_logger_returns_datetime(sink: io.StringIO) -> None: + log = get_logger("selftest", timed=True) + t = log.info("start") + assert isinstance(t, datetime) + + +def test_elapsed_substitution_and_append(sink: io.StringIO) -> None: + log = get_logger("selftest", timed=True) + now = datetime.now() + log.info("finished ({time_passed})", time=now - timedelta(seconds=5)) + log.info("done", time=now - timedelta(seconds=2)) + out = sink.getvalue() + assert "finished (0:00:05)" in out + assert "done (0:00:02)" in out + + +def test_elapsed_noop_without_time(sink: io.StringIO) -> None: + log = get_logger("selftest", timed=True) + log.info("plain message") + out = sink.getvalue() + assert "plain message" in out + assert "(" not in out.split("plain message", 1)[1] + + +def test_deep_appended(sink: io.StringIO) -> None: + log = get_logger("selftest", timed=True) + log.info("normalized", deep="3 cells dropped") + assert "normalized: 3 cells dropped" in sink.getvalue() + + +def test_deep_falsy_zero_preserved(sink: io.StringIO) -> None: + # 0 must not be dropped by a truthiness check (Deep uses `is None`) + log = get_logger("selftest", timed=True) + log.info("count", deep=0) + assert "count: 0" in sink.getvalue() + + +def test_deep_hidden_when_not_below_level(sink: io.StringIO) -> None: + # deep only renders when the effective level is strictly below the call level + config.verbosity = "info" + log = get_logger("selftest", timed=True) + log.info("msg", deep="hidden detail") + out = sink.getvalue() + assert "msg" in out + assert "hidden detail" not in out + + +def test_user_rule_composes_and_verbosity_gates(sink: io.StringIO) -> None: + class Tag(Rule): + def rewrite(self, message: str, record: logging.LogRecord) -> str: + return f"[{record.name.rsplit('.', 1)[-1]}] {message}" + + config.add_rule(Tag()) + log = get_logger("selftest", timed=True) + log.warning("hi") + assert "[selftest] hi" in sink.getvalue() + + +def test_rule_keep_can_drop_record(sink: io.StringIO) -> None: + class DropAll(Rule): + def keep(self, record: logging.LogRecord) -> bool: + return False + + config.add_rule(DropAll()) + get_logger("selftest").warning("should be dropped") + assert sink.getvalue() == "" + + +def test_base_rule_is_passthrough(sink: io.StringIO) -> None: + config.add_rule(Rule()) # base keep()/rewrite() are no-ops + get_logger("selftest").warning("unchanged") + assert "unchanged" in sink.getvalue() + + +def test_verbosity_get_set_by_name_and_int() -> None: + config.verbosity = "warning" + assert config.verbosity == "WARNING" + plain = get_logger("selftest") + assert not plain.isEnabledFor(logging.INFO) + assert plain.isEnabledFor(logging.WARNING) + config.verbosity = logging.DEBUG + assert plain.isEnabledFor(logging.DEBUG) + + +def test_universal_rules_enabled_by_default() -> None: + assert any(isinstance(r, Elapsed) for r in config._rules) + assert any(isinstance(r, Deep) for r in config._rules) + + +def test_hint_level_registered() -> None: + assert logging.getLevelName(mod.HINT) == "HINT" + + +def test_remove_rule_is_idempotent() -> None: + rule = Rule() + config.add_rule(rule) + config.remove_rule(rule) + config.remove_rule(rule) # removing again must not raise + assert rule not in config._rules + + +def test_all_level_methods_emit_and_return_datetime(sink: io.StringIO) -> None: + log = get_logger("selftest", timed=True) + for emit in (log.debug, log.hint, log.info, log.warning, log.error, log.critical): + assert isinstance(emit("msg"), datetime) + + +def test_timed_logger_delegates_unknown_attrs() -> None: + log = get_logger("selftest", timed=True) + # name / getEffectiveLevel aren't defined on _TimedLogger -> __getattr__ delegates + assert log.name == "scverse.selftest" + assert log.getEffectiveLevel() == logging.getLogger("scverse.selftest").getEffectiveLevel() + + +def test_use_rich_installs_rich_handler() -> None: + pytest.importorskip("rich") + from rich.logging import RichHandler + + try: + config.use_rich(True) + assert isinstance(config._parent.handlers[0], RichHandler) + finally: + config.use_rich(False) From 0a49b6a44d10094870b64c6ed1be74738722e0e2 Mon Sep 17 00:00:00 2001 From: anon Date: Thu, 18 Jun 2026 22:51:05 +0200 Subject: [PATCH 4/8] ci(mypy): add rich to mypy additional_dependencies logging.py imports rich.logging lazily in the rich-handler branch. rich is an optional runtime dep, but mypy needs it to resolve RichHandler (else import-not-found + a no-any-return on the Handler return). Matches the existing pattern of listing typecheck-only deps (pytest, sphinx) here. Co-Authored-By: Claude Opus 4.8 (1M context) --- .pre-commit-config.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 582e6bb..018f298 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -46,6 +46,7 @@ repos: - pydantic-settings - pydocstring-rs - pytest + - rich - sphinx - sphinx-autodoc-typehints - sphinxcontrib-katex From f5a31af045030750250527e4849202af99a4055c Mon Sep 17 00:00:00 2001 From: anon Date: Fri, 19 Jun 2026 13:47:27 +0200 Subject: [PATCH 5/8] refactor(logging): address review nits - make TimedLogger public (returned from get_logger, meant to be subclassed) - config.rich settable property for consistency with config.verbosity - rename _Config._parent -> _root (it manages the scverse root logger) Co-Authored-By: Claude Opus 4.8 (1M context) --- src/scverse_misc/logging.py | 56 ++++++++++++++++++++++++------------- tests/test_logging.py | 23 ++++++++------- 2 files changed, 49 insertions(+), 30 deletions(-) diff --git a/src/scverse_misc/logging.py b/src/scverse_misc/logging.py index b6f5c3d..1b9ef7b 100644 --- a/src/scverse_misc/logging.py +++ b/src/scverse_misc/logging.py @@ -30,7 +30,7 @@ from datetime import datetime, timedelta from typing import Any, Literal, overload -__all__ = ["Rule", "Elapsed", "Deep", "config", "get_logger"] +__all__ = ["Rule", "Elapsed", "Deep", "TimedLogger", "config", "get_logger"] _ROOT = "scverse" HINT = (logging.INFO + logging.DEBUG) // 2 # 15; used by the timed logger @@ -109,48 +109,58 @@ class _Config: """Central logging configuration; the singleton instance is :data:`config`.""" def __init__(self) -> None: - self._parent = logging.getLogger(_ROOT) - self._parent.setLevel(logging.WARNING) - self._parent.propagate = False # one handler here; don't double-log via root + self._root = logging.getLogger(_ROOT) + self._root.setLevel(logging.WARNING) + self._root.propagate = False # one handler here; don't double-log via root + self._rich: bool | None = None # None = auto-detect rich self._rules: list[Rule] = [Elapsed(), Deep()] # universal defaults; order matters - self._install(_make_handler(_rich_available())) + self._install(_make_handler(self._use_rich())) def _install(self, handler: logging.Handler) -> None: for r in self._rules: handler.addFilter(r) - self._parent.addHandler(handler) + self._root.addHandler(handler) + + def _use_rich(self) -> bool: + return _rich_available() if self._rich is None else self._rich @property def verbosity(self) -> str | int: """Central level for all scverse loggers. Set with a name (``"info"``) or int.""" - return logging.getLevelName(self._parent.level) + return logging.getLevelName(self._root.level) @verbosity.setter def verbosity(self, level: str | int) -> None: - self._parent.setLevel(level.upper() if isinstance(level, str) else level) + self._root.setLevel(level.upper() if isinstance(level, str) else level) + + @property + def rich(self) -> bool: + """Whether rich rendering is active. Set ``True``/``False`` to force, ``None`` to auto-detect.""" + return self._use_rich() - def use_rich(self, enabled: bool = True) -> None: - """Force the rich (``True``) or plain (``False``) handler.""" - for h in list(self._parent.handlers): - self._parent.removeHandler(h) - self._install(_make_handler(enabled)) + @rich.setter + def rich(self, enabled: bool | None) -> None: + self._rich = enabled + for h in list(self._root.handlers): + self._root.removeHandler(h) + self._install(_make_handler(self._use_rich())) def add_rule(self, rule: Rule) -> None: self._rules.append(rule) - for h in self._parent.handlers: + for h in self._root.handlers: h.addFilter(rule) def remove_rule(self, rule: Rule) -> None: if rule in self._rules: self._rules.remove(rule) - for h in self._parent.handlers: + for h in self._root.handlers: h.removeFilter(rule) config = _Config() -class _TimedLogger: +class TimedLogger: """Opt-in scanpy-style wrapper: ``time=``/``deep=`` keywords + a ``datetime`` return. Sets ``time_passed``/``deep`` on the record (rendered by the :class:`Elapsed` @@ -186,34 +196,40 @@ def _emit( return now def debug(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + """Log at DEBUG; return the current time (see :class:`TimedLogger`).""" return self._emit(logging.DEBUG, msg, *a, time=time, deep=deep) def hint(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + """Log at HINT; return the current time (see :class:`TimedLogger`).""" return self._emit(HINT, msg, *a, time=time, deep=deep) def info(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + """Log at INFO; return the current time (see :class:`TimedLogger`).""" return self._emit(logging.INFO, msg, *a, time=time, deep=deep) def warning(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + """Log at WARNING; return the current time (see :class:`TimedLogger`).""" return self._emit(logging.WARNING, msg, *a, time=time, deep=deep) def error(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + """Log at ERROR; return the current time (see :class:`TimedLogger`).""" return self._emit(logging.ERROR, msg, *a, time=time, deep=deep) def critical(self, msg: object, *a: object, time: datetime | None = None, deep: object = None) -> datetime: + """Log at CRITICAL; return the current time (see :class:`TimedLogger`).""" return self._emit(logging.CRITICAL, msg, *a, time=time, deep=deep) @overload def get_logger(name: str, *, timed: Literal[False] = False) -> logging.Logger: ... @overload -def get_logger(name: str, *, timed: Literal[True]) -> _TimedLogger: ... -def get_logger(name: str, *, timed: bool = False) -> logging.Logger | _TimedLogger: +def get_logger(name: str, *, timed: Literal[True]) -> TimedLogger: ... +def get_logger(name: str, *, timed: bool = False) -> logging.Logger | TimedLogger: """Return the ``scverse.`` logger a package should use. ``timed=False`` (default) returns a plain :class:`logging.Logger`. - ``timed=True`` returns a :class:`_TimedLogger` with scanpy-style ``time=`` / + ``timed=True`` returns a :class:`TimedLogger` with scanpy-style ``time=`` / ``deep=`` keywords and a ``datetime`` return. """ logger = logging.getLogger(name if name.startswith(f"{_ROOT}.") else f"{_ROOT}.{name}") - return _TimedLogger(logger) if timed else logger + return TimedLogger(logger) if timed else logger diff --git a/tests/test_logging.py b/tests/test_logging.py index 5d207e7..a69276a 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -8,7 +8,7 @@ import pytest from scverse_misc import logging as mod -from scverse_misc.logging import Deep, Elapsed, Rule, config, get_logger +from scverse_misc.logging import Deep, Elapsed, Rule, TimedLogger, config, get_logger if TYPE_CHECKING: from collections.abc import Generator @@ -17,10 +17,10 @@ @pytest.fixture def sink() -> Generator[io.StringIO, None, None]: """Force the plain handler, capture its output, and restore global state after.""" - old_level = config._parent.level + old_level = config._root.level old_rules = list(config._rules) - config.use_rich(False) - handler = config._parent.handlers[0] + config.rich = False + handler = config._root.handlers[0] assert isinstance(handler, logging.StreamHandler) buf = io.StringIO() handler.setStream(buf) @@ -32,7 +32,7 @@ def sink() -> Generator[io.StringIO, None, None]: for rule in list(config._rules): if rule not in old_rules: config.remove_rule(rule) - config._parent.setLevel(old_level) + config._root.setLevel(old_level) def test_get_logger_plain_naming() -> None: @@ -48,6 +48,7 @@ def test_get_logger_does_not_double_prefix() -> None: def test_timed_logger_returns_datetime(sink: io.StringIO) -> None: log = get_logger("selftest", timed=True) + assert isinstance(log, TimedLogger) t = log.info("start") assert isinstance(t, datetime) @@ -155,17 +156,19 @@ def test_all_level_methods_emit_and_return_datetime(sink: io.StringIO) -> None: def test_timed_logger_delegates_unknown_attrs() -> None: log = get_logger("selftest", timed=True) - # name / getEffectiveLevel aren't defined on _TimedLogger -> __getattr__ delegates + # name / getEffectiveLevel aren't defined on TimedLogger -> __getattr__ delegates assert log.name == "scverse.selftest" assert log.getEffectiveLevel() == logging.getLogger("scverse.selftest").getEffectiveLevel() -def test_use_rich_installs_rich_handler() -> None: +def test_rich_property_installs_rich_handler() -> None: pytest.importorskip("rich") from rich.logging import RichHandler try: - config.use_rich(True) - assert isinstance(config._parent.handlers[0], RichHandler) + config.rich = True + assert isinstance(config._root.handlers[0], RichHandler) + assert config.rich is True finally: - config.use_rich(False) + config.rich = False + assert config.rich is False From fc9f9745b6fc2e1a26f7d12430d143fa0d4a9b9e Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 19 Jun 2026 14:12:20 +0000 Subject: [PATCH 6/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 9a57e24..a872eee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,8 +27,8 @@ dependencies = [ "session-info2", "typing-extensions; python_version<'3.13'", ] -optional-dependencies.rich = [ "rich" ] optional-dependencies.datasets = [ "pooch", "pyyaml", "tqdm" ] +optional-dependencies.rich = [ "rich" ] optional-dependencies.settings = [ "pydantic-settings", "python-dotenv" ] optional-dependencies.spatialdata = [ "spatialdata" ] optional-dependencies.sphinx = [ "pydocstring-rs>=0.1.13", "sphinx>=9" ] From deae57d0ad1d084d1d1966c5107dd9ed39b315c9 Mon Sep 17 00:00:00 2001 From: anon Date: Fri, 19 Jun 2026 21:11:37 +0200 Subject: [PATCH 7/8] refactor(logging): build config on shared Settings; pydantic as hard dep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Promote pydantic-settings + python-dotenv to hard dependencies (anndata always pulls scverse-misc[settings] in anyway), and rebuild the logging config on the shared Settings base instead of a bespoke _Config. - verbosity/rich become validated Settings fields: env-var loading (SCVERSE_MISC_*), override/reset, and level validation come for free. - rules stay bespoke (add_rule/remove_rule) — they aren't settings. - _root/_rules derive from the live logger/handler, so there's no pydantic private-attr timing to manage and the test suite is unchanged. - keep the (now empty) settings extra for anndata <0.13 back-compat. - drop the dead suppress(ImportError) around Settings in __init__. Addresses review feedback on #46 (integrate logger with global config handling; promote pydantic; rich settable). Co-Authored-By: Claude Opus 4.8 (1M context) --- pyproject.toml | 6 ++- src/scverse_misc/__init__.py | 19 ++++---- src/scverse_misc/logging.py | 93 +++++++++++++++++++++--------------- tests/test_logging.py | 23 +++++++++ 4 files changed, 93 insertions(+), 48 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index a872eee..1c56189 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,13 +23,17 @@ classifiers = [ dynamic = [ "version" ] dependencies = [ "anndata", + "pydantic-settings", + "python-dotenv", # for debug logging (referenced from the issue template) "session-info2", "typing-extensions; python_version<'3.13'", ] optional-dependencies.datasets = [ "pooch", "pyyaml", "tqdm" ] optional-dependencies.rich = [ "rich" ] -optional-dependencies.settings = [ "pydantic-settings", "python-dotenv" ] +# Kept for backwards compat: anndata <0.13 references scverse-misc[settings]. +# The deps are now hard dependencies (see above), so this extra is empty. +optional-dependencies.settings = [ ] optional-dependencies.spatialdata = [ "spatialdata" ] optional-dependencies.sphinx = [ "pydocstring-rs>=0.1.13", "sphinx>=9" ] # https://docs.pypi.org/project_metadata/#project-urls diff --git a/src/scverse_misc/__init__.py b/src/scverse_misc/__init__.py index baeedc8..141e9c5 100644 --- a/src/scverse_misc/__init__.py +++ b/src/scverse_misc/__init__.py @@ -1,11 +1,12 @@ -from contextlib import suppress - from ._deprecated import Deprecation, deprecated, deprecated_arg from ._extensions import ExtensionNamespace, make_register_namespace_decorator - -__all__ = ["ExtensionNamespace", "make_register_namespace_decorator", "deprecated", "deprecated_arg", "Deprecation"] - -with suppress(ImportError): - from ._settings import Settings # noqa: F401 - - __all__.append("Settings") +from ._settings import Settings + +__all__ = [ + "ExtensionNamespace", + "make_register_namespace_decorator", + "deprecated", + "deprecated_arg", + "Deprecation", + "Settings", +] diff --git a/src/scverse_misc/logging.py b/src/scverse_misc/logging.py index 1b9ef7b..91901f9 100644 --- a/src/scverse_misc/logging.py +++ b/src/scverse_misc/logging.py @@ -28,7 +28,11 @@ import logging from datetime import datetime, timedelta -from typing import Any, Literal, overload +from typing import Any, Literal, Self, cast, overload + +from pydantic import field_validator, model_validator + +from ._settings import Settings __all__ = ["Rule", "Elapsed", "Deep", "TimedLogger", "config", "get_logger"] @@ -105,59 +109,72 @@ def _rich_available() -> bool: return find_spec("rich") is not None -class _Config: - """Central logging configuration; the singleton instance is :data:`config`.""" - - def __init__(self) -> None: - self._root = logging.getLogger(_ROOT) - self._root.setLevel(logging.WARNING) - self._root.propagate = False # one handler here; don't double-log via root - self._rich: bool | None = None # None = auto-detect rich - self._rules: list[Rule] = [Elapsed(), Deep()] # universal defaults; order matters - self._install(_make_handler(self._use_rich())) +class _LogConfig(Settings): + """Central logging configuration; the singleton instance is :data:`config`. - def _install(self, handler: logging.Handler) -> None: - for r in self._rules: - handler.addFilter(r) - self._root.addHandler(handler) + Subclasses :class:`~scverse_misc.Settings`, so values also load from + environment variables (prefix ``SCVERSE_MISC_``) and support the inherited + :meth:`override`/:meth:`reset` context managers. The shared ``scverse`` + logger is the source of truth for the live state; assigning a field + re-applies it via the validator below. + """ - def _use_rich(self) -> bool: - return _rich_available() if self._rich is None else self._rich + verbosity: str | int = "warning" + """Central level for all scverse loggers; a level name (``"info"``) or an int.""" + + rich: bool | None = None + """Force rich rendering on/off; ``None`` auto-detects whether rich is installed.""" + + @field_validator("verbosity") + @classmethod + def _canonical_level(cls, value: str | int) -> str: + """Validate and normalize to a canonical level name (e.g. ``"WARNING"``).""" + if isinstance(value, str): + if not isinstance(logging.getLevelName(value.upper()), int): + raise ValueError(f"unknown log level name: {value!r}") + return value.upper() + name = logging.getLevelName(value) + if name.startswith("Level "): + raise ValueError(f"unknown log level: {value!r}") + return name + + @model_validator(mode="after") + def _apply(self) -> Self: + """Push the current settings onto the shared ``scverse`` logger and handler.""" + root = logging.getLogger(_ROOT) + root.propagate = False # one handler here; don't double-log via root + root.setLevel(self.verbosity) + use_rich = _rich_available() if self.rich is None else self.rich + current = root.handlers[0] if root.handlers else None + # a plain handler is a StreamHandler, rich's RichHandler is not -> cheap rich test + if current is None or isinstance(current, logging.StreamHandler) == use_rich: + rules = list(current.filters) if current else [Elapsed(), Deep()] # carry rules across + for h in list(root.handlers): + root.removeHandler(h) + handler = _make_handler(use_rich) + for r in rules: + handler.addFilter(r) + root.addHandler(handler) + return self @property - def verbosity(self) -> str | int: - """Central level for all scverse loggers. Set with a name (``"info"``) or int.""" - return logging.getLevelName(self._root.level) - - @verbosity.setter - def verbosity(self, level: str | int) -> None: - self._root.setLevel(level.upper() if isinstance(level, str) else level) + def _root(self) -> logging.Logger: + return logging.getLogger(_ROOT) @property - def rich(self) -> bool: - """Whether rich rendering is active. Set ``True``/``False`` to force, ``None`` to auto-detect.""" - return self._use_rich() - - @rich.setter - def rich(self, enabled: bool | None) -> None: - self._rich = enabled - for h in list(self._root.handlers): - self._root.removeHandler(h) - self._install(_make_handler(self._use_rich())) + def _rules(self) -> list[logging.Filter]: + return cast("list[logging.Filter]", self._root.handlers[0].filters) def add_rule(self, rule: Rule) -> None: - self._rules.append(rule) for h in self._root.handlers: h.addFilter(rule) def remove_rule(self, rule: Rule) -> None: - if rule in self._rules: - self._rules.remove(rule) for h in self._root.handlers: h.removeFilter(rule) -config = _Config() +config = _LogConfig() class TimedLogger: diff --git a/tests/test_logging.py b/tests/test_logging.py index a69276a..d3959c0 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING import pytest +from pydantic import ValidationError from scverse_misc import logging as mod from scverse_misc.logging import Deep, Elapsed, Rule, TimedLogger, config, get_logger @@ -131,6 +132,28 @@ def test_verbosity_get_set_by_name_and_int() -> None: assert plain.isEnabledFor(logging.DEBUG) +def test_verbosity_rejects_unknown_level() -> None: + config.verbosity = "warning" + for bad in ("bogus", 999): + with pytest.raises(ValidationError): + config.verbosity = bad + assert config.verbosity == "WARNING" # rejected assignment leaves the value untouched + + +def test_override_restores_verbosity() -> None: + config.verbosity = "warning" + with config.override(verbosity="debug"): + assert config.verbosity == "DEBUG" + assert logging.getLogger("scverse").level == logging.DEBUG + assert config.verbosity == "WARNING" + + +def test_env_var_sets_verbosity(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SCVERSE_MISC_VERBOSITY", "info") + assert mod._LogConfig().verbosity == "INFO" + config.verbosity = "warning" # construction set the shared logger; restore it + + def test_universal_rules_enabled_by_default() -> None: assert any(isinstance(r, Elapsed) for r in config._rules) assert any(isinstance(r, Deep) for r in config._rules) From 943fd7d1e0ad131cc4ebbc7c8555ab66eb474175 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Fri, 19 Jun 2026 19:18:51 +0000 Subject: [PATCH 8/8] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 1c56189..b23be23 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,7 +33,7 @@ optional-dependencies.datasets = [ "pooch", "pyyaml", "tqdm" ] optional-dependencies.rich = [ "rich" ] # Kept for backwards compat: anndata <0.13 references scverse-misc[settings]. # The deps are now hard dependencies (see above), so this extra is empty. -optional-dependencies.settings = [ ] +optional-dependencies.settings = [] optional-dependencies.spatialdata = [ "spatialdata" ] optional-dependencies.sphinx = [ "pydocstring-rs>=0.1.13", "sphinx>=9" ] # https://docs.pypi.org/project_metadata/#project-urls