How to get the list of expectations supported by a datasource?

Archive
1

Hi,

We are looking for a function to programmatically fetch the list of expectations.
Even better would be a way to fetch only the expectations supported by a given datasource.

Maybe this exists already and is used to generate this table of the Documentation :books:!?
https://docs.greatexpectations.io/en/latest/reference/core_concepts/expectations/implemented_expectations.html#implemented-expectations

Cheers!

2

Not an expert, but I’ve been toying with a similar issue. If you’re in pre-v0.13 territory, you can create a batch and then use the inspect python module module to see all of the available expect_whatever() functions (and their arguments) using the getmembers() function.

In v0.13+ land, I think you can do the same but with a validator, though getting the specific arguments can be a little harder since the functions are now more generic. Haven’t quite cracked that part yet

1 Like
3

The table of implemented Expectations by backend is currently maintained manually.

The direction that Ryan is pointing is correct, but we don’t have an official code snippet for it.

1 Like
4

I found a way to list the expectations available for a specific datasource.

The function below takes a Batch object as argument that you can easily create for a specific datasource (cf. Documentation) and returns detailed information about all expectations supported by this datasource which have been found by inspecting the code.

Hoping it will help others and maybe to update automatically this part of the documentation.

import inspect
from importlib import import_module
from typing import Union, List
from great_expectations.core.batch import Batch
from great_expectations.dataset import Dataset

def get_list_expectations(batch: Union[Batch, DataAsset]) -> List[dict]:
    expectations = []
    list_names_expectations = [
        u for (u, v) in inspect.getmembers(batch) if (u.startswith("expect_"))
    ]
    entries_to_remove_from_args = ["batch_id", "table", "row_condition", "condition_parser"]
    entries_to_remove_from_kwargs = [
        "include_config",
        "catch_exceptions",
        "result_format",
        "meta",
        "row_condition",
        "condition_parser",
    ]

    for exp_name in list_names_expectations:
        class_name = "".join(map(lambda x: x.capitalize(), exp_name.split("_")))
        try:
            mod = import_module(f"great_expectations.expectations.core.{exp_name}")
            inst = getattr(mod, class_name)()
            dct_kwargs = inst.default_kwarg_values
            tpl_args = inst.domain_keys

            list_args = list(tpl_args)
            for entry in entries_to_remove_from_args:
                if entry in list_args:
                    list_args.remove(entry)
            tpl_args = tuple(list_args)
            for entry in entries_to_remove_from_kwargs:
                if entry in dct_kwargs:
                    del dct_kwargs[entry]
            expectations.append(
                {
                    "expectation_type": exp_name,
                    "expectation_args": tpl_args,
                    "expectation_kwargs": dct_kwargs,
                    "expectation_success_keys": inst.success_keys,
                    "expectation_doc": inst.__doc__,
                }
            )
        except Exception as e:
            print(e)
    return expectations