onyx.dataflow.command – A dataflow processing element for executing command line subprocesses

The CommandLineProcessor class provides a dataflow processing element that executes command line subprocesses. Activation is caused by being sent an event consisting of a dictionary of variable bindings; any other event type is an error. Upon successful completion of the subprocess, the element sends the dictionary to its successor. If the command cannot be run, or runs and returns a non-zero value, an error is thrown and no event is sent to the successor. The last_command_line, last_stdout_output, and last_stderr_output properties can be used to see what command was run and the output to the two streams.

class onyx.dataflow.command.CommandLineProcessor(command_line, sendee=None, sending=True, bypass_types=(), variable_dict={}, notify_stream=None)

Bases: onyx.dataflow.streamprocess.ProcessorBase

CommandLineProcessors are constructed with the following arguments. command_line gives the form of the command line and may be either a string or a sequence of strings. sendee, sending, and bypass_types are the usual optional arguments for a dataflow processing element. Note that there is no process_types argument; this processor always processes dicts. Also note that if sending is True, the sendee must be set before the element’s process function is called. variable_dict is a dictionary with keys which are variable names and values to bind them to; this argument defaults to {}. If notify_stream is not None, it should be Python file object (like sys.stdout), and the processor will write a time-stamped copy of the command line to the stream before the command is executed, and a time-stamped notice to the stream when it finishes.

>>> result = list()

>>> clp0 = CommandLineProcessor(r"ls -d /usr", sendee=result.append)

When a CommandLineProcessor is sent an event in the form of a dictionary, the command is run and a True is sent to the sendee.

>>> clp0.process({})
>>> result
[{}]

Other event types (unless listed in bypass_types) raise errors and there is no event sent out.

>>> clp0.process(False)
Traceback (most recent call last):
...
ValueError: processor type CommandLineProcessor expected event type in one of these two groups: (<type 'dict'>,) (), but got <type 'bool'>

>>> result
[{}]

The command may refer to variables, which will be looked up in the event dictionary. A variable is a symbol preceded by a ‘$’ or it may be written as ${symbol}.

>>> clp1 = CommandLineProcessor("ls -d $dir", result.append)
>>> clp1.process({"dir":r"/usr"})
>>> result
[{}, {'dir': '/usr'}]
>>> clp1.last_command_line
'ls -d /usr'

Variables will also be looked up in the optional variable_dict argument to the constructor.

>>> clp2 = CommandLineProcessor("ls -d ${dir}", result.append, variable_dict={"dir":r"/usr"})
>>> clp2.process({})
>>> result
[{}, {'dir': '/usr'}, {}]
>>> clp2.last_command_line
'ls -d /usr'

Commands may be given as sequences of strings. Also note that bindings in the event take precedence over bindings given to the constructor.

>>> clp3 = CommandLineProcessor(("ls", "-d", "${dir}"), result.append, variable_dict={"dir":r"/etc"})
>>> clp3.process({"dir":r"/usr"})
>>> result
[{}, {'dir': '/usr'}, {}, {'dir': '/usr'}]
>>> clp3.last_command_line
'ls -d /usr'

Variable dictionaries may have values which are not strings, the command will be built by converting the value to a string.

>>> note_stream = StringIO()
>>> var_dict = {"dir":r"/usr", "one":1}
>>> clp4 = CommandLineProcessor(("ls", "-d", "-$one", "${dir}"), result.append, notify_stream=note_stream)
>>> clp4.process(var_dict)
>>> result
[{}, {'dir': '/usr'}, {}, {'dir': '/usr'}, {'dir': '/usr', 'one': 1}]

>>> clp4.last_command_line
'ls -d -1 /usr'
>>> clp4.last_stdout_output.strip()
'/usr'

>>> clp4.last_stderr_output
''
>>> print note_stream.getvalue()  
[20...] Starting: ls -d -1 /usr
[20...] Finished: ls -d -1 /usr
<BLANKLINE>

dc

A debug context for this processor. This attribute is an object returned by dcheck() in the onyx.util.debugprint module, and may be used for debug output. The tag for turning on such output is available as debug_tag

debug_tag

Activate this tag to get debugging information, see onyx.util.debugprint.DebugPrint

graph

Return a graph for this processor. By default this is just a single node whose label is the label of processor; derived classes may wish to override this property.

label

Return a label for this processor. By default this is just the name of the class; derived classes may wish to override this property by providing a different label to __init__().

last_command_line

last_stderr_output

last_stdout_output

notify_stream

process(event)

send(result)

Internal function that pushes result into the sendee. Implementations of process() must call this to push results. To set up the sendee, (the target of the push), clients of the processor must either initialize the object with a sendee, or call set_sendee(). Processors created with a sendee of False will never send, but will not error if send is called.

sendee

The callable this processor will use to send events; see set_sendee()

sending

Whether this processor will currently send events at all; see set_sending()

set_notify_stream(stream)

set_sendee(sendee)

Clients call this to set up the callable where the processor will send its results.

set_sending(sending)

Clients call this to turn sending from a processor on or off.

static std_process_prologue(process_function)

Subclasses may use this decorater on their process function to implement the usual bypass and process semantics and to set up the debug context returned by dc().