Files

File ABC Source

class FileABCSource(app, pipeline, id=None, config=None)[source]

Bases: TriggerSource

Description:

FileABCSource.__init__()[source]

Description:

Parameters

appApplication: Name of the Application.
pipelinePipeline: Name of the Pipeline.
idID, default = None: ID
configJSON, default = None: Configuration file with additional information.

File ABC Source Methods

async FileABCSource.cycle()[source]: Cycles through a file.

async FileABCSource.simulate_event()[source]

The simulate_event method should be called in read method after a file line has been processed.

It ensures that all other asynchronous events receive enough time to perform their tasks. Otherwise, the application loop is blocked by a file reader and no other activity makes a progress.

async FileABCSource.read(filename, f)[source]

Description: Override this method to implement your File Source. f is an opened file object.

Parameters

filenamefile: Name of the file.

f :

File Block Source

class FileBlockSource(app, pipeline, id=None, config=None)[source]

Bases: FileABCSource

Description:

FileBlockSource.__init__()[source]

Description:

Parameters

appApplication: Name of the Application.
pipelinePipeline: Name of the Pipeline.
idID, default = None: ID
configJSON, default = None: Configuration file with additional information.

async FileBlockSource.read(filename, f)[source]

Loads a file.

Parameters

filenamefile: Name of the file.

f :

File Block Sink

class FileBlockSink(app, pipeline, id=None, config=None)[source]

Bases: Sink

Description:

** Config Defaults **

path : ‘’

mode : wb

flags : O_CREAT

FileBlockSink.__init__()[source]

Parameters

appApplication: Name of the Application
pipelinePipeline: Name of the Pipeline.
idID, default = None: ID
configJSON, default = None: Configuration file with additional information.

FileBlockSink.get_file_name(context, event)[source]

Override this method to gain control over output file name.

Parameters

context :

eventany type: a single unit of information that is propagated through the pipeline

Returns: config path

FileBlockSink.process(context, event)[source]

Opens a file.

Parameters

context :

eventany type: a single unit of information that is propagated through the pipeline

File csv Source

class FileCSVSource(app, pipeline, fieldnames=None, id=None, config=None)[source]

Bases: FileABCSource

Description:

FileCSVSource.__init__()[source]

Description:

Parameters

appApplication: Name of the Application.
pipelinePipeline: Name of the Pipeline.
idID, default = None: ID
configJSON, default = None: Configuration file with additional information.

FileCSVSource.reader(f)[source]

Description:

Parameters

f :

Returns

??

async FileCSVSource.read(filename, f)[source]

Description:

Parameters

filename :

f :

File csv Sink

class FileCSVSink(app, pipeline, id=None, config=None)[source]

Bases: Sink

Description:

** Default Config**

path : ‘’

dialect : ‘excel’

delimiter : ‘,’

doublequote : True

escapechar : “”

lineterminator : os.linesep

quotechar : ‘”’

quoting : csv.QUOTE_MINIMAL

skipinitialspace : False

strict : False

FileCSVSink.__init__()[source]: Description:

FileCSVSink.get_file_name(context, event)[source]

Description: Override this method to gain control over output file name.

Parameters

context :

event :

Returns: path of context and config

FileCSVSink.writer(f, fieldnames)[source]

Description:

Parameters

f :

fieldnamesfile: Name of the file.

Returns: dialect and fieldnames

FileCSVSink.process(context, event)[source]

Description:

Parameters

context :

eventany data type: Information with timestamp.

FileCSVSink.rotate()[source]: Description: Call this to close the currently open file.

File json Source

class FileJSONSource(app, pipeline, id=None, config=None)[source]

Bases: FileABCSource

Description: This file source is optimized to load even large JSONs from a file and parse that. The loading & parsing is off-loaded to the worker thread so that it doesn’t block the IO loop.

FileJSONSource.__init__()[source]

Description:

Parameters

app :

pipeline :

idID, default= None: ID
configJSON, default = None: configuration file with additional information

async FileJSONSource.read(filename, f)[source]

Description:

Parameters

filename :

f :

File line Source

class FileLineSource(app, pipeline, id=None, config=None)[source]

Bases: FileABCSource

Description:

FileLineSource.__init__()[source]

Description:

Parameters

app: Application: Name of the Application
pipelinePipeline: Name of the Pipeline

id : ID, default = None

configJSON, default = None: Configuration file with additional information

async FileLineSource.read(filename, f)[source]

Description:

Parameters

filename :

f :

File Multiline Source

class FileMultiLineSource(app, pipeline, separator, id=None, config=None)[source]

Bases: FileABCSource

Description: Read file line by line but try to join multi-line events by separator. Separator is a (fixed) pattern that should present at the begin of the line, if it is a new event.

Example: <133>1 2018-03-24T02:37:01+00:00 machine program 22068 - Start of the multiline event

2nd line of the event 3rd line of the event

<133>1 2018-03-24T02:37:01+00:00 machine program 22068 - New event

The separatpr is ‘<’ string in this case

FileMultiLineSource.__init__()[source]

Description:

Parameters

app: Application: Name of the Application
pipelinePipeline: Name of the Pipeline

separator :

id : ID, default = None

configJSON, default = None: Configuration file with additional information

async FileMultiLineSource.read(filename, f)[source]

Description:

Parameters

filename :

f :

Lookup Provider

class FileBatchLookupProvider(lookup, url, id=None, config=None)[source]

Bases: LookupBatchProviderABC

Loads lookup data from a file on local filesystem.

FileBatchLookupProvider.__init__()[source]: Description:

async FileBatchLookupProvider.load()[source]

Description:

Returns: result

FileBatchLookupProvider.load_on_thread()[source]: Description: