Source code for linchpin.cli.context

#!/usr/bin/env python

import os
import ast
import errno
import logging

from linchpin.context import LinchpinContext

[docs]class LinchpinCliContext(LinchpinContext): """ Context object, which will be used to manage the cli, and load the configuration file """ def __init__(self): """ Initializes basic variables """ # The following values are set in the parent class # # self.version = __version__ # self.verbosity = 1 # # self.lib_path = '{0}'.format(os.path.dirname( # os.path.realpath(__file__))) # # self.cfgs = {} # Load constants from linchpin.constants file. # self._load_constants() LinchpinContext.__init__(self)
[docs] def load_config(self, lpconfig=None): """ Update self.cfgs from the linchpin configuration file (linchpin.conf). The following paths are used to find the config file. The search path defaults to the first-found order:: * /etc/linchpin.conf * /linchpin/library/path/linchpin.conf * <workspace>/linchpin.conf An alternate search_path can be passed. :param search_path: A list of paths to search a linchpin config (default: None) """ if not self.workspace: self.workspace = os.path.realpath(os.path.curdir) expanded_path = None if lpconfig: CONFIG_PATH = [lpconfig] else: CONFIG_PATH = [ '/etc/linchpin.conf', '~/.config/linchpin/linchpin.conf', '{0}/linchpin.conf'.format(self.workspace) ] existing_paths = [] for path in CONFIG_PATH: expanded_path = ( "{0}".format(os.path.realpath(os.path.expanduser(path)))) # implement first found if os.path.exists(expanded_path): # logging before the config file is setup doesn't work # if messages are needed before this, use print. existing_paths.append(expanded_path) for path in existing_paths: self._parse_config(path)
@property def pinfile(self): """ getter function for pinfile name """ return self.get_cfg('init', 'pinfile') @pinfile.setter def pinfile(self, pinfile): """ setter for workspace """ self.set_cfg('init', 'pinfile', pinfile) @property def workspace(self): """ getter function for workspace """ return self.get_cfg('lp', 'workspace') @workspace.setter def workspace(self, workspace): """ setter for workspace """ self.set_cfg('lp', 'workspace', workspace) self.set_evar('workspace', workspace)
[docs] def setup_logging(self): """ Setup logging to a file, console, or both. Modifying the `linchpin.conf` appropriately will provide functionality. """ self.enable_logging = ast.literal_eval( self.cfgs['logger'].get('enable', 'True')) if self.enable_logging: # create logger self.logger = logging.getLogger('lp_logger') self.logger.setLevel(eval(self.cfgs['logger'].get('level', 'logging.DEBUG'))) logfile = os.path.realpath(os.path.expanduser( self.cfgs['logger'].get('file', 'linchpin.log'))) logdir = os.path.dirname(logfile) if not os.path.exists(logdir): try: os.makedirs(logdir) except OSError as exc: if (exc.errno == errno.EEXIST and os.path.isdir(logdir)): pass else: raise fh = logging.FileHandler(logfile) fh.setLevel(eval(self.cfgs['logger'].get('level', 'logging.DEBUG'))) formatter = logging.Formatter( self.cfgs['logger'].get('format', '%(levelname)s' ' %(asctime)s %(message)s')) fh.setFormatter(formatter) self.logger.addHandler(fh) self.console = logging.getLogger('lp_console') self.console.setLevel(eval(self.cfgs['console'].get('level', 'logging.INFO'))) ch = logging.StreamHandler() ch.setLevel(eval(self.cfgs['console'].get('level', 'logging.INFO'))) formatter = logging.Formatter( self.cfgs['console'].get('format', '%(message)s')) ch.setFormatter(formatter) self.console.addHandler(ch)
[docs] def log(self, msg, **kwargs): """ Logs a message to a logfile or the console :param msg: message to log :param lvl: keyword argument defining the log level :param msg_type: keyword argument giving more flexibility. .. note:: Only msg_type `STATE` is currently implemented. """ lvl = kwargs.get('level') msg_type = kwargs.get('msg_type') if lvl is None: lvl = logging.INFO if self.verbosity > 1 and not msg_type: self.console.log(logging.INFO, msg) state_msg = msg if msg_type == 'STATE': state_msg = 'STATE - {0}'.format(msg) self.console.log(logging.INFO, msg) if self.enable_logging: self.logger.log(lvl, state_msg)
[docs] def log_state(self, msg): """Logs a message to stdout""" self.log(msg, msg_type='STATE', level=logging.DEBUG)
[docs] def log_info(self, msg): """Logs an INFO message """ self.log(msg, level=logging.INFO)
[docs] def log_debug(self, msg): """Logs a DEBUG message""" self.log(msg, level=logging.DEBUG)