etl_lib.data_sink.CypherBatchSink module

class CypherBatchSink(context, task, predecessor, query, **kwargs)[source]

Bases: BatchProcessor

BatchProcessor to write batches of data to a Neo4j database.

Parameters:

context (ETLContext)
task (Task)
predecessor (BatchProcessor | None)
query (str)

__init__(context, task, predecessor, query, **kwargs)[source]

Constructs a new CypherBatchSink.

Parameters:

context (ETLContext) – etl_lib.core.ETLContext.ETLContext instance.
task (Task) – etl_lib.core.Task.Task instance owning this batchProcessor.
predecessor (BatchProcessor | None) – BatchProcessor which get_batch() function will be called to receive batches to process. May be None for sink instances used as parallel workers, which only ever process batches.
query (str) – Cypher to write the query to Neo4j. Data will be passed as batch parameter. Therefore, the query should start with a UNWIND $batch AS row.

get_batch(max_batch_size)[source]

Run the Cypher query for each incoming batch. :type max_batch_size: int :param max_batch_size: The maximum batch size to use when requesting from predecessor.

Return type:: Generator[BatchResults]
Returns:: Generator[BatchResults, None, None]
Parameters:: max_batch_size (int)