Skip to content

Logging

duho.logging

TRACE module-attribute

TRACEBACK_ENV = 'DUHO_TRACEBACK' module-attribute

VERBOSE_HELP = '' module-attribute

VERBOSE_LEVELS = {} module-attribute

__all__ = ['add_logging_level', 'DefaultFormatter', 'VERBOSE_LEVELS', 'VERBOSE_HELP', 'parse_loglevels', 'init_stderr_logging', 'initverbose', 'TRACEBACK_ENV', 'traceback_enabled', 'log_exception'] module-attribute

DefaultFormatter(fmt='%(asctime)s | %(levelname)8s | %(name)s: %(message)s', datefmt=None, style='%', validate=True)

Bases: Formatter

Log formatter with colored output.

Source code in src/duho/logging.py
109
110
111
112
113
114
115
116
117
def __init__(
    self,
    fmt="%(asctime)s | %(levelname)8s | %(name)s: %(message)s",
    datefmt=None,
    style: "_logging._FormatStyle" = "%",
    validate=True,
) -> None:
    self._levelsize: 'int | None' = None
    super().__init__(fmt, datefmt, style, validate)

COLORS = {_logging.DEBUG: _asicode(34), _logging.INFO: _asicode(32), _logging.WARNING: _asicode(33), _logging.ERROR: _asicode(31), _logging.CRITICAL: _asicode(31, 47)} class-attribute instance-attribute

RESET_ALL = _asicode(0) class-attribute instance-attribute

format(record)

Source code in src/duho/logging.py
119
120
121
122
123
124
125
126
def format(self, record):
    record = _copy.copy(record)
    levelsize = self._levelsize if self._levelsize is not None else _LEVELSIZE
    record.levelname = record.levelname.center(levelsize)
    color = self.COLORS.get(record.levelno, None)
    if color:
        record.levelname = f"{color}{record.levelname}{self.RESET_ALL}"
    return super().format(record)

__getattr__(name)

Source code in src/duho/logging.py
42
43
def __getattr__(name: str):
    return getattr(_logging, name)

add_logging_level(name, level, force=False, color=None)

Register a custom log level.

Source code in src/duho/logging.py
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def add_logging_level(name: str, level: int, force=False, color: 'str | None' = None):
    """Register a custom log level."""
    name = name.upper()
    if hasattr(_logging, name) and not force:
        return
    setattr(_logging, name, level)
    _logging.addLevelName(level, name)

    def log_logger(self: _logging.Logger, message: str, *args, **kwargs):
        if self.isEnabledFor(level):
            self._log(level, message, args, **kwargs)

    name = name.lower()
    setattr(_logging.getLoggerClass(), name, log_logger)

    def log_root(msg, *args, **kwargs):
        _logging.log(level, msg, *args, **kwargs)

    if color is not None:
        DefaultFormatter.COLORS[level] = _getcolor(color)

    setattr(_logging, name, log_root)

init_stderr_logging(name=None, level=None)

Initialize logging to stderr with color support.

Source code in src/duho/logging.py
171
172
173
174
175
176
177
178
179
180
def init_stderr_logging(name=None, level: 'int | None' = None):
    """Initialize logging to stderr with color support."""
    initverbose()
    handler = _logging.StreamHandler(_sys.stderr)
    logger = _logging.getLogger(name)
    if level:
        logger.setLevel(level)
    logger.addHandler(handler)
    handler.setFormatter(DefaultFormatter())
    return logger

initverbose()

Initialize verbose level mappings.

Source code in src/duho/logging.py
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def initverbose():
    """Initialize verbose level mappings."""
    global VERBOSE_LEVELS, VERBOSE_HELP, _LEVELSIZE

    for name, loglevel in get_level_names_mapping().items():
        if not loglevel:
            continue
        aliases: list[str] = VERBOSE_LEVELS.setdefault(loglevel, [])
        _LEVELSIZE = max(_LEVELSIZE, len(name))
        if name not in aliases:
            aliases.append(name)

    VERBOSE_LEVELS = dict(
        sorted(VERBOSE_LEVELS.items(), key=lambda l: l[0], reverse=True)
    )

    VERBOSE_HELP = ", ".join([aliases[0] for aliases in VERBOSE_LEVELS.values()])

log_exception(logger, msg, *args, level=_logging.ERROR)

Log a caught exception, with a traceback iff DUHO_TRACEBACK is set.

The framework's resilient paths (discovery skipping a bad command, a runpath step failing, a fan-out target raising) deliberately do NOT propagate the exception, which means the stack -- the only thing that says where it broke -- is lost unless it is logged. Logging it unconditionally would bury an ordinary "optional dependency missing" warning under 30 frames, so this helper makes it opt-in via :func:traceback_enabled.

Call it from inside an except block, where exc_info has an exception to render. When disabled the exc_info kwarg is omitted entirely rather than passed as False -- both suppress the traceback, but False is recorded verbatim on LogRecord.exc_info, so a handler or test inspecting that attribute would see False where every other un-decorated record in the process carries None.

Source code in src/duho/logging.py
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def log_exception(
    logger: "_logging.Logger",
    msg: str,
    *args: object,
    level: int = _logging.ERROR,
) -> None:
    """Log a caught exception, with a traceback iff ``DUHO_TRACEBACK`` is set.

    The framework's resilient paths (discovery skipping a bad command, a runpath
    step failing, a fan-out target raising) deliberately do NOT propagate the
    exception, which means the stack -- the only thing that says *where* it broke
    -- is lost unless it is logged. Logging it unconditionally would bury an
    ordinary "optional dependency missing" warning under 30 frames, so this
    helper makes it opt-in via :func:`traceback_enabled`.

    Call it from inside an ``except`` block, where ``exc_info`` has an exception
    to render. When disabled the ``exc_info`` kwarg is omitted entirely rather
    than passed as ``False`` -- both suppress the traceback, but ``False`` is
    recorded verbatim on ``LogRecord.exc_info``, so a handler or test inspecting
    that attribute would see ``False`` where every other un-decorated record in
    the process carries ``None``.
    """
    if traceback_enabled():
        logger.log(level, msg, *args, exc_info=True)
    else:
        logger.log(level, msg, *args)

parse_loglevels(text, itemdivider=',', valkey_separator=':')

Parse a log level specification string.

Source code in src/duho/logging.py
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def parse_loglevels(text: str, itemdivider: str = ",", valkey_separator=":"):
    """Parse a log level specification string."""
    levels: dict[str, int] = {}
    levelmapping = get_level_names_mapping()

    for entry in text.split(itemdivider):
        name, *level = entry.split(valkey_separator, maxsplit=1)
        if not level:
            level = name
            name = ""
        else:
            level = level[0]
        level = levelmapping.get(level)
        if level is not None:
            levels[name] = level
    return levels

traceback_enabled()

Return whether framework error logs should carry a full traceback.

Reads :data:TRACEBACK_ENV (DUHO_TRACEBACK) from the process environment on EVERY call rather than caching it, so a test (or an app that sets it mid-run) can flip the switch without re-importing duho. The read is a dict lookup -- cheap enough to sit on an error path.

Off by default: a CLI user seeing a framework warning wants the message, not a stack. A developer debugging where a step/command/target actually failed exports DUHO_TRACEBACK=1 and gets the traceback for free at every site.

Source code in src/duho/logging.py
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def traceback_enabled() -> bool:
    """Return whether framework error logs should carry a full traceback.

    Reads :data:`TRACEBACK_ENV` (``DUHO_TRACEBACK``) from the process
    environment on EVERY call rather than caching it, so a test (or an app that
    sets it mid-run) can flip the switch without re-importing duho. The read is
    a dict lookup -- cheap enough to sit on an error path.

    Off by default: a CLI user seeing a framework warning wants the message, not
    a stack. A developer debugging *where* a step/command/target actually failed
    exports ``DUHO_TRACEBACK=1`` and gets the traceback for free at every site.
    """
    import os as _os

    return _os.environ.get(TRACEBACK_ENV, "").strip().lower() not in _FALSEY