Module refinery.lib.environment

A common interface to all binary refinery configuration settings available via environment variables. This module is also host to the logging configuration.

Expand source code Browse git
"""
A common interface to all binary refinery configuration settings available via environment
variables. This module is also host to the logging configuration.
"""
from __future__ import annotations

import abc
import logging
import os

from enum import IntEnum
from pathlib import Path
from typing import Generic, TypeVar

_T = TypeVar('_T')

Logger = logging.Logger


class LogLevel(IntEnum):
    """
    An enumeration representing the current log level:
    """
    DETACHED = logging.CRITICAL + 100
    """
    This unit is not attached to a terminal but has been instantiated in
    code. This means that the only way to communicate problems is to throw
    an exception.
    """
    ALWAYS = logging.CRITICAL + 80
    """
    This log level is used to output messages that should appear always.
    """
    NONE = logging.CRITICAL + 50
    PROFILE = logging.CRITICAL + 10

    @classmethod
    def FromVerbosity(cls, verbosity: int):
        if verbosity < 0:
            return cls.DETACHED
        return {
            0: cls.WARNING,
            1: cls.INFO,
            2: cls.DEBUG
        }.get(verbosity, cls.DEBUG)

    NOTSET   = logging.NOTSET    # noqa
    CRITICAL = logging.CRITICAL  # noqa
    FATAL    = logging.FATAL     # noqa
    ERROR    = logging.ERROR     # noqa
    WARNING  = logging.WARNING   # noqa
    WARN     = logging.WARN      # noqa
    INFO     = logging.INFO      # noqa
    DEBUG    = logging.DEBUG     # noqa

    @property
    def verbosity(self) -> int:
        if self.value >= LogLevel.DETACHED:
            return -1
        if self.value >= LogLevel.WARNING:
            return +0
        if self.value >= LogLevel.INFO:
            return +1
        if self.value >= LogLevel.DEBUG:
            return +2
        else:
            return -1


class RefineryFormatter(logging.Formatter):
    """
    The binary refinery log formatter class.
    """

    NAMES = {
        logging.CRITICAL : 'failure',
        logging.ERROR    : 'failure',
        logging.WARNING  : 'warning',
        logging.INFO     : 'comment',
        logging.DEBUG    : 'verbose',
        LogLevel.PROFILE : 'profile',
        LogLevel.ALWAYS  : 'message',
    }

    def __init__(self, format, **kwargs):
        super().__init__(format, **kwargs)

    def formatMessage(self, record: logging.LogRecord) -> str:
        record.custom_level_name = self.NAMES[record.levelno]
        return super().formatMessage(record)


def logger(name: str) -> logging.Logger:
    """
    Obtain a logger which is configured with the default refinery format.
    """
    logger = logging.getLogger(name)
    if not logger.hasHandlers():
        stream = logging.StreamHandler()
        stream.setFormatter(RefineryFormatter(
            '({asctime}) {custom_level_name} in {name}: {message}',
            style='{',
            datefmt='%H:%M:%S',
        ))
        logger.addHandler(stream)
    logger.propagate = False
    return logger


class EnvironmentVariableSetting(abc.ABC, Generic[_T]):
    """
    Abstraction of an environment variable based setting.
    """
    key: str
    value: _T | None

    def __init__(self, name: str):
        key = F'REFINERY_{name}'
        try:
            value = os.environ[key]
        except KeyError:
            value = None
        else:
            value = self.read(value)
        self.key = key
        self.value = value

    @abc.abstractmethod
    def read(self, value: str) -> _T | None:
        return None


class EVBool(EnvironmentVariableSetting[bool]):
    """
    Boolean setting stored in an environment variable.
    """
    def read(self, value: str):
        if value := value.lower().strip():
            if value.isdigit():
                return bool(int(value))
            elif value in {'no', 'off', 'false', 'disable', 'disabled'}:
                return False
            elif value in {'yes', 'on', 'true', 'enable', 'enabled', 'active'}:
                return True
            else:
                return None
        else:
            return False


class EVPath(EnvironmentVariableSetting[Path]):
    """
    A system path stored in an environment variable.
    """
    def read(self, value: str):
        return Path(value)


class EVInt(EnvironmentVariableSetting[int]):
    """
    An integer value stored in an environment variable.
    """
    def read(self, value: str):
        try:
            return int(value, 0)
        except (KeyError, ValueError):
            return None


class EVLog(EnvironmentVariableSetting[LogLevel]):
    """
    A log level stored in an environment variable. This can be specified as either the name of the
    log level or its integer value.
    """
    def read(self, value: str):
        if value.isdigit():
            return LogLevel.FromVerbosity(int(value))
        try:
            return LogLevel[value]
        except KeyError:
            levels = ', '.join(ll.name for ll in LogLevel)
            logger(__name__).warning(
                F'ignoring unknown verbosity "{value!r}"; pick from: {levels}')
            return None


class environment:
    """
    Provides access to refinery-related configuration in environment variables.
    """
    verbosity = EVLog('VERBOSITY')
    term_size = EVInt('TERM_SIZE')
    colorless = EVBool('COLORLESS')
    storepath = EVPath('STOREPATH')
    disable_size_format = EVBool('DISABLE_SIZE_FORMAT')
    silence_ps1_warning = EVBool('SILENCE_PS1_WARNING')
    disable_ps1_bandaid = EVBool('DISABLE_PS1_BANDAID')

Functions

def logger(name)

Obtain a logger which is configured with the default refinery format.

Expand source code Browse git
def logger(name: str) -> logging.Logger:
    """
    Obtain a logger which is configured with the default refinery format.
    """
    logger = logging.getLogger(name)
    if not logger.hasHandlers():
        stream = logging.StreamHandler()
        stream.setFormatter(RefineryFormatter(
            '({asctime}) {custom_level_name} in {name}: {message}',
            style='{',
            datefmt='%H:%M:%S',
        ))
        logger.addHandler(stream)
    logger.propagate = False
    return logger

Classes

class LogLevel (*args, **kwds)

An enumeration representing the current log level:

Expand source code Browse git
class LogLevel(IntEnum):
    """
    An enumeration representing the current log level:
    """
    DETACHED = logging.CRITICAL + 100
    """
    This unit is not attached to a terminal but has been instantiated in
    code. This means that the only way to communicate problems is to throw
    an exception.
    """
    ALWAYS = logging.CRITICAL + 80
    """
    This log level is used to output messages that should appear always.
    """
    NONE = logging.CRITICAL + 50
    PROFILE = logging.CRITICAL + 10

    @classmethod
    def FromVerbosity(cls, verbosity: int):
        if verbosity < 0:
            return cls.DETACHED
        return {
            0: cls.WARNING,
            1: cls.INFO,
            2: cls.DEBUG
        }.get(verbosity, cls.DEBUG)

    NOTSET   = logging.NOTSET    # noqa
    CRITICAL = logging.CRITICAL  # noqa
    FATAL    = logging.FATAL     # noqa
    ERROR    = logging.ERROR     # noqa
    WARNING  = logging.WARNING   # noqa
    WARN     = logging.WARN      # noqa
    INFO     = logging.INFO      # noqa
    DEBUG    = logging.DEBUG     # noqa

    @property
    def verbosity(self) -> int:
        if self.value >= LogLevel.DETACHED:
            return -1
        if self.value >= LogLevel.WARNING:
            return +0
        if self.value >= LogLevel.INFO:
            return +1
        if self.value >= LogLevel.DEBUG:
            return +2
        else:
            return -1

Ancestors

  • enum.IntEnum
  • builtins.int
  • enum.ReprEnum
  • enum.Enum

Class variables

var DETACHED

This unit is not attached to a terminal but has been instantiated in code. This means that the only way to communicate problems is to throw an exception.

var ALWAYS

This log level is used to output messages that should appear always.

var NONE

The type of the None singleton.

var PROFILE

The type of the None singleton.

var NOTSET

The type of the None singleton.

var CRITICAL

The type of the None singleton.

var FATAL

The type of the None singleton.

var ERROR

The type of the None singleton.

var WARNING

The type of the None singleton.

var WARN

The type of the None singleton.

var INFO

The type of the None singleton.

var DEBUG

The type of the None singleton.

Static methods

def FromVerbosity(verbosity)

Instance variables

var verbosity
Expand source code Browse git
@property
def verbosity(self) -> int:
    if self.value >= LogLevel.DETACHED:
        return -1
    if self.value >= LogLevel.WARNING:
        return +0
    if self.value >= LogLevel.INFO:
        return +1
    if self.value >= LogLevel.DEBUG:
        return +2
    else:
        return -1
class RefineryFormatter (format, **kwargs)

The binary refinery log formatter class.

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of '%', '{' or '$' to specify that you want to use one of %-formatting, :meth:str.format ({}) formatting or :class:string.Template formatting in your format string.

Changed in version: 3.2

Added the style parameter.

Expand source code Browse git
class RefineryFormatter(logging.Formatter):
    """
    The binary refinery log formatter class.
    """

    NAMES = {
        logging.CRITICAL : 'failure',
        logging.ERROR    : 'failure',
        logging.WARNING  : 'warning',
        logging.INFO     : 'comment',
        logging.DEBUG    : 'verbose',
        LogLevel.PROFILE : 'profile',
        LogLevel.ALWAYS  : 'message',
    }

    def __init__(self, format, **kwargs):
        super().__init__(format, **kwargs)

    def formatMessage(self, record: logging.LogRecord) -> str:
        record.custom_level_name = self.NAMES[record.levelno]
        return super().formatMessage(record)

Ancestors

  • logging.Formatter

Class variables

var NAMES

The type of the None singleton.

Methods

def formatMessage(self, record)
Expand source code Browse git
def formatMessage(self, record: logging.LogRecord) -> str:
    record.custom_level_name = self.NAMES[record.levelno]
    return super().formatMessage(record)
class EnvironmentVariableSetting (name)

Abstraction of an environment variable based setting.

Expand source code Browse git
class EnvironmentVariableSetting(abc.ABC, Generic[_T]):
    """
    Abstraction of an environment variable based setting.
    """
    key: str
    value: _T | None

    def __init__(self, name: str):
        key = F'REFINERY_{name}'
        try:
            value = os.environ[key]
        except KeyError:
            value = None
        else:
            value = self.read(value)
        self.key = key
        self.value = value

    @abc.abstractmethod
    def read(self, value: str) -> _T | None:
        return None

Ancestors

  • abc.ABC
  • typing.Generic

Subclasses

Class variables

var key

The type of the None singleton.

var value

The type of the None singleton.

Methods

def read(self, value)
Expand source code Browse git
@abc.abstractmethod
def read(self, value: str) -> _T | None:
    return None
class EVBool (name)

Boolean setting stored in an environment variable.

Expand source code Browse git
class EVBool(EnvironmentVariableSetting[bool]):
    """
    Boolean setting stored in an environment variable.
    """
    def read(self, value: str):
        if value := value.lower().strip():
            if value.isdigit():
                return bool(int(value))
            elif value in {'no', 'off', 'false', 'disable', 'disabled'}:
                return False
            elif value in {'yes', 'on', 'true', 'enable', 'enabled', 'active'}:
                return True
            else:
                return None
        else:
            return False

Ancestors

Methods

def read(self, value)
Expand source code Browse git
def read(self, value: str):
    if value := value.lower().strip():
        if value.isdigit():
            return bool(int(value))
        elif value in {'no', 'off', 'false', 'disable', 'disabled'}:
            return False
        elif value in {'yes', 'on', 'true', 'enable', 'enabled', 'active'}:
            return True
        else:
            return None
    else:
        return False

Inherited members

class EVPath (name)

A system path stored in an environment variable.

Expand source code Browse git
class EVPath(EnvironmentVariableSetting[Path]):
    """
    A system path stored in an environment variable.
    """
    def read(self, value: str):
        return Path(value)

Ancestors

Methods

def read(self, value)
Expand source code Browse git
def read(self, value: str):
    return Path(value)

Inherited members

class EVInt (name)

An integer value stored in an environment variable.

Expand source code Browse git
class EVInt(EnvironmentVariableSetting[int]):
    """
    An integer value stored in an environment variable.
    """
    def read(self, value: str):
        try:
            return int(value, 0)
        except (KeyError, ValueError):
            return None

Ancestors

Methods

def read(self, value)
Expand source code Browse git
def read(self, value: str):
    try:
        return int(value, 0)
    except (KeyError, ValueError):
        return None

Inherited members

class EVLog (name)

A log level stored in an environment variable. This can be specified as either the name of the log level or its integer value.

Expand source code Browse git
class EVLog(EnvironmentVariableSetting[LogLevel]):
    """
    A log level stored in an environment variable. This can be specified as either the name of the
    log level or its integer value.
    """
    def read(self, value: str):
        if value.isdigit():
            return LogLevel.FromVerbosity(int(value))
        try:
            return LogLevel[value]
        except KeyError:
            levels = ', '.join(ll.name for ll in LogLevel)
            logger(__name__).warning(
                F'ignoring unknown verbosity "{value!r}"; pick from: {levels}')
            return None

Ancestors

Methods

def read(self, value)
Expand source code Browse git
def read(self, value: str):
    if value.isdigit():
        return LogLevel.FromVerbosity(int(value))
    try:
        return LogLevel[value]
    except KeyError:
        levels = ', '.join(ll.name for ll in LogLevel)
        logger(__name__).warning(
            F'ignoring unknown verbosity "{value!r}"; pick from: {levels}')
        return None

Inherited members

class environment

Provides access to refinery-related configuration in environment variables.

Expand source code Browse git
class environment:
    """
    Provides access to refinery-related configuration in environment variables.
    """
    verbosity = EVLog('VERBOSITY')
    term_size = EVInt('TERM_SIZE')
    colorless = EVBool('COLORLESS')
    storepath = EVPath('STOREPATH')
    disable_size_format = EVBool('DISABLE_SIZE_FORMAT')
    silence_ps1_warning = EVBool('SILENCE_PS1_WARNING')
    disable_ps1_bandaid = EVBool('DISABLE_PS1_BANDAID')

Class variables

var verbosity

The type of the None singleton.

var term_size

The type of the None singleton.

var colorless

The type of the None singleton.

var storepath

The type of the None singleton.

var disable_size_format

The type of the None singleton.

var silence_ps1_warning

The type of the None singleton.

var disable_ps1_bandaid

The type of the None singleton.