Custom Transformations - VladimirSiv/pytransflow GitHub Wiki

pytransflow comes with a plethora of predefined and highly customizable transformations, for example:

• add_field - Adds a field to the record
• validate - Validates and imposes a schema on a record
• prefix - Adds a prefix to the field's value
• postfix - Adds a postfix to the field's value
• regex_extract - Extracts a string using regex pattern
• remove_fields - Removes one or multiple fields from the record
• And many more...

However, you can also define your own transformations by following the blueprint for creating transformations and including them in the TransformationCatalogue.

TransformationCatalogue contains mappings of transformation names to their definition class Transformation and transformation schema class TransformationSchema.

To include a new transformation to the TransformationCatalogue, use the following

from pytransflow.core import TransformationCatalogue

TransformationCatalogue.add_transformation(
    transformation_name="<name>",
    transformation=<Transformation>,
    schema=<TransformationSchema>,
)

The <name> defines the name that will be used in the YAML configuration files to identify the transformation. Transformation is the implementation of the transformation, it has to implement transform method which operates on a record. TransformationSchema is the schema configuration of the transformation which will be used to parse the configuration from YAML files.

You can override built-in transformations by defining your own implementation and registering them under the same name in the TransformationCatalogue.

Let's define a transformation that operates on a single field, capitalizing all characters if it's a string; otherwise, it will raise the FieldWrongTypeException exception. Since this operation is the same as upper() we will call it Upper.

Let's define UpperSchema, it has two arguments:

• field - Name of the field which contains the value
• output - Name of the field where we want to store the transformed value. If output is not specified, overwrite the field.

The UpperSchema has to inherit from TransformationSchema.

from typing import Optional
from typing_extensions import Self
from pydantic import Field, model_validator
from pytransflow.core import TransformationSchema


class UpperSchema(TransformationSchema):
    """Implements Upper Transformation Schema"""

    field: str = Field(
        title="Field",
        description="Input field",
        required_in_record=True,
    )
    output: Optional[str] = Field(
        default=None,
        title="Output field",
        description="Output field where the processed data will be stored",
        output_field=True,
    )

    @model_validator(mode="after")
    def set_output(self) -> Self:
        if self.output is None:
            self.output = self.field
            self.ignore_errors.append("output_already_exists")
        self.set_dynamic_fields()
        return self

Notes:

• If the field is required in the record, use required_in_record=True inside Field. This will be used to check the existence of the field in a record before actual invocation of the transformation
• If the field is output of the transformation, please use output_field=True. This will be used to check if output field already exists in the record and if it should be overwritten
• If the output is the same as input field, we should add output_already_exists in ignore_errors, otherwise the transformation will always throw OutputAlreadyExistsException exception
• Always call self.set_dynamic_fields() at the end of model_validator. This method parses fields and looks for additional arguments like required_in_record, output_field, etc, and sets required parameters.

The pydantic's model_validator(mode="after") takes care of setting output if it's not defined. Likewise, you can use any other pydantic feature to manipulate and define transformation schemas.

The actual transformation should take the original value, check its type, and transform it if the value is str, otherwise throw FieldWrongTypeException exception.

The Upper class has to inherit from pytransflow Transformation base class, and implement transform method.

from pytransflow.core import Transformation, Record,
from pytransflow.exceptions import FieldWrongTypeException


class Upper(Transformation):
    """Implements Upper transformation logic

    Upper Transformation transforms a record by applying `upper` function to a
    specified field value and output is to defined `output` field

    """

    def transform(
        self,
        record: Record,
    ) -> Record:
        output_field = self.config.schema.output
        field = self.config.schema.field

        initial_value = record[field]

        if not isinstance(initial_value, str):
            raise FieldWrongTypeException(
                field,
                type(record[field]),
                "str",
            )

        record[output_field] = initial_value.upper()

        return record

Now, we can register this transformation in TransformationCatalogue

from pytransflow.core import TransformationCatalogue


TransformationCatalogue.add_transformation(
    transformation_name="upper",
    transformation=Upper,
    schema=UpperSchema,
)

and use in a flow

transformations:
  - add_field:
      name: a
      value: lower
  - upper:
      field: a

Running this flow with an empty record, we get

Result: {'default': [{'a': 'LOWER'}]}

Besides creating a custom transformation, you can also create custom exceptions that can be used in ignore_errors arguments.

Let's say that you created a CustomTransformation transformation and you want to raise CustomException if something goes wrong. All you have to do is to inherit from TransformationBaseException and define name class attribute in order for pytransflow to handle the exception with all the features it provides.

from pytransflow.exceptions import TransformationBaseException


class CustomException(TransformationBaseException):

    name = "custom_exception"

    def __init__(self, error) -> None:
        super().__init__(f"Error: {error}")

Here, we are defining name which can be used in flow configuration:

transformations:
  - custom_transformation:
      field: a
      ignore_errors:
        - custom_exception