<s> from __future__ import absolute_import <EOL> from . base import SourceNode <EOL> from . . ds . csv_streams import CSVDataSource <EOL> from . . ds . elasticsearch_streams import ESDataSource <EOL> from . . ds . gdocs_streams import GoogleSpreadsheetDataSource <EOL> from . . ds . sql_streams import SQLDataSource <EOL> from . . ds . xls_streams import XLSDataSource <EOL> from . . ds . yaml_dir_streams import YamlDirectoryDataSource <EOL> class RowListSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT:list>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , a_list = None , fields = None ) : <EOL> if a_list : <EOL> self . list = a_list <EOL> else : <EOL> self . list = [ ] <EOL> self . fields = fields <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . fields <EOL> def run ( self ) : <EOL> for row in self . list : <EOL> self . put ( row ) <EOL> class RecordListSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , a_list = None , fields = None ) : <EOL> super ( RecordListSourceNode , self ) . __init__ ( ) <EOL> if a_list : <EOL> self . list = a_list <EOL> else : <EOL> self . list = [ ] <EOL> self . fields = fields <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . fields <EOL> def run ( self ) : <EOL> for record in self . list : <EOL> self . put ( record ) <EOL> class StreamSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , stream ) : <EOL> super ( StreamSourceNode , self ) . __init__ ( ) <EOL> self . stream = stream <EOL> def initialize ( self ) : <EOL> self . stream . initialize ( ) <EOL> @ property <EOL> def output_fields ( self ) : <EOL> return self . stream . fields <EOL> def run ( self ) : <EOL> for row in self . stream . rows ( ) : <EOL> self . put ( row ) <EOL> def finalize ( self ) : <EOL> self . stream . finalize ( ) <EOL> class CSVSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:default>" : "<STR_LIT:True>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , resource = None , * args , ** kwargs ) : <EOL> super ( CSVSourceNode , self ) . __init__ ( ) <EOL> self . resource = resource <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> self . stream = None <EOL> self . fields = None <EOL> self . _output_fields = None <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . stream : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not self . _output_fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . _output_fields <EOL> def initialize ( self ) : <EOL> self . stream = CSVDataSource ( self . resource , * self . args , ** self . kwargs ) <EOL> if self . fields : <EOL> self . stream . fields = self . fields <EOL> self . stream . initialize ( ) <EOL> self . _output_fields = self . stream . fields . copy ( ) <EOL> self . _output_fields . retype ( self . _retype_dictionary ) <EOL> def run ( self ) : <EOL> for row in self . stream . rows ( ) : <EOL> self . put ( row ) <EOL> def finalize ( self ) : <EOL> self . stream . finalize ( ) <EOL> class XLSSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:default>" : "<STR_LIT:True>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( XLSSourceNode , self ) . __init__ ( ) <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> self . stream = None <EOL> self . _fields = None <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . stream : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not self . stream . fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . stream . fields <EOL> def __set_fields ( self , fields ) : <EOL> self . _fields = fields <EOL> if self . stream : <EOL> self . stream . fields = fields <EOL> def __get_fields ( self ) : <EOL> return self . _fields <EOL> fields = property ( __get_fields , __set_fields ) <EOL> def initialize ( self ) : <EOL> self . stream = XLSDataSource ( * self . args , ** self . kwargs ) <EOL> if self . _fields : <EOL> self . stream . fields = self . _fields <EOL> self . stream . initialize ( ) <EOL> self . _fields = self . stream . fields <EOL> def run ( self ) : <EOL> for row in self . stream . rows ( ) : <EOL> self . put ( row ) <EOL> def finalize ( self ) : <EOL> self . stream . finalize ( ) <EOL> class YamlDirectorySourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT:path>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT:default>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:default>" : "<STR_LIT:True>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( YamlDirectorySourceNode , self ) . __init__ ( ) <EOL> self . kwargs = kwargs <EOL> self . args = args <EOL> self . stream = None <EOL> self . fields = None <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . stream : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not self . stream . fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . stream . fields <EOL> def initialize ( self ) : <EOL> self . stream = YamlDirectoryDataSource ( * self . args , ** self . kwargs ) <EOL> self . stream . fields = self . fields <EOL> self . stream . initialize ( ) <EOL> def run ( self ) : <EOL> for row in self . stream . rows ( ) : <EOL> self . put ( row ) <EOL> def finalize ( self ) : <EOL> self . stream . finalize ( ) <EOL> class GoogleSpreadsheetSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT:username>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT:password>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( GoogleSpreadsheetSourceNode , self ) . __init__ ( ) <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> self . stream = None <EOL> self . _fields = None <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . stream : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not self . stream . fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . stream . fields <EOL> def __getattr__ ( self , key ) : <EOL> try : <EOL> return getattr ( self . stream , key ) <EOL> except AttributeError : <EOL> return object . __getattr__ ( self , key ) <EOL> def __set_fields ( self , fields ) : <EOL> self . _fields = fields <EOL> if self . stream : <EOL> self . stream . fields = fields <EOL> def __get_fields ( self ) : <EOL> return self . _fields <EOL> fields = property ( __get_fields , __set_fields ) <EOL> def initialize ( self ) : <EOL> self . stream = GoogleSpreadsheetDataSource ( * self . args , ** self . kwargs ) <EOL> if self . _fields : <EOL> self . stream . fields = self . _fields <EOL> self . stream . initialize ( ) <EOL> self . _fields = self . stream . fields <EOL> def run ( self ) : <EOL> for row in self . stream . rows ( ) : <EOL> self . put ( row ) <EOL> def finalize ( self ) : <EOL> self . stream . finalize ( ) <EOL> class SQLSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> } , <EOL> ] <EOL> } <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( SQLSourceNode , self ) . __init__ ( ) <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> self . stream = None <EOL> self . _fields = None <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . stream : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not self . stream . fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . stream . fields <EOL> def __set_fields ( self , fields ) : <EOL> self . _fields = fields <EOL> if self . stream : <EOL> self . stream . fields = fields <EOL> def __get_fields ( self ) : <EOL> return self . _fields <EOL> fields = property ( __get_fields , __set_fields ) <EOL> def initialize ( self ) : <EOL> self . stream = SQLDataSource ( * self . args , ** self . kwargs ) <EOL> self . stream . initialize ( ) <EOL> self . _fields = self . stream . fields <EOL> def run ( self ) : <EOL> for row in self . stream . rows ( ) : <EOL> self . put ( row ) <EOL> def finalize ( self ) : <EOL> self . stream . finalize ( ) <EOL> class ESSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT:host>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT:port>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( ESSourceNode , self ) . __init__ ( ) <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> self . stream = None <EOL> self . _fields = None <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . stream : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if not self . stream . fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . stream . fields <EOL> def __set_fields ( self , fields ) : <EOL> self . _fields = fields <EOL> if self . stream : <EOL> self . stream . fields = fields <EOL> def __get_fields ( self ) : <EOL> return self . _fields <EOL> fields = property ( __get_fields , __set_fields ) <EOL> def initialize ( self ) : <EOL> self . stream = ESDataSource ( * self . args , ** self . kwargs ) <EOL> self . stream . initialize ( ) <EOL> self . _fields = self . stream . fields <EOL> def run ( self ) : <EOL> for row in self . stream . rows ( ) : <EOL> self . put ( row ) <EOL> def finalize ( self ) : <EOL> self . stream . finalize ( ) <EOL> class GeneratorFunctionSourceNode ( SourceNode ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT:args>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , function = None , fields = None , * args , ** kwargs ) : <EOL> super ( GeneratorFunctionSourceNode , self ) . __init__ ( ) <EOL> self . function = function <EOL> self . fields = fields <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . fields : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . fields <EOL> def run ( self ) : <EOL> for row in self . function ( * self . args , ** self . kwargs ) : <EOL> self . put ( row ) </s>
