mirror of
https://github.com/Telecominfraproject/openafc_final.git
synced 2025-11-01 10:37:51 +00:00
138 lines
4.5 KiB
Python
Executable File
138 lines
4.5 KiB
Python
Executable File
""" Common stuff for ULS Service related scripts """
|
|
|
|
# Copyright (C) 2022 Broadcom. All rights reserved.
|
|
# The term "Broadcom" refers solely to the Broadcom Inc. corporate affiliate
|
|
# that owns the software below.
|
|
# This work is licensed under the OpenAFC Project License, a copy of which is
|
|
# included with this software program.
|
|
|
|
# pylint: disable=wrong-import-order, invalid-name,
|
|
# pylint: disable=logging-fstring-interpolation, global-statement
|
|
|
|
import __main__
|
|
import inspect
|
|
import logging
|
|
import os
|
|
import pydantic
|
|
import sys
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
# Default directory for status filers
|
|
DEFAULT_STATUS_DIR = os.path.join(os.path.dirname(__file__), "status")
|
|
|
|
# Exit code to use in error()
|
|
error_exit_code = 1
|
|
|
|
# Logging level to use in error()
|
|
error_log_level = logging.CRITICAL
|
|
|
|
|
|
def set_error_exit_params(exit_code: Optional[int] = None,
|
|
log_level: Optional[int] = None) -> None:
|
|
""" Sets what to do in error
|
|
|
|
Arguments:
|
|
exit_code -- None or what exit code to use
|
|
log_level -- None or log level to use
|
|
"""
|
|
global error_exit_code, error_log_level
|
|
if exit_code is not None:
|
|
error_exit_code = exit_code
|
|
if log_level is not None:
|
|
error_log_level = log_level
|
|
|
|
|
|
def error(msg: str) -> None:
|
|
""" Prints given msg as error message and exit abnormally """
|
|
logging.log(error_log_level, msg)
|
|
sys.exit(error_exit_code)
|
|
|
|
|
|
def error_if(cond: Any, msg: str) -> None:
|
|
""" If condition evaluates to true prints given msg as error message and
|
|
exits abnormally """
|
|
if cond:
|
|
error(msg)
|
|
|
|
|
|
def dp(*args, **kwargs) -> None:
|
|
"""Print debug message
|
|
|
|
Arguments:
|
|
args -- Format and positional arguments. If latter present - formatted
|
|
with %
|
|
kwargs -- Keyword arguments. If present formatted with format()
|
|
"""
|
|
msg = args[0] if args else ""
|
|
if len(args) > 1:
|
|
msg = msg % args[1:]
|
|
if args and kwargs:
|
|
msg = msg.format(**kwargs)
|
|
cur_frame = inspect.currentframe()
|
|
assert (cur_frame is not None) and (cur_frame.f_back is not None)
|
|
frameinfo = inspect.getframeinfo(cur_frame.f_back)
|
|
logging.info(f"DP {frameinfo.function}()@{frameinfo.lineno}: {msg}")
|
|
|
|
|
|
def setup_logging(verbose: bool) -> None:
|
|
""" Logging setup
|
|
|
|
Arguments:
|
|
verbose -- enable debug printout
|
|
"""
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setFormatter(
|
|
logging.Formatter(
|
|
f"{os.path.basename(__main__.__file__)}. "
|
|
f"%(levelname)s: %(asctime)s %(message)s"))
|
|
logging.getLogger().addHandler(console_handler)
|
|
logging.getLogger().setLevel(
|
|
logging.DEBUG if verbose else logging.INFO)
|
|
if verbose:
|
|
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
|
|
|
|
|
|
def env_help(settings_class: Any, arg: str, prefix: str = ". ") -> str:
|
|
""" Prints help on environment variable for given command line argument
|
|
(aka setting name).
|
|
|
|
Environment variable name must be explicitly defined in Field() with 'env='
|
|
|
|
Arguments:
|
|
settings_class -- Type, derived from pydantic.BaseSettings
|
|
arg -- Command line argument
|
|
prefix -- Prefix to use if result is nonempty
|
|
Returns fragment for help message
|
|
"""
|
|
props = settings_class.schema()["properties"].get(arg)
|
|
error_if(props is None,
|
|
f"Command line argument '--{arg}' not found in settings class "
|
|
f"{settings_class.schema()['title']}")
|
|
assert props is not None
|
|
ret: List[str] = []
|
|
default = props.get("default")
|
|
if default is not None:
|
|
ret.append(f"Default is '{default}'")
|
|
if "env" in props:
|
|
ret.append(f"May be set with '{props['env']}' environment variable")
|
|
value = os.environ.get(props["env"])
|
|
if value is not None:
|
|
ret[-1] += f" (which is currently '{value}')"
|
|
if "default" not in props:
|
|
ret.append("This parameter is mandatory")
|
|
return (prefix + ". ".join(ret)) if ret else ""
|
|
|
|
|
|
def merge_args(settings_class: Any, args: Any) -> pydantic.BaseSettings:
|
|
""" Merges settings from command line arguments and Pydantic settings
|
|
|
|
Arguments:
|
|
settings_class -- Type, derived from pydantic.BaseSettings
|
|
args -- Arguments, parsed by ArgumentParser
|
|
Returns Object of type derived from pydantic.BaseSettings
|
|
"""
|
|
kwargs: Dict[str, Any] = \
|
|
{k: getattr(args, k) for k in settings_class.schema()["properties"]
|
|
if getattr(args, k) not in (None, False)}
|
|
return settings_class(**kwargs)
|