Base Classes

Base classes

class base.Logger(lvl='INFO')

Logger class used by Pipe. There are five levels of logs: INFO, DEBUG, WARNING, ERROR and CRITICAL. By default logger is set to INFO.

Parameters:lvl – log level, one of: info, debug, warning, error or critical
Returns:None
log(msg, name, lvl='info')

Logs messages.

Parameters:
  • msg – message to log
  • name – name of logger
  • lvl – log level, one of: info, debug, warning, error or critical
Returns:

None
class base.Pipe(functor_obj, name, upstreams=None, downstreams=None, ignore_exceptions=None, init_kwargs=None)

Base class for all pipe segments. Pipes use two sets of Streams: upstreams and downstreams. Generally Pipes except data from upstreams and pass downstream after a transformation. All pipe segments run on their own thread or process, which allows them to run in parallel with other segments.

Number of upstreams should be equal to number of functor args. Likewise, number of downstreams should be equal to number of functor outputs.

When Pipe produces a None it will not be passed downstream. In this case nothing will be placed on the downstreams. This allows the user to create ‘switches’ based on internal logic in the functor.

Parameters:
  • functor – Python function, class, generator or corountine
  • name – String associated with pipe segment
  • upstreams – List of Streams that are inputs to functor
  • downstreams – List of Streams that are outputs of functor
  • ignore_exceptions – List of exceptions to ignore while pipeline is running
  • init_kwargs – Kwargs to initiate class object on process (no used when func_type = ‘function’)
  • stateful – Set to True when using a class functor. Class functors must implement a ‘run’ method
class base.Sentinel

This class is used to indicate the end of a stream. When a instance of Sentinel is passed to a Pipe it will shut itself down.

class base.Stream(name='stream', buffer_size=3, timeout=None, monitor=False, queue_type='multiprocessing.Queue')

Based off of a multiprocessing queue, Stream handles moving data between Pipe segments.

Parameters:
  • name – String, name of Stream
  • buffer_size – Int, max size of queue. Will be ignored if queue_type = “multiprocessing.SimpleQueue”
  • timeout – Int, timeout(s) value for Queue.put and Queue.get methods
  • monitor – Bool, log stream I/O times
Queue_type:

String, multiprocesses queue type to be used. Valid types: ‘multiprocessing.Queue’, ‘multiprocessing.SimpleQueue’
Returns:

None