Other Python API Docs

Reporter for diagnostic messages.

class metomi.rose.reporter.Reporter(verbosity=1, contexts=None, raise_on_exc=False)

Report diagnostic messages.

Note: How about the “logging” module in the standard library? It needs a lot of fiddling to get it working in our reporting model. We want: * [FAIL] in any verbosity to stderr. * [WARN] in default verbosity to stderr. * Raw output to stdout in any verbosity. * [INFO] in default verbosity to stdout. * [INFO] in -v verbosity to stdout. * [INFO] in -vv verbosity to stdout. * and everything goes to the log file with -vv verbosity by default.

classmethod default(verbosity=None, reset=False)

Return the default reporter.

format_msg(msg, verbosity, prefix=None, clip=None)

Format a message for reporting.

report(message, kind=None, level=None, prefix=None, clip=None)

Report a message, if relevant for the reporter contexts.

message:

The message to report. An Event, an Exception, an object with a __str__ method, or a callable that returns an object with a __str__ method.

kind:

The message kind. The default is determined by message. If it is an Event, the default is message.kind. If it is an Exception, the default is KIND_ERR. Otherwise, the default is KIND_OUT.

level:

The level of the message. The default is determined by message. If it is an Event, the default is message.level. If it is an Exception, the default is FAIL. Otherwise, the default is DEFAULT.

prefix:

Prefix each line of the message with this prefix. Default is context dependent.

clip:

The maximum length of the message to print.

If self.event_handler is defined, self.event_handler with all the arguments and return its result instead.

Wraps Python’s subprocess.Popen.

class metomi.rose.popen.RosePopener(event_handler=None)

Wrap Python’s subprocess.Popen.

get_cmd(key: str, *args: str) → List[str]

Return default options and arguments of a known command as a list.

If a setting [external] <key> is defined in the site/user configuration, use the setting.

Otherwise, if RosePopener.ENVS_OF_CMDS[key] exists, it looks for each environment variable in the list in RosePopener.ENVS_OF_CMDS[key] in order. If the environment variable is defined and is not a null string, use the value of the environment variable.

Otherwise, return RosePopener.CMDS[key]

key: must be a key of RosePopener.CMDS args: if specified, will be added to the returned list

handle_event(*args, **kwargs)

Handle an event using the runner’s event handler.

static list_to_shell_str(args: Iterable[str]) → str

Convert a list of strings into a shell command, escaping whitespace and quotes, but otherwise allowing special chars so user defined commands can run without sanitisation.

Examples: >>> RosePopener.list_to_shell_str([]) ‘’ >>> _ = RosePopener.list_to_shell_str([“it’s”]) >>> print(_) it's >>> _ “it\’s” >>> RosePopener.list_to_shell_str([“echo”, “-n”, “Multiple words”]) ‘echo -n Multiple\ words’ >>> RosePopener.list_to_shell_str([“ls”, “my_dir;allow_this”]) ‘ls my_dir;allow_this’ >>> RosePopener.list_to_shell_str([“what”, “about”, “globs*”]) ‘what about globs*’

run(*args, **kwargs)

Provide a Rose-friendly interface to subprocess.Popen.

Return ret_code, out, err.

If kwargs[“stdin”] is a str, communicate it to the command via a pipe.

async run_async(*args, **kwargs)

Provide a Rose-friendly interface to subprocess.Popen.

Return ret_code, out, err.

If kwargs[“stdin”] is a str, communicate it to the command via a pipe.

run_bg(*args, **kwargs)

Provide a Rose-friendly interface to subprocess.Popen.

Return a subprocess.Popen object.

If kwargs[“stdin”] is a str, turn it into subprocess.PIPE. However, it cannot communicate stdin to the Popen object, because the Popen.communicate() method implies a wait().

async run_bg_async(*args, **kwargs)

Provide a Rose-friendly interface to subprocess.Popen.

Return a subprocess.Popen object.

If kwargs[“stdin”] is a str, turn it into subprocess.PIPE. However, it cannot communicate stdin to the Popen object, because the Popen.communicate() method implies a wait().

run_nohup_gui(cmd)

Launch+detach a GUI command with nohup.

Launch the GUI command with nohup bash -c ‘exec …’ redirecting standard input and outputs from and to /dev/null, and setting it to become a process group leader. Equivalent to a double fork.

Parameters:

cmd (str) – command string of the GUI.

Return (subprocess.Popen):

Return the process object for the “nohup” command. Return None if no DISPLAY is set.

run_ok(*args, **kwargs)

Same as RosePopener.run, but raise RosePopenError if ret_code != 1.

Return out, err.

run_simple(*args, **kwargs)

Similar to RosePopener.run_ok, but event handle stdout and stderr.

kwargs[“stderr_level”] – set event level of stderr kwargs[“stdout_level”] – set event level of stdout

Return None.

static shlex_join(args: Iterable[str]) → str

Convert a list of strings into a shell command, safely quoting when the strings contain whitespace and special chars.

Basically a back-port of shlex.join(), needed for py 3.7.

Examples: >>> RosePopener.shlex_join([]) ‘’ >>> RosePopener.shlex_join([“echo”, “-n”, “Multiple words”]) “echo -n ‘Multiple words’” >>> RosePopener.shlex_join([“ls”, “my_dir;foiled_injection”]) “ls ‘my_dir;foiled_injection’” >>> RosePopener.shlex_join([“what”, “about”, “globs*”]) “what about ‘globs*’”

static which(name)

Search an executable file name in PATH, and return its full path.

If name is an absolute path and is an executable file, return name. If name is not found in PATH, return None.

Module to contain internal system macros for operating on a configuration.