API Reference

thecodecrate_pipeline

Pipeline

Pipeline(
    processor_class: Optional[
        type[ProcessorInterface]
    ] = None,
    processor_instance: Optional[ProcessorInterface] = None,
    processor: Optional[
        type[ProcessorInterface] | ProcessorInterface
    ] = None,
    *args: Any,
    **kwds: Any,
)

Pipeline Class

processor_class `instance-attribute`

processor_class: Optional[
    type[ProcessorInterface[T_in, T_out]]
]

processor_instance `instance-attribute`

processor_instance: Optional[
    ProcessorInterface[T_in, T_out]
]

stage_instances `instance-attribute`

stage_instances: StageInstanceCollection

stages `instance-attribute`

stages: StageCollection

clone

clone(attributes: dict[str, Any]) -> Self

get_processor_class

get_processor_class() -> Optional[type[ProcessorInterface]]

get_processor_instance

get_processor_instance() -> Optional[ProcessorInterface]

pipe

pipe(stage: StageInstance) -> Self

Adds a single stage to the pipeline.

process `async`

process(payload: T_in, *args: Any, **kwds: Any) -> T_out

Process the given payload through the pipeline.

with_processor

with_processor(
    processor: (
        type[ProcessorInterface] | ProcessorInterface
    ),
) -> Self

Attachs a processor (class or instance) to the pipeline.

with_processor_class

with_processor_class(
    processor_class: type[ProcessorInterface],
) -> Self

Attachs a processor class to the pipeline.

with_processor_instance

with_processor_instance(
    processor_instance: ProcessorInterface,
) -> Self

Attachs a processor instance to the pipeline.

with_stages

with_stages(stages: StageCollection) -> Self

Adds a collection of stages to the pipeline.

PipelineFactory

PipelineFactory(
    processor_class: Optional[
        type[ProcessorInterface]
    ] = None,
    processor_instance: Optional[ProcessorInterface] = None,
    processor: Optional[
        type[ProcessorInterface] | ProcessorInterface
    ] = None,
    *args: Any,
    **kwds: Any,
)

pipeline_class `instance-attribute`

pipeline_class: Optional[type[TPipeline]]

processor_class `class-attribute` `instance-attribute`

processor_class: Optional[
    type[ProcessorInterface[T_in, T_out]]
] = None

processor_instance `class-attribute` `instance-attribute`

processor_instance: Optional[
    ProcessorInterface[T_in, T_out]
] = None

stages `instance-attribute`

stages: StageCollection

add_stage

add_stage(stage: StageClassOrInstance) -> Self

Adds a single stage to the pipeline.

make

make() -> TModel

with_processor

with_processor(
    processor: (
        type[ProcessorInterface] | ProcessorInterface
    ),
) -> Self

Attachs a processor (class or instance) to the pipeline factory.

with_processor_class

with_processor_class(
    processor_class: type[ProcessorInterface],
) -> Self

Attachs a processor class to the pipeline factory.

with_processor_instance

with_processor_instance(
    processor_instance: ProcessorInterface,
) -> Self

Attachs a processor instance to the pipeline factory.

with_stages

with_stages(stages: StageCollection) -> Self

Adds a collection of stages to the pipeline.

PipelineFactoryInterface

PipelineFactoryInterface(
    processor_class: Optional[
        type[ProcessorInterface]
    ] = None,
    processor_instance: Optional[ProcessorInterface] = None,
    processor: Optional[
        type[ProcessorInterface] | ProcessorInterface
    ] = None,
    *args: Any,
    **kwds: Any,
)

add_stage

add_stage(stage: StageClassOrInstance) -> Self

make

make() -> Any

with_processor

with_processor(
    processor: (
        type[ProcessorInterface] | ProcessorInterface
    ),
) -> Self

with_processor_class

with_processor_class(
    processor_class: type[ProcessorInterface],
) -> Self

with_processor_instance

with_processor_instance(
    processor_instance: ProcessorInterface,
) -> Self

with_stages

with_stages(stages: StageCollection) -> Self

PipelineInterface

PipelineInterface(
    processor_class: Optional[
        type[ProcessorInterface]
    ] = None,
    processor_instance: Optional[ProcessorInterface] = None,
    processor: Optional[
        type[ProcessorInterface] | ProcessorInterface
    ] = None,
    *args: Any,
    **kwds: Any,
)

clone

clone(attributes: dict[str, Any]) -> Self

get_processor_class

get_processor_class() -> Optional[type[ProcessorInterface]]

get_processor_instance

get_processor_instance() -> Optional[ProcessorInterface]

pipe

pipe(stage: StageInstance) -> Self

process `async`

process(payload: T_in, *args: Any, **kwds: Any) -> T_out

with_processor

with_processor(
    processor: (
        type[ProcessorInterface] | ProcessorInterface
    ),
) -> Self

with_processor_class

with_processor_class(
    processor_class: type[ProcessorInterface],
) -> Self

with_processor_instance

with_processor_instance(
    processor_instance: ProcessorInterface,
) -> Self

with_stages

with_stages(stages: StageCollection) -> Self

Processor

Processor(*args: Any, **kwds: Any)

Processor Class

clone

clone(attributes: dict[str, Any]) -> Self

process `abstractmethod` `async`

process(
    payload: T_in,
    stages: StageInstanceCollection,
    *args: Any,
    **kwds: Any,
) -> T_out

Process the given payload through the provided stages.

Parameters:

payload
(T_in) –

The input payload to process.
stages
(StageInstanceCollection) –

The collection of stages to process the payload through.
*args
(Any, default: () ) –

Additional positional arguments.
**kwds
(Any, default: {} ) –

Additional keyword arguments.

Returns:

T_out ( T_out ) –

The processed output.

ProcessorInterface

ProcessorInterface(*args: Any, **kwds: Any)

clone

clone(attributes: dict[str, Any]) -> Self

process `abstractmethod` `async`

process(
    payload: T_in,
    stages: StageInstanceCollection,
    *args: Any,
    **kwds: Any,
) -> T_out