Pipelines#

class papercast.pipelines.Pipeline(name: str)[source]#

Bases: object

add_processor(name: str, processor: papercast.base.BasePipelineComponent)[source]#

Adds a processor to the pipeline.

Parameters
Raises

ValueError – If a processor with the given name already exists in the pipeline.

connect(a_name: str, a_output: str, b_name: str, b_input: str)[source]#

Connects two processors in the pipeline.

Parameters

  • a_name (str) – The name of the first processor.

  • a_output (str) – The name of the output type of the first processor.

  • b_name (str) – The name of the second processor.

  • b_input (str) – The name of the input type of the second processor.

Raises

TypeError – If the output type of the first processor does not match the input type of the second processor.

get_downstream_processors(collector_subscriber_name: str) → Iterable[str][source]#

Get all processors downstream of the collector with name collector_name by recursively traversing the graph of connections.

Parameters

collector_subscriber_name (str) – The name of the collector or subscriber.

Returns

A list of processor names.

Return type

Iterable[str]

property input_types: Dict[str, Any]#

Returns a dictionary of input types for all collectors in the pipeline.

Returns

A dictionary of input types, where the keys are input names and the values are input types.

Return type

Dict[str, Any]

process(production: papercast.production.Production, collector_subscriber_name: str, **options) → None[source]#

Run the pipeline synchronously on a production, from a collector or subscriber.

run(**kwargs)[source]#

Run the pipeline synchronously, from kwargs.