great_expectations.self_check.util

Module Contents

Classes

SqlAlchemyConnectionManager()
LockingConnectionCheck(sa, connection_string)

Functions

get_sqlite_connection_url(sqlite_db_path)
get_dataset(dataset_type, data, schemas=None, profiler=ColumnsExistProfiler, caching=True, table_name=None, sqlite_db_path=None)	Utility to create datasets for json-formatted tests
get_test_validator_with_data(execution_engine, data, schemas=None, profiler=ColumnsExistProfiler, caching=True, table_name=None, sqlite_db_path=None)	Utility to create datasets for json-formatted tests.
build_pandas_validator_with_data(df: pd.DataFrame, batch_definition: Optional[BatchDefinition] = None)
build_sa_validator_with_data(df, sa_engine_name, schemas=None, caching=True, table_name=None, sqlite_db_path=None, batch_definition: Optional[BatchDefinition] = None)
modify_locale(func)
build_spark_validator_with_data(df: Union[pd.DataFrame, SparkDataFrame], spark: SparkSession, batch_definition: Optional[BatchDefinition] = None)
build_pandas_engine(df: pd.DataFrame)
build_sa_engine(df: pd.DataFrame, sa: ModuleType, schema: Optional[str] = None, if_exists: str = ‘fail’, index: bool = False, dtype: Optional[dict] = None)
build_spark_engine(spark: SparkSession, df: Union[pd.DataFrame, SparkDataFrame], batch_id: Optional[str] = None, batch_definition: Optional[BatchDefinition] = None)
candidate_getter_is_on_temporary_notimplemented_list(context, getter)
candidate_test_is_on_temporary_notimplemented_list(context, expectation_type)
candidate_test_is_on_temporary_notimplemented_list_cfe(context, expectation_type)
build_test_backends_list(include_pandas=True, include_spark=True, include_sqlalchemy=True, include_sqlite=True, include_postgresql=False, include_mysql=False, include_mssql=False, include_bigquery=False, raise_exceptions_for_backends: bool = True)	Attempts to identify supported backends by checking which imports are available.
generate_expectation_tests(expectation_type: str, test_data_cases: List[ExpectationTestDataCases], execution_engine_diagnostics: ExpectationExecutionEngineDiagnostics, raise_exceptions_for_backends: bool = False)	param expectation_type snake_case name of the expectation type
should_we_generate_this_test(backend: str, expectation_test_case: ExpectationTestCase)
sort_unexpected_values(test_value_list, result_value_list)
evaluate_json_test(data_asset, expectation_type, test)	This method will evaluate the result of a test build using the Great Expectations json test format.
evaluate_json_test_cfe(validator, expectation_type, test)	This method will evaluate the result of a test build using the Great Expectations json test format.
check_json_test_result(test, result, data_asset=None)
generate_test_table_name(default_table_name_prefix: str = ‘test_data_’)
_create_bigquery_engine()
_bigquery_dataset()

great_expectations.self_check.util.expectationValidationResultSchema

great_expectations.self_check.util.expectationSuiteValidationResultSchema

great_expectations.self_check.util.expectationConfigurationSchema

great_expectations.self_check.util.expectationSuiteSchema

great_expectations.self_check.util.logger

great_expectations.self_check.util.tmp_dir

great_expectations.self_check.util.sqlalchemy

great_expectations.self_check.util.SparkSession

great_expectations.self_check.util.spark_DataFrame

great_expectations.self_check.util.SQLITE_TYPES

great_expectations.self_check.util._BIGQUERY_MODULE_NAME = sqlalchemy_bigquery

great_expectations.self_check.util.bigquery_types_tuple

great_expectations.self_check.util.POSTGRESQL_TYPES

great_expectations.self_check.util.MYSQL_TYPES

great_expectations.self_check.util.MSSQL_TYPES

great_expectations.self_check.util.SQL_DIALECT_NAMES = ['sqlite', 'postgresql', 'mysql', 'mssql', 'bigquery']

class great_expectations.self_check.util.SqlAlchemyConnectionManager

get_engine(self, connection_string)

great_expectations.self_check.util.connection_manager

class great_expectations.self_check.util.LockingConnectionCheck(sa, connection_string)

is_valid(self)

great_expectations.self_check.util.get_sqlite_connection_url(sqlite_db_path)

great_expectations.self_check.util.get_dataset(dataset_type, data, schemas=None, profiler=ColumnsExistProfiler, caching=True, table_name=None, sqlite_db_path=None)

Utility to create datasets for json-formatted tests

great_expectations.self_check.util.get_test_validator_with_data(execution_engine, data, schemas=None, profiler=ColumnsExistProfiler, caching=True, table_name=None, sqlite_db_path=None)

Utility to create datasets for json-formatted tests.

great_expectations.self_check.util.build_pandas_validator_with_data(df: pd.DataFrame, batch_definition: Optional[BatchDefinition] = None) → Validator

great_expectations.self_check.util.build_sa_validator_with_data(df, sa_engine_name, schemas=None, caching=True, table_name=None, sqlite_db_path=None, batch_definition: Optional[BatchDefinition] = None)

great_expectations.self_check.util.modify_locale(func)

great_expectations.self_check.util.build_spark_validator_with_data(df: Union[pd.DataFrame, SparkDataFrame], spark: SparkSession, batch_definition: Optional[BatchDefinition] = None) → Validator

great_expectations.self_check.util.build_pandas_engine(df: pd.DataFrame) → PandasExecutionEngine

great_expectations.self_check.util.build_sa_engine(df: pd.DataFrame, sa: ModuleType, schema: Optional[str] = None, if_exists: str = 'fail', index: bool = False, dtype: Optional[dict] = None) → SqlAlchemyExecutionEngine

great_expectations.self_check.util.build_spark_engine(spark: SparkSession, df: Union[pd.DataFrame, SparkDataFrame], batch_id: Optional[str] = None, batch_definition: Optional[BatchDefinition] = None) → SparkDFExecutionEngine

great_expectations.self_check.util.candidate_getter_is_on_temporary_notimplemented_list(context, getter)

great_expectations.self_check.util.candidate_test_is_on_temporary_notimplemented_list(context, expectation_type)

great_expectations.self_check.util.candidate_test_is_on_temporary_notimplemented_list_cfe(context, expectation_type)

great_expectations.self_check.util.build_test_backends_list(include_pandas=True, include_spark=True, include_sqlalchemy=True, include_sqlite=True, include_postgresql=False, include_mysql=False, include_mssql=False, include_bigquery=False, raise_exceptions_for_backends: bool = True) → List[str]

Attempts to identify supported backends by checking which imports are available.

great_expectations.self_check.util.generate_expectation_tests(expectation_type: str, test_data_cases: List[ExpectationTestDataCases], execution_engine_diagnostics: ExpectationExecutionEngineDiagnostics, raise_exceptions_for_backends: bool = False)

Parameters

• expectation_type – snake_case name of the expectation type

• examples_config – a dictionary that defines the data and test cases for the expectation

• expectation_execution_engines_dict –

  shows which backends/execution engines the expectation is implemented for. It can be obtained from the output of the expectation’s self_check method Example: {

  ”PandasExecutionEngine”: True, “SqlAlchemyExecutionEngine”: False, “SparkDFExecutionEngine”: False

  }

Returns

great_expectations.self_check.util.should_we_generate_this_test(backend: str, expectation_test_case: ExpectationTestCase)

great_expectations.self_check.util.sort_unexpected_values(test_value_list, result_value_list)

great_expectations.self_check.util.evaluate_json_test(data_asset, expectation_type, test)

This method will evaluate the result of a test build using the Great Expectations json test format.

NOTE: Tests can be suppressed for certain data types if the test contains the Key ‘suppress_test_for’ with a list

of DataAsset types to suppress, such as [‘SQLAlchemy’, ‘Pandas’].

Parameters

• data_asset – (DataAsset) A great expectations DataAsset

• expectation_type – (string) the name of the expectation to be run using the test input

• test –

  (dict) a dictionary containing information for the test to be run. The dictionary must include: - title: (string) the name of the test - exact_match_out: (boolean) If true, match the ‘out’ dictionary exactly against the result of the expectation - in: (dict or list) a dictionary of keyword arguments to use to evaluate the expectation or a list of positional arguments - out: (dict) the dictionary keys against which to make assertions. Unless exact_match_out is true, keys must come from the following list:

  • success

  • observed_value

  • unexpected_index_list

  • unexpected_list

  • details

  • traceback_substring (if present, the string value will be expected as a substring of the exception_traceback)

Returns

None. asserts correctness of results.

great_expectations.self_check.util.evaluate_json_test_cfe(validator, expectation_type, test)

This method will evaluate the result of a test build using the Great Expectations json test format.

NOTE: Tests can be suppressed for certain data types if the test contains the Key ‘suppress_test_for’ with a list

of DataAsset types to suppress, such as [‘SQLAlchemy’, ‘Pandas’].

Parameters

• data_asset – (DataAsset) A great expectations DataAsset

• expectation_type – (string) the name of the expectation to be run using the test input

• test –

  (dict) a dictionary containing information for the test to be run. The dictionary must include: - title: (string) the name of the test - exact_match_out: (boolean) If true, match the ‘out’ dictionary exactly against the result of the expectation - in: (dict or list) a dictionary of keyword arguments to use to evaluate the expectation or a list of positional arguments - out: (dict) the dictionary keys against which to make assertions. Unless exact_match_out is true, keys must come from the following list:

  • success

  • observed_value

  • unexpected_index_list

  • unexpected_list

  • details

  • traceback_substring (if present, the string value will be expected as a substring of the exception_traceback)

Returns

None. asserts correctness of results.

great_expectations.self_check.util.check_json_test_result(test, result, data_asset=None)

great_expectations.self_check.util.generate_test_table_name(default_table_name_prefix: str = 'test_data_') → str

great_expectations.self_check.util._create_bigquery_engine() → Engine

great_expectations.self_check.util._bigquery_dataset() → str