great_expectations.cli.toolkit

Module Contents

Classes

MyYAML(: Any, _kw: Optional[Text] = enforce, typ: Any = None, pure: Any = False, output: Any = None, plug_ins=None)

Functions

prompt_profile_to_create_a_suite(data_context: DataContext, expectation_suite_name: str)

get_or_create_expectation_suite(expectation_suite_name: str, data_context: DataContext, data_asset_name: Optional[str] = None, usage_event: Optional[str] = None, suppress_usage_message: Optional[bool] = False, batch_request: Optional[Union[str, Dict[str, Union[str, int, Dict[str, Any]]]]] = None, create_if_not_exist: Optional[bool] = True)

get_default_expectation_suite_name(data_asset_name: str, batch_request: Optional[Union[str, Dict[str, Union[str, int, Dict[str, Any]]]]] = None)

tell_user_suite_exists(data_context: DataContext, expectation_suite_name: str, usage_event: str, suppress_usage_message: Optional[bool] = False)

launch_jupyter_notebook(notebook_path: str)

get_validator(context: DataContext, batch_request: Union[dict, BatchRequest], suite: Union[str, ExpectationSuite])

load_expectation_suite(data_context: DataContext, expectation_suite_name: str, usage_event: str, suppress_usage_message: Optional[bool] = False, create_if_not_exist: Optional[bool] = True)

Load an expectation suite from a given context.

exit_with_failure_message_and_stats(data_context: DataContext, usage_event: str, suppress_usage_message: Optional[bool] = False, message: Optional[str] = None)

delete_checkpoint(context: DataContext, checkpoint_name: str, usage_event: str, assume_yes: bool)

Delete a Checkpoint or raise helpful errors.

run_checkpoint(context: DataContext, checkpoint_name: str, usage_event: str)

Run a Checkpoint or raise helpful errors.

validate_checkpoint(context: DataContext, checkpoint_name: str, usage_event: str, failure_message: Optional[str] = None)

load_checkpoint(context: DataContext, checkpoint_name: str, usage_event: str)

Load a Checkpoint or raise helpful errors.

select_datasource(context: DataContext, datasource_name: str = None)

Select a datasource interactively.

load_data_context_with_error_handling(directory: str, from_cli_upgrade_command: bool = False)

Return a DataContext with good error handling and exit codes.

upgrade_project(context_root_dir, ge_config_version, from_cli_upgrade_command=False)

upgrade_project_one_version_increment(context_root_dir: str, ge_config_version: float, continuation_message: str, from_cli_upgrade_command: bool = False)

confirm_proceed_or_exit(confirm_prompt: str = ‘Would you like to proceed?’, continuation_message: str = ‘Ok, exiting now. You can always read more at https://docs.greatexpectations.io/ !’, exit_on_no: bool = True, exit_code: int = 0, data_context: Optional[DataContext] = None, usage_stats_event: Optional[str] = None)

Every CLI command that starts a potentially lengthy (>1 sec) computation

parse_cli_config_file_location(config_file_location: str)

Parse CLI yaml config file or directory location into directory and filename.

send_usage_message(data_context: DataContext, event: str, event_payload: Optional[dict] = None, success: bool = False)

is_cloud_file_url(file_path: str)

Check for commonly used cloud urls.

get_relative_path_from_config_file_to_base_path(context_root_directory: str, data_path: str)

This function determines the relative path from a given data path relative

load_json_file_into_dict(filepath: str, data_context: DataContext, usage_event: Optional[str] = None)

get_batch_request_from_citations(expectation_suite: Optional[ExpectationSuite] = None)

add_citation_with_batch_request(data_context: DataContext, expectation_suite: ExpectationSuite, batch_request: Optional[Dict[str, Union[str, int, Dict[str, Any]]]] = None)

get_batch_request_from_json_file(batch_request_json_file_path: str, data_context: DataContext, usage_event: Optional[str] = None, suppress_usage_message: Optional[bool] = False)

get_batch_request_using_datasource_name(data_context: DataContext, datasource_name: Optional[str] = None, usage_event: Optional[str] = None, suppress_usage_message: Optional[bool] = False, additional_batch_request_args: Optional[Dict[str, Union[str, int, Dict[str, Any]]]] = None)

great_expectations.cli.toolkit.EXIT_UPGRADE_CONTINUATION_MESSAGE =

Ok, exiting now. To upgrade at a later time, use the following command: <cyan>great_expectations project upgrade</cyan>

To learn more about the upgrade process, visit <cyan>https://docs.greatexpectations.io/en/latest/how_to_guides/migrating_versions.html</cyan>.

class great_expectations.cli.toolkit.MyYAML(: Any, _kw: Optional[Text] = enforce, typ: Any = None, pure: Any = False, output: Any = None, plug_ins=None)

Bases: ruamel.yaml.YAML

dump(self, data, stream=None, **kw)
great_expectations.cli.toolkit.yaml
great_expectations.cli.toolkit.default_flow_style = False
great_expectations.cli.toolkit.prompt_profile_to_create_a_suite(data_context: DataContext, expectation_suite_name: str)
great_expectations.cli.toolkit.get_or_create_expectation_suite(expectation_suite_name: str, data_context: DataContext, data_asset_name: Optional[str] = None, usage_event: Optional[str] = None, suppress_usage_message: Optional[bool] = False, batch_request: Optional[Union[str, Dict[str, Union[str, int, Dict[str, Any]]]]] = None, create_if_not_exist: Optional[bool] = True) → ExpectationSuite
great_expectations.cli.toolkit.get_default_expectation_suite_name(data_asset_name: str, batch_request: Optional[Union[str, Dict[str, Union[str, int, Dict[str, Any]]]]] = None) → str
great_expectations.cli.toolkit.tell_user_suite_exists(data_context: DataContext, expectation_suite_name: str, usage_event: str, suppress_usage_message: Optional[bool] = False)
great_expectations.cli.toolkit.launch_jupyter_notebook(notebook_path: str)
great_expectations.cli.toolkit.get_validator(context: DataContext, batch_request: Union[dict, BatchRequest], suite: Union[str, ExpectationSuite]) → Validator
great_expectations.cli.toolkit.load_expectation_suite(data_context: DataContext, expectation_suite_name: str, usage_event: str, suppress_usage_message: Optional[bool] = False, create_if_not_exist: Optional[bool] = True) → Optional[ExpectationSuite]

Load an expectation suite from a given context.

Handles a suite name with or without .json :param data_context: :param expectation_suite_name: :param usage_event: :param suppress_usage_message: :param create_if_not_exist:

great_expectations.cli.toolkit.exit_with_failure_message_and_stats(data_context: DataContext, usage_event: str, suppress_usage_message: Optional[bool] = False, message: Optional[str] = None)
great_expectations.cli.toolkit.delete_checkpoint(context: DataContext, checkpoint_name: str, usage_event: str, assume_yes: bool)

Delete a Checkpoint or raise helpful errors.

great_expectations.cli.toolkit.run_checkpoint(context: DataContext, checkpoint_name: str, usage_event: str) → CheckpointResult

Run a Checkpoint or raise helpful errors.

great_expectations.cli.toolkit.validate_checkpoint(context: DataContext, checkpoint_name: str, usage_event: str, failure_message: Optional[str] = None)
great_expectations.cli.toolkit.load_checkpoint(context: DataContext, checkpoint_name: str, usage_event: str) → Union[Checkpoint, LegacyCheckpoint]

Load a Checkpoint or raise helpful errors.

great_expectations.cli.toolkit.select_datasource(context: DataContext, datasource_name: str = None) → BaseDatasource

Select a datasource interactively.

great_expectations.cli.toolkit.load_data_context_with_error_handling(directory: str, from_cli_upgrade_command: bool = False) → DataContext

Return a DataContext with good error handling and exit codes.

great_expectations.cli.toolkit.upgrade_project(context_root_dir, ge_config_version, from_cli_upgrade_command=False)
great_expectations.cli.toolkit.upgrade_project_one_version_increment(context_root_dir: str, ge_config_version: float, continuation_message: str, from_cli_upgrade_command: bool = False) → [bool, bool]
great_expectations.cli.toolkit.confirm_proceed_or_exit(confirm_prompt: str = 'Would you like to proceed?', continuation_message: str = 'Ok, exiting now. You can always read more at https://docs.greatexpectations.io/ !', exit_on_no: bool = True, exit_code: int = 0, data_context: Optional[DataContext] = None, usage_stats_event: Optional[str] = None) → Optional[bool]

Every CLI command that starts a potentially lengthy (>1 sec) computation or modifies some resources (e.g., edits the config file, adds objects to the stores) must follow this pattern: 1. Explain which resources will be created/modified/deleted 2. Use this method to ask for user’s confirmation

The goal of this standardization is for the users to expect consistency - if you saw one command, you know what to expect from all others.

If the user does not confirm, the program should exit. The purpose of the exit_on_no parameter is to provide the option to perform cleanup actions before exiting outside of the function.

great_expectations.cli.toolkit.parse_cli_config_file_location(config_file_location: str) → dict

Parse CLI yaml config file or directory location into directory and filename. Uses pathlib to handle windows paths. :param config_file_location: string of config_file_location

Returns

{

“directory”: “directory/where/config/file/is/located”, “filename”: “great_expectations.yml”

}

great_expectations.cli.toolkit.send_usage_message(data_context: DataContext, event: str, event_payload: Optional[dict] = None, success: bool = False)
great_expectations.cli.toolkit.is_cloud_file_url(file_path: str) → bool

Check for commonly used cloud urls.

great_expectations.cli.toolkit.get_relative_path_from_config_file_to_base_path(context_root_directory: str, data_path: str) → str

This function determines the relative path from a given data path relative to the great_expectations.yml file independent of the current working directory.

This allows a user to use the CLI from any directory, type a relative path from their current working directory and have the correct relative path be put in the great_expectations.yml file.

great_expectations.cli.toolkit.load_json_file_into_dict(filepath: str, data_context: DataContext, usage_event: Optional[str] = None) → Optional[Dict[str, Union[str, int, Dict[str, Any]]]]
great_expectations.cli.toolkit.get_batch_request_from_citations(expectation_suite: Optional[ExpectationSuite] = None) → Optional[Union[str, Dict[str, Union[str, int, Dict[str, Any]]]]]
great_expectations.cli.toolkit.add_citation_with_batch_request(data_context: DataContext, expectation_suite: ExpectationSuite, batch_request: Optional[Dict[str, Union[str, int, Dict[str, Any]]]] = None)
great_expectations.cli.toolkit.get_batch_request_from_json_file(batch_request_json_file_path: str, data_context: DataContext, usage_event: Optional[str] = None, suppress_usage_message: Optional[bool] = False) → Optional[Union[str, Dict[str, Union[str, int, Dict[str, Any]]]]]
great_expectations.cli.toolkit.get_batch_request_using_datasource_name(data_context: DataContext, datasource_name: Optional[str] = None, usage_event: Optional[str] = None, suppress_usage_message: Optional[bool] = False, additional_batch_request_args: Optional[Dict[str, Union[str, int, Dict[str, Any]]]] = None) → Optional[Union[str, Dict[str, Union[str, int, Dict[str, Any]]]]]