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.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
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 os
import logging

from enum import IntEnum
from typing import Optional, TypeVar, Generic

_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.
    """
    NONE = logging.CRITICAL + 50

    @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):

    NAMES = {
        logging.CRITICAL : 'failure',
        logging.ERROR    : 'failure',
        logging.WARNING  : 'warning',
        logging.INFO     : 'comment',
        logging.DEBUG    : 'verbose',
    }

    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(Generic[_T]):
    key: str
    value: Optional[_T]

    def __init__(self, name: str):
        self.key = F'REFINERY_{name}'
        self.value = self.read()

    def read(self) -> _T:
        return None


class EVBool(EnvironmentVariableSetting[bool]):
    def read(self):
        value = os.environ.get(self.key, None)
        if value is None:
            return False
        else:
            value = value.lower().strip()
        if not value:
            return False
        if value.isdigit():
            return bool(int(value))
        return value not in {'no', 'off', 'false'}


class EVInt(EnvironmentVariableSetting[int]):
    def read(self):
        try:
            return int(os.environ[self.key], 0)
        except (KeyError, ValueError):
            return 0


class EVLog(EnvironmentVariableSetting[Optional[LogLevel]]):
    def read(self):
        try:
            loglevel = os.environ[self.key]
        except KeyError:
            return None
        if loglevel.isdigit():
            return LogLevel.FromVerbosity(int(loglevel))
        try:
            loglevel = LogLevel[loglevel]
        except KeyError:
            levels = ', '.join(ll.name for ll in LogLevel)
            logger(__name__).warning(
                F'ignoring unknown verbosity "{loglevel!r}"; pick from: {levels}')
            return None
        else:
            return loglevel


class environment:
    verbosity = EVLog('VERBOSITY')
    term_size = EVInt('TERM_SIZE')
    colorless = EVBool('COLORLESS')
    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 (value, names=None, *, module=None, qualname=None, type=None, start=1)

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.
    """
    NONE = logging.CRITICAL + 50

    @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.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 NONE

var NOTSET

var CRITICAL

var FATAL

var ERROR

var WARNING

var WARN

var INFO

var DEBUG

Static methods

def FromVerbosity(verbosity)

Expand source code Browse git

@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)

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)

Formatter instances are used to convert a LogRecord to text.

Formatters need to know how a LogRecord is constructed. They are responsible for converting a LogRecord to (usually) a string which can be interpreted by either a human or an external system. The base Formatter allows a formatting string to be specified. If none is supplied, the the style-dependent default value, "%(message)s", "{message}", or "${message}", is used.

The Formatter can be initialized with a format string which makes use of knowledge of the LogRecord attributes - e.g. the default value mentioned above makes use of the fact that the user's message and arguments are pre- formatted into a LogRecord's message attribute. Currently, the useful attributes in a LogRecord are described by:

%(name)s Name of the logger (logging channel) %(levelno)s Numeric logging level for the message (DEBUG, INFO, WARNING, ERROR, CRITICAL) %(levelname)s Text logging level for the message ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") %(pathname)s Full pathname of the source file where the logging call was issued (if available) %(filename)s Filename portion of pathname %(module)s Module (name portion of filename) %(lineno)d Source line number where the logging call was issued (if available) %(funcName)s Function name %(created)f Time when the LogRecord was created (time.time() return value) %(asctime)s Textual time when the LogRecord was created %(msecs)d Millisecond portion of the creation time %(relativeCreated)d Time in milliseconds when the LogRecord was created, relative to the time the logging module was loaded (typically at application startup time) %(thread)d Thread ID (if available) %(threadName)s Thread name (if available) %(process)d Process ID (if available) %(message)s The result of record.getMessage(), computed just as the record is emitted

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):

    NAMES = {
        logging.CRITICAL : 'failure',
        logging.ERROR    : 'failure',
        logging.WARNING  : 'warning',
        logging.INFO     : 'comment',
        logging.DEBUG    : 'verbose',
    }

    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

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)

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::

class Mapping(Generic[KT, VT]): def getitem(self, key: KT) -> VT: … # Etc.

This class can then be used as follows::

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default

Expand source code Browse git

class EnvironmentVariableSetting(Generic[_T]):
    key: str
    value: Optional[_T]

    def __init__(self, name: str):
        self.key = F'REFINERY_{name}'
        self.value = self.read()

    def read(self) -> _T:
        return None

Ancestors

• typing.Generic

Subclasses

• EVBool
• EVInt
• EVLog

Class variables

var key

var value

Methods

def read(self)

Expand source code Browse git

def read(self) -> _T:
    return None

class EVBool (name)

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::

class Mapping(Generic[KT, VT]): def getitem(self, key: KT) -> VT: … # Etc.

This class can then be used as follows::

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default

Expand source code Browse git

class EVBool(EnvironmentVariableSetting[bool]):
    def read(self):
        value = os.environ.get(self.key, None)
        if value is None:
            return False
        else:
            value = value.lower().strip()
        if not value:
            return False
        if value.isdigit():
            return bool(int(value))
        return value not in {'no', 'off', 'false'}

Ancestors

• EnvironmentVariableSetting
• typing.Generic

Class variables

var key

var value

Methods

def read(self)

Expand source code Browse git

def read(self):
    value = os.environ.get(self.key, None)
    if value is None:
        return False
    else:
        value = value.lower().strip()
    if not value:
        return False
    if value.isdigit():
        return bool(int(value))
    return value not in {'no', 'off', 'false'}

class EVInt (name)

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::

class Mapping(Generic[KT, VT]): def getitem(self, key: KT) -> VT: … # Etc.

This class can then be used as follows::

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default

Expand source code Browse git

class EVInt(EnvironmentVariableSetting[int]):
    def read(self):
        try:
            return int(os.environ[self.key], 0)
        except (KeyError, ValueError):
            return 0

Ancestors

• EnvironmentVariableSetting
• typing.Generic

Class variables

var key

var value

Methods

def read(self)

Expand source code Browse git

def read(self):
    try:
        return int(os.environ[self.key], 0)
    except (KeyError, ValueError):
        return 0

class EVLog (name)

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as::

class Mapping(Generic[KT, VT]): def getitem(self, key: KT) -> VT: … # Etc.

This class can then be used as follows::

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default

Expand source code Browse git

class EVLog(EnvironmentVariableSetting[Optional[LogLevel]]):
    def read(self):
        try:
            loglevel = os.environ[self.key]
        except KeyError:
            return None
        if loglevel.isdigit():
            return LogLevel.FromVerbosity(int(loglevel))
        try:
            loglevel = LogLevel[loglevel]
        except KeyError:
            levels = ', '.join(ll.name for ll in LogLevel)
            logger(__name__).warning(
                F'ignoring unknown verbosity "{loglevel!r}"; pick from: {levels}')
            return None
        else:
            return loglevel

Ancestors

• EnvironmentVariableSetting
• typing.Generic

Class variables

var key

var value

Methods

def read(self)

Expand source code Browse git

def read(self):
    try:
        loglevel = os.environ[self.key]
    except KeyError:
        return None
    if loglevel.isdigit():
        return LogLevel.FromVerbosity(int(loglevel))
    try:
        loglevel = LogLevel[loglevel]
    except KeyError:
        levels = ', '.join(ll.name for ll in LogLevel)
        logger(__name__).warning(
            F'ignoring unknown verbosity "{loglevel!r}"; pick from: {levels}')
        return None
    else:
        return loglevel

class environment

Expand source code Browse git

class environment:
    verbosity = EVLog('VERBOSITY')
    term_size = EVInt('TERM_SIZE')
    colorless = EVBool('COLORLESS')
    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

var term_size

var colorless

var disable_size_format

var silence_ps1_warning

var disable_ps1_bandaid