def METHOD_NAME(
    self,
    df: pd.DataFrame = None,
    expectations_path: str = None,
    keep_output: bool = None,
    **kwargs,
):
    ge_project_path = str(Path(expectations_path).parent)
    batch_kwargs = self._get_batch_kwargs(df)
    context = self._get_ge_context_local(ge_project_path)
    self.logger.info("Beginning validation run...")
    try:
        results = super().METHOD_NAME(
            batch_kwargs=batch_kwargs,  # input data
            context=context,  # ~project config
            **kwargs,
        )
    except signals.FAIL as e:
        results = e.state.result
    # Show summary of results
    n_successful, n_expectations = self._get_stats_from_results(results)
    status = "success" if results.success else "failure"
    level = logging.INFO if results.success else logging.ERROR
    self.logger.log(
        msg=f"Validation finished with status '{status}'. {n_successful}/{n_expectations} test(s) passed.",
        level=level,
    )
    validation_ids = [res for res in results["run_results"]]
    validation_id = validation_ids[0]
    url_dicts = context.get_docs_sites_urls(resource_identifier=validation_id)
    validation_site_url = url_dicts[0]["site_url"]
    if keep_output:
        docs_msg = f"To explore the docs, visit {validation_site_url}"
        docs_msg += " or the 'Artifacts' tab on the Prefect flow run dashboard."
        self.logger.info(docs_msg)
    else:
        docs_path = os.path.join(ge_project_path, "uncommitted")
        checkpoints_path = os.path.join(ge_project_path, "checkpoints")
        shutil.rmtree(docs_path)
        shutil.rmtree(checkpoints_path)
    if not results.success:
        raise signals.FAIL(result=results)
    return results