Tags
🔥 #kedrotips hooks can be created using modules
# kedro-hooks/src/kedro-hooks/preflight.py from kedro.hooks import hook_impl from kedro.io.core import DataSetNotFoundError from colorama import Fore import textwrap @hook_impl def before_pipeline_run(run_params, pipeline, catalog): missing_input = [i for i in pipeline.inputs() if not getattr(catalog.datasets, i)._exists()] if len(missing_input) != 0: raise DataSetNotFoundError(textwrap.dedent(f''' {Fore.LIGHTBLACK_EX}―――――――― {Fore.RED}PREFLIGHT ERROR {Fore.LIGHTBLACK_EX}――――――――― {Fore.RESET} preflight of pipeline failed due to {Fore.YELLOW}missing datasets {Fore.BLUE} {missing_input}{Fore.RESET} '''))