File size: 2,190 Bytes
e8059bf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
import great_expectations as ge
from great_expectations.core.batch import BatchRequest
from great_expectations.checkpoint import SimpleCheckpoint
from great_expectations.exceptions import DataContextError
"""
**IMPORTANT**
This file is meant to be run as an end-2-end test command.
It is an example and might require changes to suit your data and use case.
Please refer to the `example-project` branch of the Cookiecutter-MLOps project for a complete working example –
https://dagshub.com/DagsHub/Cookiecutter-MLOps/src/example-project
"""
context = ge.data_context.DataContext()
batch_request = {
"datasource_name": "{{fill in datasource name}}",
"data_connector_name": "default_inferred_data_connector_name",
"data_asset_name": "{{fill in names of files that have tests}}", # Define the data files that should be tested
"limit": 1000,
}
# Feel free to change the name of your suite here. Renaming this will not remove the other one.
expectation_suite_name = "{{fill in your test suite name}}"
try:
suite = context.get_expectation_suite(expectation_suite_name=expectation_suite_name)
print(
f'Loaded ExpectationSuite "{suite.expectation_suite_name}" containing {len(suite.expectations)} expectations'
)
except DataContextError:
suite = context.create_expectation_suite(
expectation_suite_name=expectation_suite_name
)
print(f'Created ExpectationSuite "{suite.expectation_suite_name}".')
validator = context.get_validator(
batch_request=BatchRequest(**batch_request),
expectation_suite_name=expectation_suite_name,
)
checkpoint_config = {
"class_name": "SimpleCheckpoint",
"validations": [
{
"batch_request": batch_request,
"expectation_suite_name": expectation_suite_name,
}
],
}
checkpoint = SimpleCheckpoint(
f"{validator.active_batch_definition.data_asset_name}_{expectation_suite_name}",
context,
**checkpoint_config,
)
checkpoint_result = checkpoint.run()
context.build_data_docs()
validation_result_identifier = checkpoint_result.list_validation_result_identifiers()[0]
context.open_data_docs(resource_identifier=validation_result_identifier)
|