kedro.framework.cli.utils

module documentation

(source)

Utilities for use with click.

Class	`CommandCollection`	Modified from the Click one to still run the source groups function.
Class	`KedroCliError`	Exceptions generated from the Kedro CLI.
Function	`call`	Run a subprocess command and raise if it fails.
Function	`env_option`	Add `--env` CLI option to a function.
Function	`find_stylesheets`	Fetch all stylesheets used in the official Kedro documentation
Function	`forward_command`	A command that receives the rest of the command line as 'args'.
Function	`get_pkg_version`	Get package version from requirements.txt.
Function	`python_call`	Run a subprocess command that invokes a Python module.
Function	`split_node_names`	Split string by comma, ignoring commas enclosed by square parentheses. This avoids splitting the string of nodes names on commas included in default node names, which have the pattern <function_name>([<input_name>,...
Function	`split_string`	Split string by comma.
Constant	`CONTEXT_SETTINGS`	Undocumented
Constant	`CUTOFF`	Undocumented
Constant	`ENTRY_POINT_GROUPS`	Undocumented
Constant	`ENV_HELP`	Undocumented
Constant	`MAX_SUGGESTIONS`	Undocumented
Variable	`logger`	Undocumented
Function	`_check_module_importable`	Undocumented
Function	`_clean_pycache`	Recursively clean all __pycache__ folders from `path`.
Function	`_click_verbose`	Click option for enabling verbose mode.
Function	`_config_file_callback`	CLI callback that replaces command line options with values specified in a config file. If command line options are passed, they override config file values.
Function	`_deprecate_options`	Undocumented
Function	`_filter_deprecation_warnings`	Temporarily suppress all DeprecationWarnings.
Function	`_get_entry_points`	Get all kedro related entry points
Function	`_get_values_as_tuple`	Undocumented
Function	`_reformat_load_versions`	Reformat data structure from tuple to dictionary for `load-version`, e.g.: ('dataset1:time1', 'dataset2:time2') -> {"dataset1": "time1", "dataset2": "time2"}.
Function	`_safe_load_entry_point`	Load entrypoint safely, if fails it will just skip the entrypoint.
Function	`_split_load_versions`	Undocumented
Function	`_split_params`	Undocumented
Function	`_suggest_cli_command`	Undocumented
Function	`_update_verbose_flag`	Undocumented

def call(cmd: List[str], **kwargs): (source) ¶

Run a subprocess command and raise if it fails.

Parameters
cmd:`List[str]`	List of command parts.
**kwargs	Optional keyword arguments passed to `subprocess.run`.
Raises
`click.exceptions.Exit`	If `subprocess.run` returns non-zero code.

def env_option(func_=None, **kwargs): (source) ¶

Add --env CLI option to a function.

def find_stylesheets() -> Iterable[str]: (source) ¶

Fetch all stylesheets used in the official Kedro documentation

def forward_command(group, name=None, forward_help=False): (source) ¶

A command that receives the rest of the command line as 'args'.

def get_pkg_version(reqs_path: Union[str, Path], package_name: str) -> str: (source) ¶

Get package version from requirements.txt.

Parameters
reqs_path:`Union[str, Path]`	Path to requirements.txt file.
package_name:`str`	Package to search for.
Returns
`str`	Package and its version as specified in requirements.txt.
Raises
`KedroCliError`	If the file specified in `reqs_path` does not exist or `package_name` was not found in that file.

def python_call(module: str, arguments: Iterable[str], **kwargs): (source) ¶

Run a subprocess command that invokes a Python module.

def split_node_names(ctx, param, to_split: str) -> List[str]: (source) ¶

Split string by comma, ignoring commas enclosed by square parentheses. This avoids splitting the string of nodes names on commas included in default node names, which have the pattern <function_name>([<input_name>,...]) -> [<output_name>,...])

Note

to_split will have such commas if and only if it includes a

default node name. User-defined node names cannot include commas or square brackets. - This function will no longer be necessary from Kedro 0.19.*, in which default node names will no longer contain commas

Parameters
ctx	Undocumented
param	Undocumented
to_split:`str`	the string to split safely
Returns
`List[str]`	A list containing the result of safe-splitting the string.

def split_string(ctx, param, value): (source) ¶

Split string by comma.

CONTEXT_SETTINGS: dict = (source) ¶

Undocumented

Value

{'help_option_names': ['-h', '--help']}

CUTOFF: float = (source) ¶

Undocumented

Value

0.5

ENTRY_POINT_GROUPS: dict[str, str] = (source) ¶

Undocumented

Value

{'global': 'kedro.global_commands',
 'project': 'kedro.project_commands',
 'init': 'kedro.init',
 'line_magic': 'kedro.line_magic',
 'hooks': 'kedro.hooks',
 'cli_hooks': 'kedro.cli_hooks',
 'starters': 'kedro.starters'}

ENV_HELP: str = (source) ¶

Undocumented

Value

'Kedro configuration environment name. Defaults to `local`.'

MAX_SUGGESTIONS: int = (source) ¶

Undocumented

Value

logger = (source) ¶

Undocumented

def _check_module_importable(module_name: str): (source) ¶

Undocumented

def _clean_pycache(path: Path): (source) ¶

Recursively clean all __pycache__ folders from path.

Parameters
path:`Path`	Existing local directory to clean __pycache__ folders from.

def _click_verbose(func): (source) ¶

Click option for enabling verbose mode.

def _config_file_callback(ctx, param, value): (source) ¶

CLI callback that replaces command line options with values specified in a config file. If command line options are passed, they override config file values.

def _deprecate_options(ctx, param, value): (source) ¶

Undocumented

@contextmanager
def _filter_deprecation_warnings(): (source) ¶

Temporarily suppress all DeprecationWarnings.

def _get_entry_points(name: str) -> importlib_metadata.EntryPoints: (source) ¶

Get all kedro related entry points

def _get_values_as_tuple(values: Iterable[str]) -> Tuple[str, ...]: (source) ¶

Undocumented

def _reformat_load_versions(ctx, param, value) -> Dict[str, str]: (source) ¶

Reformat data structure from tuple to dictionary for load-version, e.g.: ('dataset1:time1', 'dataset2:time2') -> {"dataset1": "time1", "dataset2": "time2"}.

def _safe_load_entry_point(entry_point): (source) ¶

Load entrypoint safely, if fails it will just skip the entrypoint.

def _split_load_versions(ctx, param, value): (source) ¶

Undocumented

def _split_params(ctx, param, value): (source) ¶

Undocumented

def _suggest_cli_command(original_command_name: str, existing_command_names: Iterable[str]) -> str: (source) ¶

Undocumented

def _update_verbose_flag(ctx, param, value): (source) ¶

Undocumented