opmonlib.utils Namespace Reference

Classes
class	LoggingFormatter
class	LogLevelError

Functions
RichHandler	setup_rich_handler ()
str	logging_log_level_from_int (int level)
int	logging_log_level_from_str (str level)
dict[str:str]	parse_opmon_conf (logging.Logger log, dict[str:str]|"conffwk.dal.OpMonConf" conf, dict[str:str]|conffwk.dal.OpMonURI uri, str session, str application)
str	to_string (OpMonId opmon_id)
str	extract_opmon_file_path (str file_path, OpMonId|None origin=None)

Variables
dict	logging_log_levels
	logging_log_level_keys = list(logging_log_levels.keys())
	logging_log_level_values = list(logging_log_levels.values())
dict	oks_log_levels
	oks_log_level_keys = list(oks_log_levels.keys())
	oks_log_level_values = list(oks_log_levels.values())
dict	oks_to_logging_map
	log_level_keys = logging_log_level_keys + oks_log_level_keys
	log_level_values = logging_log_level_values + logging_log_level_values
tuple	e_log_levels
dict	CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]}
	CONSOLE_THEMES = Theme({"info": "dim cyan", "warning": "magenta", "danger": "bold red"})
str	full_log_format = "%(asctime)s %(levelname)s %(filename)s %(name)s %(message)s"
str	rich_log_format = "%(filename)s %(name)s %(message)s"
str	date_time_format = "[%Y/%m/%d %H:%M:%S]"
	time_zone = pytz.utc

Function Documentation

extract_opmon_file_path()

str opmonlib.utils.extract_opmon_file_path	(	str	file_path,
		OpMonId | None	origin = None )

Verify the file path can be opened.

Definition at line 270 of file utils.py.

def extract_opmon_file_path(file_path: str, origin: OpMonId | None = None) -> str:
    """Verify the file path can be opened."""
    hook = "://"
    hook_position = file_path.find(hook)
    fname = None
    if hook_position == -1:
        fname = file_path
    else:
        fname = file_path[hook_position + len(hook) :]
 
    if origin:
        slash_pos = fname.rfind("/")
        if slash_pos == -1:
            dot_pos = fname.find(".")
        else:
            dot_pos = fname.find(".", slash_pos)
        origin = to_string(origin)
        if dot_pos == -1:
            fname += "." + origin + ".json"
        else:
            fname = fname[:dot_pos] + "." + origin + fname[dot_pos:]
 
    try:
        with open(fname, "a"):
            pass
    except OSError:
        err_str = f"Can not open file {fname}"
        raise OSError(err_str) from None
 
    return fname

logging_log_level_from_int()

str opmonlib.utils.logging_log_level_from_int

(

int

level

)

Get the level name from its int value.

Definition at line 143 of file utils.py.

def logging_log_level_from_int(level: int) -> str:
    """Get the level name from its int value."""
    if not isinstance(level, int):
        return level
    for k, v in logging_log_levels.items():
        if v == level:
            return k
    for k, v in oks_log_levels.items():
        if v == level:
            return oks_to_logging_map[k]
    err_str = f"Requested log level with value {level} is not standard ({e_log_levels})"
    raise ValueError(err_str) from None
 
 

logging_log_level_from_str()

int opmonlib.utils.logging_log_level_from_str

(

str

level

)

Get the level int from its str value.

Definition at line 157 of file utils.py.

def logging_log_level_from_str(level: str) -> int:
    """Get the level int from its str value."""
    if not isinstance(level, str):
        return level
    for k, v in logging_log_levels.items():
        if k == level.upper():
            return v
    for k in oks_log_levels.keys():
        if k == level:
            return logging_log_levels[oks_to_logging_map[k]]
    err_str = f"Requested log level with value {level} is not standard ({e_log_levels})"
    raise ValueError(err_str) from None
 
 

parse_opmon_conf()

dict[str:str] opmonlib.utils.parse_opmon_conf	(	logging.Logger	log,
		dict[str:str] | "conffwk.dal.OpMonConf"	conf,
		dict[str:str] | conffwk.dal.OpMonURI	uri,
		str	session,
		str	application )

Parse the OpMonConf and OpMonURI.

Definition at line 171 of file utils.py.

) -> dict[str:str]:
    """Parse the OpMonConf and OpMonURI."""
    if not conf:
        log.error("Missing opmon configuration, exiting.")
        sys.exit(1)
    if not uri:
        log.error("Missing opmon URI, exiting.")
        sys.exit(1)
 
    opmon_type = (
        uri.get("type") if isinstance(uri, dict) else getattr(uri, "type", None)
    )
    if opmon_type:
        log.debug("Found OpMon type: %s", opmon_type)
    else:
        log.debug(
            "Missing 'type' in the opmon configuration, [yellow]using default value "
            "'stdout'[/yellow]."
        )
        opmon_type = "stdout"
 
    path = uri.get("path") if isinstance(uri, dict) else getattr(uri, "path", None)
    if path:
        log.debug("Found OpMon path: %s", path)
    elif opmon_type != "stdout":
        log.error("Missing 'path' in the opmon configuration, exiting.")
        sys.exit(1)
    else:
        if path == []:
            path = ""
        log.debug("No OpMon path required for type 'stdout'.")
 
    bootstrap = None
    topic = None
    if opmon_type == "file" and not Path(path).parent.is_dir():
        err_str = "Requested directory to put file in does not exist."
        raise ValueError(err_str) from None
    if "monkafka" in path:
        bootstrap, topic = path.split("/", 1)
    if not topic:
        topic = "opmon_stream"
    log.debug("Using OpMon topic: [green]'%s'[/green]", topic)
    log.debug("Using OpMon bootstrap: [green]'%s'[/green]", bootstrap)
 
    level = (
        conf.get("level") if isinstance(conf, dict) else getattr(conf, "level", None)
    )
    if level:
        log.debug("Found OpMon level: [green]%s[/green]", level)
    else:
        log.debug(
            "Missing 'level' in the OpMon configuration, [yellow]using default "
            "'DEBUG'[/yellow]."
        )
        level = logging.DEBUG
 
    interval_s = (
        conf.get("interval_s")
        if isinstance(conf, dict)
        else getattr(conf, "interval_s", None)
    )
    if interval_s:
        log.debug("Found OpMon interval_s: %s", interval_s)
    else:
        log.debug(
            "Missing 'interval_s' in the opmon configuration, [yellow]using default "
            "10s[/yellow]."
        )
        interval_s = 10.0
 
    return OpMonConf(
        opmon_type, bootstrap, topic, level, interval_s, path, session, application
    )
 
 

setup_rich_handler()

RichHandler opmonlib.utils.setup_rich_handler

(

)

Initialize a Rich handler for terminal logging.

Definition at line 125 of file utils.py.

def setup_rich_handler() -> RichHandler:
    """Initialize a Rich handler for terminal logging."""
    try:
        width = os.get_terminal_size()[0]
    except OSError:
        width = 150
    handler = RichHandler(
        console=Console(width=width),
        omit_repeated_times=False,
        markup=True,
        rich_tracebacks=True,
        show_path=False,
        tracebacks_width=width,
    )
    handler.setFormatter(LoggingFormatter(fmt=rich_log_format))
    return handler
 
 

to_string()

str opmonlib.utils.to_string

(

OpMonId

opmon_id

)

Map the OpMonId to a string.

Definition at line 252 of file utils.py.

def to_string(opmon_id: OpMonId) -> str:
    """Map the OpMonId to a string."""
    ret = opmon_id.get("session")
    if not ret:
        err_msg = "Missing session in OpMonId."
        raise ValueError(err_msg) from None
 
    application = opmon_id.get("application")
    if application:
        ret += "." + application
 
    substructures = opmon_id.get("substructure")
    for substructure in substructures:
        ret += "." + substructure
 
    return ret
 
 

Variable Documentation

CONSOLE_THEMES

opmonlib.utils.CONSOLE_THEMES = Theme({"info": "dim cyan", "warning": "magenta", "danger": "bold red"})

Definition at line 77 of file utils.py.

CONTEXT_SETTINGS

dict opmonlib.utils.CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]}

Definition at line 76 of file utils.py.

date_time_format

str opmonlib.utils.date_time_format = "[%Y/%m/%d %H:%M:%S]"

Definition at line 84 of file utils.py.

e_log_levels

tuple opmonlib.utils.e_log_levels

Initial value:

=  (
    f"{logging_log_level_keys} python logging or "
    f"{oks_log_level_keys} for oks log levels."
)

Definition at line 71 of file utils.py.

full_log_format

str opmonlib.utils.full_log_format = "%(asctime)s %(levelname)s %(filename)s %(name)s %(message)s"

Definition at line 80 of file utils.py.

log_level_keys

opmonlib.utils.log_level_keys = logging_log_level_keys + oks_log_level_keys

Definition at line 47 of file utils.py.

log_level_values

opmonlib.utils.log_level_values = logging_log_level_values + logging_log_level_values

Definition at line 48 of file utils.py.

logging_log_level_keys

opmonlib.utils.logging_log_level_keys = list(logging_log_levels.keys())

Definition at line 27 of file utils.py.

logging_log_level_values

opmonlib.utils.logging_log_level_values = list(logging_log_levels.values())

Definition at line 28 of file utils.py.

logging_log_levels

dict opmonlib.utils.logging_log_levels

Initial value:

=  {
    "CRITICAL": logging.CRITICAL,
    "ERROR": logging.ERROR,
    "WARNING": logging.WARNING,
    "INFO": logging.INFO,
    "DEBUG": logging.DEBUG,
    "NOTSET": logging.NOTSET,
}

Definition at line 18 of file utils.py.

oks_log_level_keys

opmonlib.utils.oks_log_level_keys = list(oks_log_levels.keys())

Definition at line 37 of file utils.py.

oks_log_level_values

opmonlib.utils.oks_log_level_values = list(oks_log_levels.values())

Definition at line 38 of file utils.py.

oks_log_levels

dict opmonlib.utils.oks_log_levels

Initial value:

=  {
    "kTopPriority": 0,
    "kEventDriven": 1073741824,
    "kDefault": 2147483648,
    "kLowestPriority": 4294967295,
}

Definition at line 30 of file utils.py.

oks_to_logging_map

dict opmonlib.utils.oks_to_logging_map

Initial value:

=  {
    "kTopPriority": "ERROR",
    "kEventDriven": "WARNING",
    "kDefault": "INFO",
    "kLowestPriority": "DEBUG",
}

Definition at line 40 of file utils.py.

rich_log_format

str opmonlib.utils.rich_log_format = "%(filename)s %(name)s %(message)s"

Definition at line 82 of file utils.py.

time_zone

opmonlib.utils.time_zone = pytz.utc

Definition at line 85 of file utils.py.