Create Pipeline Steps that require implementing only handler methods

If a pipeline step only needs to implement handler methods, then you can inherit from the ForceHandleMixin as demonstrated here. Handler methods are useful when :

  • You need to change the shape of the data container passed to the following steps, or the wrapped steps.

  • You want to apply side effects based on the data container, and the execution context.

  • You want to change the pipeline execution flow based on the data container, and the execution context.

import numpy as np

from neuraxle.base import BaseStep, DataContainer, ExecutionContext, ForceHandleMixin


class ForceHandleMixinStep(ForceHandleMixin, BaseStep):
    """
    Please make your steps inherit from ForceHandleMixin when they only implement handle_methods, but also
    when you want to make impossible the use of regular fit, transform, and fit_transform methods
    Also, make sure that BaseStep is the last step you inherit from.
    """

    def __init__(self):
        ForceHandleMixin.__init__(self)
        BaseStep.__init__(self)

    def handle_fit(self, data_container: DataContainer, context: ExecutionContext):
        """
        Change the shape of the data container.
        and/or
        Apply any side effects based on the data container
        And/or
        Change the execution flow of the pipeline
        """
        data_container = self.hash_data_container(data_container)
        return self

    def handle_transform(self, data_container: DataContainer, context: ExecutionContext):
        """
        Change the shape of the data container.
        and/or
        Apply any side effects based on the data container
        And/or
        Change the execution flow of the pipeline
        """
        data_container = self.hash_data_container(data_container)
        return data_container

    def handle_fit_transform(self, data_container: DataContainer, context: ExecutionContext):
        """
        Change the shape of the data container.
        and/or
        Apply any side effects based on the data container
        And/or
        Change the execution flow of the pipeline
        """
        data_container = self.hash_data_container(data_container)
        return self, data_container


def main():
    p = ForceHandleMixinStep()
    data_inputs = np.array([0, 1])
    expected_outputs = np.array([0, 1])

    p = p.fit(data_inputs, expected_outputs)
    outputs = p.transform(data_inputs)


if __name__ == '__main__':
    main()

Total running time of the script: ( 0 minutes 0.041 seconds)

Gallery generated by Sphinx-Gallery