sqlalchemy-logging.py

import great_expectations as gx
from great_expectations.data_context.types.base import DataContextConfig, FilesystemStoreBackendDefaults
import os
import logging

logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)


context_dir = os.path.abspath("./mysql-context")
data_context_config = DataContextConfig(
    store_backend_defaults=FilesystemStoreBackendDefaults(
        root_directory=context_dir),
)
context = gx.get_context(project_config=data_context_config)
with open("./mysql-context/great_expectations.yml", 'w') as outfile:
    context.config.to_yaml(outfile)

datasource_config = {
    "name": "my_mysql_datasource",
    "class_name": "Datasource",
    "execution_engine": {
        "class_name": "SqlAlchemyExecutionEngine",
        "connection_string": "mysql+pymysql://root:root@localhost:3307/test_db",
    },
    "data_connectors": {
        "default_runtime_data_connector_name": {
            "class_name": "RuntimeDataConnector",
            "batch_identifiers": ["default_identifier_name"],
        },
        "default_inferred_data_connector_name": {
            "class_name": "InferredAssetSqlDataConnector",
            "include_schema_name": True,
        },
    },
}

context.add_datasource(**datasource_config)

from great_expectations.core.batch import BatchRequest, RuntimeBatchRequest

batch_request = RuntimeBatchRequest(
    datasource_name="my_mysql_datasource",
    data_connector_name="default_runtime_data_connector_name",
    data_asset_name="default_name",
    runtime_parameters={"query": "SELECT * from test_db.students"},
    batch_identifiers={"default_identifier_name": "default_identifier"},
)
context.add_or_update_expectation_suite(expectation_suite_name="test_suite")
validator = context.get_validator(
    batch_request=batch_request, expectation_suite_name="test_suite"
)

validator.expect_column_values_to_be_unique("email")