Utility¶

livy.utils is a collection of helper classes to make the CLI tool more friendly, this module is not included in livy import by default.

Most of the classes in this module need extra dependency to work. If you call them in a non-compatible environment, it would automatically fallback to some builtin python function with the similar features.

livy.utils.ConfigBase¶

Config utility provides an interface to get typed config values and could store/read them into an unified file. By default, it reads settings from ~/.config/python-livy.json. Then fallback to default value hardcoded in the code if the key not exists.

If you are going to repack this tool to suit your environment, it is suggested to put values in default-configuration.json in the root directory of this repo. It could be easier to maintain the configuration rather than change every thing in the code.

class livy.utils.configbase.ConfigBase¶

Base class for setting configurations. Inspired by pydantic. Thought this, please just treat this class as a simplified version of dataclasses.

This class does lake of validation, it is created to perform dictionary updating liked config merge and json transform.

To use this base, inherit this class and use type annotation to define the fields. Fields would be automatically created during __init__().

__init__(**kwargs)¶

Parameters: **kwargs – Field name and values. It would fill with default value defined in annotations if not specific, or set to None when default is not set.
Return type: None

classmethod load(section)¶

Create a config instance and load settings from files. It reads the config from CONFIG_LOAD_ORDER.

Parameters: section (str) – Key name for reading config
Returns: config – Config instance with data loaded from the file.
Return type: ConfigBase

mergedict(data)¶

Merge configs. Overwrite all the value in this instance from a dict.

Parameters: data (dict) – A dict to provide values.
Return type: None

livy.utils.EnhancedConsoleHandler¶

class livy.utils.logging.EnhancedConsoleHandler¶

A stream handler that could shows progress bar on task set related logs found.

It would create a progress bar using tqdm. However, frequently suppressing the progress bar (for printing logs) would make the screen flashing. To deal with this issue, this handler uses a producer-consumer architecture inside. Logs are not emitted in real time, they would be proceed by batch in flush(). And a backend worker is created to regularly flushing the logs.

This implementation comes with a pitfall: if the program exits too early, some of the logs would be dropped. PR is welcome if you could resolve this issue.

__init__(stream)¶

Parameters: stream (typing.TextIO) – Output stream. Might be stdout or stderr.
Return type: None

static __new__(cls, stream)¶

Automatically fallback to normal handler if requirement not satisfied.

Parameters: stream (TextIO) –
Return type: logging.StreamHandler

close()¶: Close this handler. Stop emitting logs to console.

flush()¶

Flush all logs to console

Return type: None

handle(record)¶

Override logging.StreamHandler for reteriving the log before it is filtered.

Parameters: record (logging.LogRecord) –
Return type: None

livy.utils.ColoredFormatter¶

class livy.utils.logging.ColoredFormatter¶

A formatter that could add ANSI colors to logs. Inspired by python-colorlog, and add feature that supports different color scheme via logger name.

__init__(fmt, datefmt, highlight_loggers=None)¶

Parameters

fmt (str) – Format for log message
datefmt (str) – Format for datetime
highlight_loggers (List[str]) – List of logger name to be highlighted

Return type

None

static __new__(cls, fmt, datefmt, highlight_loggers=None)¶

Automatically fallback to normal formatter if requirement not satisfied.

Parameters

fmt (str) –
datefmt (str) –
highlight_loggers (Optional[Iterable[str]]) –

Return type

logging.StreamHandler

formatMessage(record)¶

Override formatMessage to add color

Parameters: record (logging.LogRecord) –
Return type: str

get_color_map(record)¶

Get dict with color code to be updated into log record’s __dict__.

Parameters: record (logging.LogRecord) –
Return type: Dict[str, str]

livy.utils.IngoreLogFilter¶

class livy.utils.logging.IngoreLogFilter¶

Drop logs from the unwanted logger list.

__init__(unwanted_loggers)¶

Parameters: unwanted_loggers (List[str]) – List of logger name to be ignored
Return type: None

filter(record)¶

Determine if the specified record is to be logged.

Parameters: record (logging.LogRecord) – Log record
Returns: ok – Returns True if the record should be logged, or False otherwise.
Return type: bool

Core library Plugins