command_watcher¶

Module to watch the execution of shell scripts. Both streams (stdout and stderr) are captured.

watch = Watch()
watch.log.critical(msg)
watch.log.error(msg)
watch.log.warning(msg)
watch.log.info(msg)
watch.log.debug(msg)
watch.run(['rsync', '-av', '/home', '/backup'])

class jflib.command_watcher.BaseChannel[source]¶

Base class for all reporters

abstract report(message)[source]¶

Return type: None

class jflib.command_watcher.BaseClass[source]¶

class jflib.command_watcher.BeepChannel[source]¶

Send beep sounds.

beep(frequency=4186.01, length=50)[source]¶

Generate a beep sound using the “beep” command.

A success tone: frequency=4186.01, length=40
A failure tone: frequency=65.4064, length=100

Parameters

frequency (float) – Frequency in Hz.
length (float) – Length in milliseconds.

report(message)[source]¶

Send a beep sounds.

Parameters: message (Message) – A message object. The only attribute that takes an effect is the status attribute (0-3).

exception jflib.command_watcher.CommandWatcherError(msg, **data)[source]¶: Exception raised by this module.

class jflib.command_watcher.EmailChannel(smtp_server, smtp_login, smtp_password, to_addr, from_addr='', to_addr_critical='')[source]¶

Send reports by e-mail.

report(message)[source]¶

Send an e-mail message.

Parameters: message (Message) – A message object.

class jflib.command_watcher.ExtendedLogger(name, level=0)[source]¶

stderr(line, *args, **kws)[source]¶

Return type: None

stdout(line, *args, **kws)[source]¶

Return type: None

class jflib.command_watcher.IcingaChannel(url, user, password, service_name)[source]¶

password: str¶

report(message)[source]¶

service_name: str¶

url: str¶

user: str¶

class jflib.command_watcher.LoggingHandler(master_logger=None)[source]¶

Store of all logging records in the memory. Print all records on emit.

property all_records¶: All log messages joined by line breaks.

emit(record)[source]¶

Parameters: record (LogRecord) – A record object.

property stderr¶

property stdout¶

class jflib.command_watcher.Message(**data)[source]¶

This message class bundles all available message data into an object. The different reporters can choose which data they use.

property body: str¶

Text body for the e-mail message.

Return type: str

property custom_message: str¶

Return type: str

property message: str¶

Return type: str

property message_monitoring: str¶

message + performance_data

Return type: str

property performance_data: str¶

Returns: A concatenated string
Return type: str

property prefix: str¶

Return type: str

property processes: Optional[str]¶

Return type: Optional[str]

property service_name: str¶

Return type: str

property status: int¶

0 (OK), 1 (WARNING), 2 (CRITICAL), 3 (UNKOWN): see Nagios / Icinga monitoring status / state.

Return type: int

property status_text: str¶

The status as a text word like OK.

Return type: str

property user: str¶

Return type: str

class jflib.command_watcher.MessageParams(*args, **kwargs)[source]¶

body: str¶

custom_message: str¶

log_records: str¶: Log records separated by new lines

prefix: str¶

preformance_data: Dict[str, Any]¶

processes: List[Process]¶

service_name: str¶: The name of the service.

status: Literal[0, 1, 2, 3]¶: 0 (OK), 1 (WARNING), 2 (CRITICAL), 3 (UNKOWN): see Nagios / Icinga monitoring status / state.

class jflib.command_watcher.MinimalMessageParams(*args, **kwargs)[source]¶

body: str¶: A longer report text.

custom_message: str¶: Custom message

prefix: str¶: Prefix of the report message.

preformance_data: Dict[str, Any]¶: A dictionary like {‘perf_1’: 1, ‘perf_2’: ‘test’}

class jflib.command_watcher.Process(args, master_logger=None, **kwargs)[source]¶

Run a process.

You can use all keyword arguments from subprocess.Popen except bufsize, stderr, stdout.

Parameters: args (Union[str, List[str], Tuple[str]]) – List, tuple or string. A sequence of process arguments, like subprocess.Popen(args).

args: Args¶: Process arguments in various types.

property args_normalized: Sequence[str]¶

Normalized args, always a list

Return type: Sequence[str]

property line_count_stderr: int¶

The count of lines of the current stderr.

Return type: int

property line_count_stdout: int¶

The count of lines of the current stderr.

Return type: int

log: ExtendedLogger¶: A ready to go and configured logger.

log_handler: LoggingHandler¶

property stderr: str¶

Alias / shortcut for self.log_handler.stderr.

Return type: str

property stdout: str¶

Alias / shortcut for self.log_handler.stdout.

Return type: str

subprocess: subprocess.Popen[Any]¶

class jflib.command_watcher.ProcessArgs(*args, **kwargs)[source]¶

cwd: str¶: Sets the current directory before the child is executed.

env: Dict[str, Any]¶: Defines the environment variables for the new process.

shell: bool¶: If true, the command will be executed through the shell.

class jflib.command_watcher.Reporter[source]¶

Collect all channels.

add_channel(channel)[source]¶

channels: List[BaseChannel]¶

class jflib.command_watcher.Timer[source]¶

Measure the execution time of a command run.

interval¶: The time interval between start and stop.

result()[source]¶

Measure the time intervale

Returns: A formatted string displaying the result.
Return type: str

start¶: “The start time. (UNIX timestamp)

stop¶: “The time when the timer stops. (UNIX timestamp)

class jflib.command_watcher.Watch(config_file=None, service_name='command_watcher', raise_exceptions=True, config_reader=None, report_channels=None)[source]¶

Watch the execution of a command. Capture all output of a command. provide and setup a logging facility.

Parameters

config_file (Optional[str]) – The file path of the configuration file in the INI format.
service_name (str) – A name of the watched service.
raise_exceptions (bool) – Raise exceptions if watch.run() exists with a non-zero exit code.
config_reader (Optional[ConfigReader]) – A custom configuration reader. Specify this parameter to not use the build in configuration reader.

log: ExtendedLogger¶: A ready to go and configured logger.

processes: List[Process]¶: A list of completed processes Process. Everytime you use the method run() the process object is appened in the list.

property stderr¶: Alias / shortcut for self._log_handler.stderr.

property stdout¶: Alias / shortcut for self._log_handler.stdout.

jflib.command_watcher.setup_logging(master_logger=None)[source]¶

Setup a fresh logger for each watch action.

Parameters: master_logger (Optional[Logger]) – Forward all log messages to a master logger.
Return type: Tuple[ExtendedLogger, LoggingHandler]