Viewing file:  __init__.py (2.8 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# https://cloudlinux.com/docs/LICENCE.TXT
#
from __future__ import absolute_import
import logging
import inspect
import sys
from dataclasses import dataclass
from clcommon.utils import run_command, ExternalProgramFailed
def _get_default_args(func):
"""
Get dict with key-value pairs of default
argument values
"""
signature = inspect.signature(func)
return {
k: v.default
for k, v in signature.parameters.items()
if v.default is not inspect.Parameter.empty
}
def _get_args_dict(func, args, kwargs):
"""
Get all func arguments including defaults, args and kwargs
"""
func_args = {}
func_args.update(_get_default_args(func))
func_args.update(dict(zip(func.__code__.co_varnames, args)))
func_args.update(kwargs)
return func_args
def task(name):
"""
Simple wrapper that logs execution begin and end
"""
def decorator(func):
def wrapped_func(*args, **kwargs):
func_args = _get_args_dict(func, args, kwargs)
logging.info(name.format(**func_args))
try:
func(*args, **kwargs)
except Exception:
logging.exception('FAILED to %s', name.format(**func_args))
raise
return wrapped_func
return decorator
def setup_logger(logger_name, log_file):
"""
Logger setup for all modules
"""
app_logger = logging.getLogger(logger_name)
app_logger.setLevel(logging.DEBUG)
try:
fh = logging.FileHandler(log_file)
except IOError:
pass
else:
fh.formatter = logging.Formatter(
'[%(levelname)s | %(asctime)s]: %(message)s')
fh.level = logging.DEBUG
app_logger.addHandler(fh)
for handler in app_logger.handlers:
if isinstance(handler, logging.StreamHandler) and handler.stream == sys.stderr:
handler.formatter = logging.Formatter('[cloudlinux-customizer]: %(message)s')
handler.level = logging.INFO
return app_logger
@dataclass
class RunResult:
"""
Describes process call result
"""
exitcode: int
stdout: str
stderr: str
def run(cmd, fail_ok=False):
"""
Just a wrapper above run_command that provides some
cool logging of what happens in subprocess
"""
logging.debug('Running %s', ' '.join(cmd))
try:
exitcode, stdout, stderr = run_command(cmd, return_full_output=True)
logging.debug('Stdout: %s', stdout)
logging.debug('Stderr: %s', stderr)
return RunResult(exitcode, stdout, stderr)
except ExternalProgramFailed as e:
if not fail_ok:
logging.error(str(e))
return RunResult(-1, None, None)
|