pbcommand Utils

Util functions

Utils for common funcs, such as setting up a log, composing functions.

class pbcommand.utils.Constants[source]

Log Level format strings

exception pbcommand.utils.ExternalCommandNotFoundError[source]

External command is not found in Path

class pbcommand.utils.Singleton(name, bases, dct)[source]

General Purpose singleton class

Usage:

>>> class MyClass(object):
>>>     __metaclass__ = Singleton
>>>     def __init__(self):
>>>         self.name = 'name'
pbcommand.utils.compose(*funcs)[source]

Functional composition of a non-empty list

[f, g, h] will be f(g(h(x)))

Example:
>>> f = lambda x: x * x
>>> g = lambda x: x + 1
>>> h = lambda x: x * 2
>>> funcs = [f, g, h]
>>> fgh = compose(*funcs)
>>> fgh(3) # 49
>>> compose(f, g, h)(3)
pbcommand.utils.get_dataset_metadata(path)[source]

Returns DataSetMeta data or raises ValueError if dataset XML is missing the required UniqueId and MetaType values.

Parameters:path – Path to DataSet XML
Raises:ValueError
Returns:DataSetMetaData
pbcommand.utils.get_dataset_metadata_or_none(path)[source]

Returns DataSetMeta data, else None if the file doesn’t exist or a processing of the XML raises.

Parameters:path – Path to DataSet XML
Returns:DataSetMetaData or None
pbcommand.utils.get_parsed_args_log_level(pargs, default_level=20)[source]

Utility for handling logging setup flexibly in a variety of use cases, assuming standard command-line arguments.

Parameters:
  • pargs – argparse namespace or equivalent
  • default_level – logging level to use if the parsed arguments do not specify one
pbcommand.utils.is_dataset(path)[source]

peek into the XML to get the MetaType and verify that it’s a valid dataset

Parameters:path – Path to DataSet XML
pbcommand.utils.log_traceback(alog, ex, ex_traceback)[source]

Log a python traceback in the log file

Parameters:
  • ex – python Exception instance
  • ex_traceback – exception traceback

Example Usage (assuming you have a log instance in your scope)

Example:
>>> value = 0
>>> try:
>>>    1 / value
>>> except Exception as e:
>>>    msg = "{i} failed validation. {e}".format(i=value, e=e)
>>>    log.error(msg)
>>>    _, _, ex_traceback = sys.exc_info()
>>>    log_traceback(log, e, ex_traceback)
pbcommand.utils.nfs_exists_check(ff)[source]

Central place for all NFS hackery

Return whether a file or a dir ff exists or not. Call listdir() instead of os.path.exists() to eliminate NFS errors.

Added try/catch black hole exception cases to help trigger an NFS refresh

Rtype bool:
pbcommand.utils.pool_map(func, args, nproc)[source]

Wrapper for calling a function in parallel using the multiprocessing module and blocking until results are available.

pbcommand.utils.setup_log(alog, level=20, file_name=None, log_filter=None, str_formatter='[%(levelname)s] %(asctime)-15sZ [%(name)s %(funcName)s %(lineno)d] %(message)s')[source]

Core Util to setup log handler

Parameters:
  • alog – a log instance
  • level – (int) Level of logging debug
  • file_name – (str, None) if None, stdout is used, str write to file
  • log_filter – (LogFilter, None)
  • str_formatter – (str) log formatting str

Warning

THIS NEEDS TO BE DEPRECATED

pbcommand.utils.setup_logger(file_name_or_none, level, formatter='[%(levelname)s] %(asctime)-15sZ [%(name)s %(funcName)s %(lineno)d] %(message)s')[source]
Parameters:
  • file_name_or_none – Path to log file, None will default to stdout
  • level – logging.LEVEL of
  • formatter – Log Formatting string
pbcommand.utils.walker(root_dir, file_filter_func)[source]

Walk the file sytem and filter by the supplied filter function.

Filter function F(path) -> bool

pbcommand.utils.which(exe_str)[source]

walk the current PATH for exe_str to get the absolute path of the exe

Parameters:exe_str – Executable name
Return type:str | None

:returns Absolute path to the executable or None if the exe is not found

pbcommand.utils.which_or_raise(cmd)[source]

Find exe in path or raise ExternalCommandNotFoundError