data_validation_framework.task

Specific tasks.

Functions

SkippableMixin([default_value])

Create a mixin class to add a skip parameter.

Classes

BaseValidationTask(*args, **kwargs)

Base luigi task used for validation steps.

ElementValidationTask(*args, **kwargs)

A class to validate each element of a data set without considering the other elements.

InputParameters(col_mapping, **kwargs)

A helper to build task inputs.

SetValidationTask(*args, **kwargs)

A class to validate an entire data set (useful for global properties).

TagResultOutputMixin(*args, **kwargs)

Initialize target prefixes and optionally add a tag to the resut directory.

ValidationWorkflow(*args, **kwargs)

Class to define and process a validation workflow.

Exceptions

ValidationError

Exception raised if the validation could not be performed properly.

class data_validation_framework.task.BaseValidationTask(*args, **kwargs)

Bases: luigi_tools.task.LogTargetMixin, luigi_tools.task.RerunMixin, data_validation_framework.task.TagResultOutputMixin, luigi.task.Task

Base luigi task used for validation steps.

Warning

Only use child of this class such as ElementValidationTask or SetValidationTask.

To provide a validation function, use it as:

class Task(ChildBaseValidationTask):
    data_dir = luigi.Parameter(default='my_data')
    validation_function = staticmethod(my_validation_function)
    output_columns = {'my_new_col': None}

    def inputs(self):
        return {PreviousTask: {"input_data", "input_data"}

Note

The dataset_df and result_path attributes are passed to the requirements when they have None values for these attributes.

args()

List of addititonal arguments to provide to validation_function().

This method can be overridden to pass these arguments to the validation_function() and to the pre_process() and post_process() methods.

static check_inputs(inputs)

Check a given dict of inputs.

extra_input()

Targets of the tasks given to extra_requires().

extra_requires()

Requirements that should not be considered as validation tasks.

static filter_columns(all_dfs, mappings)

Filter the columns of the given DataFrames according to a given mapping.

gather_inputs = False

Gather the inputs in the pandas.DataFrame given to the validation_function().

Type

bool

inputs()

Information about required input data.

This method can be overridden and should return a dict of one of the following forms:

{<task_name>: {"<input_column_name>": "<current_column_name>"}}
{
    <task_name>: (
        {"<input_column_name>": "<current_column_name>"},
        {
            "<kwarg_1>": "<value_1>",
            "<kwarg_2>": "<value_2>",
        }
    )
}
{
    <task_name>: InputParameters(
        {"<input_column_name>": "<current_column_name>"},
        kwarg_1="<value_1>",
        kwarg_2="<value_2>",
    )
}
where:
  • <task_name> is the name of the required task,

  • <input_column_name> is the name of the column we need from the report of task_name,

  • <current_column_name> is the name of the same column in the current task.

  • <kwarg_*> is the name of a keyword argument passed to the constructor of the required task.

kwargs()

Dict of addititonal keyword arguments to provide to validation_function().

This method can be overridden to pass these keyword arguments to the validation_function() and to the pre_process() and post_process() methods.

nb_total = None

Total number of processed elements.

Type

int

nb_valid = None

Total number of valid elements.

Type

int

output()

The targets of the current task.

output_columns = None

A dict with names as keys and empty values as values for new columns created by the current task.

Type

dict

post_process(df, args, kwargs)

Method executed after applying the external function.

pre_process(df, args, kwargs)

Method executed before applying the external function.

processed_inputs()

Process the inputs to automatically propagate the values from the workflow.

read_dataset()

Import the dataset to a pandas.DataFrame.

This method can be overridden to load custom data (e.g. GeoDataFrame, etc.). The dataset should always be loaded from the path given by self.dataset_df.

requires()

Process the inputs to generate the requirements.

results = None

The results of the validation_function().

Type

pandas.DataFrame

run()

The main process of the current task.

property task_name

The name of the task.

static validation_function(*args, **kwargs)

The validation function to apply to the current data set.

class data_validation_framework.task.ElementValidationTask(*args, **kwargs)

Bases: data_validation_framework.task.BaseValidationTask

A class to validate each element of a data set without considering the other elements.

The validation_function will receive the row and the output_path as first arguments.

class data_validation_framework.task.InputParameters(col_mapping, **kwargs)

Bases: object

A helper to build task inputs.

This class should only be used to build the input dictionaries:

class Task(ChildBaseValidationTask):
    data_dir = luigi.Parameter(default='my_data')
    validation_function = staticmethod(my_validation_function)
    output_columns = {'my_new_col': None}

    def inputs(self):
        return {
            PreviousTask: InputParameters(
                {"input_data", "input_data"},
                kwarg_1="value_1",
                kwarg_2="value_2"
            )
        }
Parameters

col_mapping (dict) – The column mapping.

Keyword Arguments

kwargs – All the keyword arguments passed to the constructor.

Warning

The keyword arguments will always override the arguments from the config file.

class data_validation_framework.task.SetValidationTask(*args, **kwargs)

Bases: data_validation_framework.task.BaseValidationTask

A class to validate an entire data set (useful for global properties).

The validation_function will receive the DataFrame and the output_path as first arguments.

Note

The given pandas.DataFrame will always have the columns ["is_valid", "ret_code", "comment", "exception"]. The validation_function should at least update the values for the is_valid column.

data_validation_framework.task.SkippableMixin(default_value=False)

Create a mixin class to add a skip parameter.

This mixin must be applied to a data_validation_framework.ElementValidationTask. It will create a skip parameter and wrap the validation function to just skip it if the skip argument is set to True. If skipped, it will keep the is_valid values as is and add a specific comment to inform the user.

Parameters

default_value (bool) – The default value for the skip argument.

class data_validation_framework.task.TagResultOutputMixin(*args, **kwargs)

Bases: object

Initialize target prefixes and optionally add a tag to the resut directory.

Warning

If this mixin is used alongside the luigi_tools.task.RerunMixin, then it can have two different behaviors:

  • if placed on the right side of the RerunMixin, a new tag is created and thus rerun does not do anything.

  • if placed on the left side of the RerunMixin, the targets from the tagged directory without conflict are removed then a new tagged directory with conflict is created.

target_cls

alias of data_validation_framework.target.TaggedOutputLocalTarget

exception data_validation_framework.task.ValidationError

Bases: Exception

Exception raised if the validation could not be performed properly.

class data_validation_framework.task.ValidationWorkflow(*args, **kwargs)

Bases: data_validation_framework.task.SetValidationTask

Class to define and process a validation workflow.

static validation_function(*args, **kwargs)

The validation function to apply to the current data set .

This method should usually do nothing for ValidationWorkflow as this class is only supposed to gather validation steps.