Custom Entities

RTV can be extended by custom user entities (classes) to provide missing functionality for user validation scenario (e.g. implementing some custom error metrics) or extend supported configuration file formats.

Implementig and registering

Custom entities should implement pre-defiened framework’s interfaces.

from pydantic import BaseModel

from rtv.core.base import BaseEntity
from rtv.transformer.interfaces import ITransformer

class MyAwesomeTransformer(BaseEntity, ITransformer, idf="awesome"):
    class Params(BaseModel):
        my_awesome_param: int
        ...

    def transform(self, data):
        ...

Usage of nested Params(BaseModel) class provides auto-validation for your entitiy’s parameters.

Extending from Base class is not mandatory if you do not want to reuse any common behaviour you can omit this.

All interface classes provided by the framework start with I followed by the type of the entitiy. It is important to inherit from the interface class when realizing your custom entities because interfaces trigger some under the hood mechanisms that allow you later to use your entitiy via configuration files.

NOTE:idf is optional for almost all entities, and most probably not needed if users intend to use the entity in python script, however there are some exceptions (will be mentioned below). It is just an alias to make config files more concise.

Following table shows which types you can implement and from which classes you should inherit in order to do that:

Entity Name	Inherit from
Reader	BaseReader, IReader
Transformer	BaseEntity, ITransformer
Validation	BaseEntity, IValidation
Validation Strategy	BaseValidationStrategy, IValidationStrategy
Action *	BaseAction, IAction
Writer	BaseWriter, IWriter
Config Loader **	IConfigLoader, idf=”<extension_suffix>”

* - See implementing custom actions

** - See implementing custom config loaders

More details on mentioned interfaces can be found in this section.

Implementing custom actions

When implementing custom actions we recommend to add a short and descriptive identifier:

# ...
class MyCustomAction(BaseAction, IAction, idf="greet"):
     class Params(BaseModel):
         message: str

     def execute(self):
         print(self.message)
# ...

That would make it more convenient to use in the configuration files.

actions:
  - greet:
      message: "Hello World!"

To say more, custom actions implementation only makes sense for usage in configuration files.

Implementing custom config loaders

IConfigLoader is a special case, this class is not inheriting from BaseEntity, does not support nested Params class, and requires idf to be the same as file extension suffix:

class TxtConfigLoader(IConfigLoader, idf="txt"):
    ...

Otherwise it should crash the run.

Registering for use in config files

The framework will automatically handle the addition of this custom class to the registry, and it will become available for use in the config files.

However, the framework needs to know where to look for the custom code. So, users need to set up an environment variable RTV_USER_CODE_PATH:

export RTV_USER_CODE_PATH=<custom_code_directory_path>

Substitute <custom_code_directory_path with an actual path on your file system where you gonna store the custom code for RTV. You can structure and and name those files as you want.

Defining custom entities in config

YAML Config Example:

• Using custom class name:

  definitions:
      - name: my_awesome_transformer
        class: MyAwesomeTransformer
        my_awesome_param: 42
  # ...
  

• Using idf (identifier/alias):

  definitions:
      - name: my_awesome_transformer
        class: awesome
        my_awesome_param: 42
  # ...
  

NOTE: Custom actions should not be defined, just used by alias instead.

Using custom entities in actions

YAML Config Example:

• Custom transformer:

  actions:
      - transform:
          input: data
          output_name: transformed_data
          transformers: my_awesome_transformer
          # ...
  

• Custom action:

  actions:
      - my_awesome_action:
          awesome_parameter: 42
          # ...
  

Core Interfaces

class IReader(Interface):
    @abstractmethod
    def read(
        self, *args, **kwargs
    ) -> DataCollection:
        ...

class IWriter(Interface):
    @abstractmethod
    def write(self,
        data: DataCollection,
        *args, **kwargs
    ) -> None:
        ...

class ITransformer(Interface):
    @abstractmethod
    def transform(
        self, collection: DataCollection, **kwargs
    ) -> DataCollection:
        ...

class IValidation(Interface):
    @abstractmethod
    def execute(
        self, reference: DataCollection, target: DataCollection
    ) -> List[Tuple[str, ValidationResultModel]]:
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        ...

class IValidationStrategy(Interface):
    @abstractmethod
    def validate(
        self,
        key: str,
        reference: DataCollection,
        target: DataCollection,
    ) -> ValidationResultModel:
        ...

    @property
    @abstractmethod
    def details(self) -> StrategyDetailsModel:
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        ...

class IAction(Interface):
    @abstractmethod
    def execute(self) -> None:
        ...

class IConfigLoader(Interface):
    @abstractmethod
    def load(self, config_path: str) -> Dict[str, Any]:
        ...