Logging#

Logger configuration and structured log helpers used throughout the library. CustomFormatter adds colour and aligned level labels; configure_logger_level_and_handlers is called by Cohort.__init__ to apply the logger_args parameter.

class corr_vars.utils.logging.CustomFormatter(colored_output=None, formatted_numbers=True, fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)[source]#

Bases: Formatter

Logging formatter that optionally colorizes output and highlights numbers.

Features: - Optionally colorizes level names and certain keywords (e.g. SUCCESS, DROP). - Optionally underlines numeric tokens in messages for emphasis. - Respects NO_COLOR environment variable when deciding to colorize.

Parameters:
  • colored_output (bool | None)

  • formatted_numbers (bool)

  • fmt (str | None)

  • datefmt (str | None)

  • style (Literal['%', '{', '$'])

  • validate (bool)

  • defaults (Mapping[str, Any] | None)

white = '\x1b[37m'#
green = '\x1b[32m'#
yellow = '\x1b[33m'#
red = '\x1b[31m'#
bold_red = '\x1b[31;1m'#
reset = '\x1b[0m'#
underline = '\x1b[4m'#
LEVEL_COLORS = {'CRITICAL': '\x1b[31;1m', 'DEBUG': '\x1b[37m', 'ERROR': '\x1b[31m', 'INFO': '\x1b[37m', 'WARNING': '\x1b[33m'}#
format(record)[source]#

Format a LogRecord.

Performs the following extra steps before delegating to the base formatter:

  • Optionally underline numeric tokens in record.msg for emphasis.

  • Optionally colorize the level name and highlight certain keywords (e.g. ‘SUCCESS’ in green, ‘DROP’ in red).

Parameters:

record (LogRecord) – logging.LogRecord to be formatted.

Return type:

str

Returns:

The formatted log message string.

corr_vars.utils.logging.configure_logger_level_and_handlers(logger, level=20, file_path=None, file_mode='a', verbose_fmt=False, colored_output=None, formatted_numbers=True)[source]#

Configure logger level and attach standard handlers.

This helper: - Sets the logger level. - Removes existing handlers. - Adds a StreamHandler using CustomFormatter. - Optionally adds a FileHandler when file_path is provided.

Parameters:
  • logger (Logger) – Logger instance to configure.

  • level (int | str) – Logging level (int or string).

  • file_path (str | None) – Optional filesystem path to write logs to.

  • file_mode (str) – File mode for the file handler (default ‘a’).

  • verbose_fmt (bool) – If True, use a verbose formatter including timestamps and source file.

  • colored_output (bool | None) – If True/False forces colored output for the stream handler.

  • formatted_numbers (bool) – If True, numbers in messages are underlined.

Return type:

None

corr_vars.utils.logging.text_indent(text, indent=0)[source]#

Indent a block of text.

Parameters:
  • text (str | Collection[str]) – The input text block or collection of lines to indent.

  • indent (str | int) – String or number of spaces to use for indentation.

Return type:

str

Returns:

The indented text block.

corr_vars.utils.logging.text_tree_indent(text)[source]#

Indent a block of text with tree-style prefixes.

Each line is prefixed with ‘├── ‘ except the last line, which is prefixed with ‘└── ‘.

Parameters:

text (str | Collection[str]) – The input text block or collection of lines to indent.

Return type:

str

Returns:

The tree-indented text block.

corr_vars.utils.logging.log_collection(logger, collection, level=20, as_tree=True, indent=0, serialiser=<class 'str'>)[source]#

Log elements of a collection in a readable form.

When as_tree=True each element is prefixed with tree-style markers:

├── for all but the last item └── for the last item

When as_tree=False each item is indented uniformly without tree markers.

Parameters:
  • logger (Logger) – Logger to emit the messages.

  • collection (Collection[TypeVar(T)]) – Iterable collection of items to log.

  • level (int) – Logging level to use for each emitted message.

  • as_tree (bool) – Whether to render the collection with tree markers.

  • indent (str | int) – String or number of spaces to use for indentation when as_tree=False.

  • serialiser (Callable[[TypeVar(T)], str]) – Function to convert each item to a string.

Return type:

None

corr_vars.utils.logging.log_multiline_string(logger, multiline, level=20, indent=0)[source]#

Log a multiline string as separate lines.

Splits the input on newline characters and logs each line separately. Useful for showing multi-line messages in logs.

Parameters:
  • logger (Logger) – Logger to emit the messages.

  • multiline (str) – The multi-line string to log.

  • level (int) – Logging level to use for each emitted message.

  • indent (str | int)

Return type:

None

corr_vars.utils.logging.log_dict(logger, dictionary, level=20, json_indent=0, indent=0)[source]#

Log a dictionary in pretty-printed JSON format.

Parameters:
  • logger (Logger) – Logger to emit the messages.

  • dictionary (dict[Any, Any]) – The dictionary to log.

  • level (int) – Logging level to use for each emitted message.

  • json_indent (str | int)

  • indent (str | int)

Return type:

None

corr_vars.utils.logging.pretty_join(input, sep=', ', width=120, indent=0, sort=True, serialiser=<class 'str'>)[source]#

Join an iterable into a pretty-formatted, indented block.

Parameters:
  • input (Iterable[TypeVar(T)]) – Iterable of items to join.

  • sep (str) – Separator string to use between items.

  • width (int) – Maximum line width for wrapping.

  • indent (str | int) – String or number of spaces to indent the output.

  • sort (bool) – Whether to sort the input strings before joining.

  • serialiser (Callable[[TypeVar(T)], str]) – Function to convert each input item to a string.

Return type:

str

Returns:

A single string with the joined, wrapped, and indented content.