Getting Started - VladimirSiv/pytransflow GitHub Wiki

pytransflow is available on PyPI at pytransflow.

pip install pytransflow

After installing pytransflow, create the flows directory in the root of your project, in it create a file test_flow.yaml:

Note: You can also use .yml, both are supported.

description: Simple Flow
transformations:
  - add_field:
      name: new_field
      value: [1, 2, 3]

Then create a python script in the root folder:

from pytransflow.core import Flow


records = [{"a": 1}]
flow = Flow("test_flow")
flow.process(records)

print(f"Result: {flow.datasets}")
print(f"Failed Records: {flow.failed_records}")

Note: name parameter matches the filename of .yaml

The output is:

Result: {'default': [{'a': 1, 'new_field': [1, 2, 3]}]}
Failed Records: []

Path separator defines how nested fields are defined and accessed. The default path separator is /.

If we use the previous example, but use nested name for add_field

transformations:
  - add_field:
      name: b/c/d
      value: [1, 2, 3]

You'll get

Result: {'default': [{'a': 1, 'b': {'c': {'d': [1, 2, 3]}}}]}

Default path separator value can be easily changed in the global configuration file - Please see Global Configuration, or defined for a specific flow using path_separator configuration.

For example

path_separator: ","
transformations:
  - add_field:
      name: b,c,d
      value: [1, 2, 3]

Give the same result as previous example

Result: {'default': [{'a': 1, 'b': {'c': {'d': [1, 2, 3]}}}]}

Conditions can be applied on two levels:

• Transformation - All transformations can have a condition parameter. This condition is evaluated before the transformation is executed. If condition is met, the transformation will be applied, if condition is not met, the whole record will be skipped and included in the output_datasets.
• Output Dataset - Transformation output_datasets can contain a condition. If condition is met, the record will be included in the output dataset, if not it will be skipped. This is evaluated after the transformation is applied, not before. For more information, please see Datasets - Using Conditions

The following two sections showcase how we can apply a condition on transformation level.

Let's look at the following example:

transformations:
  - add_field:
      name: b
      value: 1
      condition: "@a == 'A'"

The character @ is used to define a record field.

Note: pytransflow also allows defining variables on a flow level, which can be used in conditions using !:. For more information, please see Flow Variables

Here, we are setting a condition that field a in record has to have a value equal to A. If that's the case, the record will be processed, otherwise the record will be skipped

records = [
  {"a": "A"},
  {"a": "a"},
]
flow = Flow("<filename>")
flow.process(records)

The output dataset is:

{'default': [{'a': 'A', 'b': 1}, {'a': 'a'}]}

As we can see, only the first record is processed, the second is skipped because the condition for processing is not met.

Condition expression can be extended to use nested field and evaluation functions.

For example:

transformations:
  - add_field:
      name: b/e/f
      value: 1
      condition: "int(@a/b/c) * 2 == @d/e"

Using the following input record

records = [{"a": {"b": {"c": "2"}}, "d": {"e": 4}}]
flow = Flow("<filename>")
flow.process(records)

We get the resulting dataset

{
  'default': [
    {
      'a': {'b': {'c': '2'}},
      'd': {'e': 4},
      'b': {'e': {'f': 1}}
    }
  ]
}

Since the condition is met, the field b/e/f is added, as we can see in the resulting dataset.

pytransflow supports creating custom evaluation function, which you can use in all condition parameters. Please see Custom-Evaluation for more information on how to achieve that.

Dataset is a collection of records and each transformation has input_datasets and output_datasets parameters:

• input_datasets - list of dataset names from which the records will be extracted for processing.
• output_datasets - list of dataset names, along with optional conditions, to which we will direct processed records as output.

Default behavior:

• If input_datasets is not specified it will default to ["default"]. This name can be configured check Global Configuration
• If output_datasets is not specified it will be the same as input_datasets

Let's look at an example of 3 transformations and 4 datasets:

transformations:
  - add_field:
      name: a
      value: test
      output_datasets:
        - dataset-1
  - prefix:
      field: a
      value: pre_
      input_datasets:
        - dataset-1
      output_datasets:
        - dataset-2
  - postfix:
      field: a
      value: _post
      input_datasets:
        - dataset-2
      output_datasets:
        - dataset-3
        - dataset-4

Running this flow with an empty record

records = [{}]
flow = Flow("<filename>")
flow.process(records)

Gives the following result

{'dataset-3': [{'a': 'pre_test_post'}],
 'dataset-4': [{'a': 'pre_test_post'}]}

Additionally, output_datasets can contain a condition, which can be defined in two ways.

Using <dataset-name>: <condition> pattern:

- postfix:
    field: a
    value: b
    output_datasets:
      - dataset: "@a == 2"

Or name, condition dictionary:

- postfix:
    field: a
    value: b
    output_datasets:
      - name: dataset
        condition: "@a == 2"

Extending the previous example by adding condition to the last dataset:

transformations:
  - add_field:
      name: a
      value: test
      output_datasets:
        - dataset-1
  - prefix:
      field: a
      value: pre_
      input_datasets:
        - dataset-1
      output_datasets:
        - dataset-2
  - postfix:
      field: a
      value: _post
      input_datasets:
        - dataset-2
      output_datasets:
        - dataset-3
        - dataset-4: "@b == '1'"

Since there is no field b in the record, this condition will fail and final dataset includes only dataset-3.

{'dataset-3': [{'a': 'pre_test_post'}]}

If some error happens during the processing, the flow won't immediately fail.

Note: This can be changed by setting instant_fail: True. For more information see: Instant Fail

The records that failed during processing are stored in the failed records dataset, which can be accessed on a Flow instance after processing.

from pytransflow.core import Flow


records = [{"a": 1}]
flow = Flow("test_flow")
flow.process(records)

print(f"Failed Records: {flow.failed_records}")

Failed record contains information about the initial record, the failed record, transformation that failed, and the error that was raised.

Note: There is a difference between initial record and failed record, for more information please see How It Works

This is especially useful for debugging but also for redirecting failed records during actual processing.

If the error is ignored using ignore_errors, the record won't end up as a failed record, even thought the error occurred. Ignore errors configuration is primarily used to handle errors that are expected and can be safely ignored, these errors won't impact the processing, since the record will remain unprocessed and redirected to output datasets.

Ignore errors configuration is a way to handle errors that are expected and can be safely ignored. These errors won't impact the processing i.e. the flow won't fail. The ignore_errors is set on a transformation level, if the transformation encounters the ignored error during the processing, the result will be

• Initial record - No processing will occur and transformation just returns initial record
• Processed record - Error is ignored, but the record is processed either way. For example, if transformation wants to store a value in field x but that field already exists it will raise OutputFieldAlreadyExists exception. However, if it's ignored, the output field will be overwritten and the end result is a processed record, not the initial record.

Let's illustrate that using the following examples.

transformations:
  - add_field:
      name: class
      value: 1

If we have an initial record {"class": "a"}, this flow will result in

Result: {}
Failed Records: [
  {
    ..., 
    error=OutputAlreadyExistsException(Output field 'class' already exists))], 
    'record': {'class': 'a'},
    ...
  }
]

However, when we add ignore_errors for OutputAlreadyExistsException

transformations:
  - add_field:
      name: class
      value: 1
      ignore_errors:
        - output_already_exists

we get

Result: {'default': [{'class': 1}]}
Failed Records: []

The next example showcases that ignoring an error returns initial record that's redirected to an output dataset for further processing.

transformations:
  - add_field:
      name: class
      value: 1
      output_datasets:
        - added
      ignore_errors:
        - output_already_exists
  - prefix:
      field: class
      value: pre_
      ignore_errors:
        - field_wrong_type
      input_datasets:
        - added
      output_datasets:
        - prefix

As in previous example, we are ignoring output_already_exists and overwriting the field, this transformed record is then routed to added dataset, which is used for prefix transformation. Since the record will be {"class": 1} i.e. it's value is int the prefix should fail but we are ignoring field_wrong_type, and the end result is

Result: {'prefix': [{'class': 1}]}
Failed Records: []

Notice that the end result dataset is prefix, which is the output of prefix.

pytransflow allows users to validate a record schema by utilizing validate transformation which takes one argument, schema_name.

transformations:
  - add_field:
      name: a
      value: b
  - validate:
      schema_name: test.TestSchema

The input format for schema_name is <name_of_the_file>.<name_of_the_schema_class>.

The file is stored under SCHEMA_PATH which can be defined in pyproject.toml or .pytransflowrc, but defaults to <project root>/schema.

For more information about configuring SCHEMA_PATH, please see Global Configuration

Schema must inherit from pydantic's BaseModel. This allows user to leverage all advanced features that pydantic offers, and can be highly customizable to suit any particular needs.

Schema is imposed on a record, meaning, unnecessary fields will be removed, field types will be cast, and schema validated.

Example of a schema

# test.py
from pydantic import BaseModel
from typing import Optional


class TestSchema(BaseModel):
    """Example schema"""

    first_name: str
    last_name: str
    middle_name: Optional[str]
    birth_year: Optional[int]
    ...

Since the filename is test.py, and class TestSchema, we will use it as test.TestSchema in validate transformation.