Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test(args):
"""Run the test suite."""
try:
import pytest # pylint: disable=unused-import
except ImportError:
raise KedroCliError(NO_DEPENDENCY_MESSAGE.format("pytest"))
else:
python_call("pytest", args)
if isinstance(reqs_path, str):
reqs_path = Path(reqs_path)
reqs_path = reqs_path.absolute()
if not reqs_path.is_file():
raise KedroCliError(
"Given path `{0}` is not a regular file.".format(str(reqs_path))
)
pattern = re.compile(package_name + r"([^\w]|$)")
with open(str(reqs_path), "r") as reqs_file:
for req_line in reqs_file:
req_line = req_line.strip()
if pattern.search(req_line):
return req_line
msg = "Cannot find `{0}` package in `{1}`.".format(package_name, str(reqs_path))
raise KedroCliError(msg)
def _assert_include_example_ok(include_example):
if not isinstance(include_example, bool):
message = (
"`{}` value for `include_example` is invalid. It must be a boolean value "
"True or False.".format(include_example)
)
raise KedroCliError(message)
KedroCliError: If the resulting ``Pipeline`` is empty.
"""
# Report project name
logging.info("** Kedro project {}".format(Path.cwd().name))
# Load Catalog
conf = get_config(project_path=str(Path.cwd()), env=env)
catalog = create_catalog(config=conf)
# Load the pipeline
pipeline = create_pipeline()
pipeline = pipeline.only_nodes_with_tags(*tags) if tags else pipeline
if not pipeline.nodes:
if tags:
raise KedroCliError("Pipeline contains no nodes with tags: " + str(tags))
raise KedroCliError("Pipeline contains no nodes")
# Load the runner
# When either --parallel or --runner is used, class_obj is assigned to runner
runner_func = load_obj(runner, "kedro.runner") if runner else SequentialRunner
# Run the runner
runner_func().run(pipeline, catalog)
"""
# Report project name
logging.info("** Kedro project {}".format(Path.cwd().name))
# Load Catalog
conf = get_config(project_path=str(Path.cwd()), env=env)
catalog = create_catalog(config=conf)
# Load the pipeline
pipeline = create_pipeline()
pipeline = pipeline.only_nodes_with_tags(*tags) if tags else pipeline
if not pipeline.nodes:
if tags:
raise KedroCliError("Pipeline contains no nodes with tags: " + str(tags))
raise KedroCliError("Pipeline contains no nodes")
# Load the runner
# When either --parallel or --runner is used, class_obj is assigned to runner
runner_func = load_obj(runner, "kedro.runner") if runner else SequentialRunner
# Run the runner
runner_func().run(pipeline, catalog)
kedro_project_path = context.project_path
kedro_package_name = "kedro_tutorial"
if all_flag:
# pathlib glob does not ignore hidden directories,
# whereas Python glob does, which is more useful in
# ensuring checkpoints will not be included
pattern = kedro_project_path / "**" / "*.ipynb"
notebooks = sorted(Path(p) for p in iglob(str(pattern), recursive=True))
else:
notebooks = [Path(f) for f in filepath]
counter = Counter(n.stem for n in notebooks)
non_unique_names = [name for name, counts in counter.items() if counts > 1]
if non_unique_names:
raise KedroCliError(
"Found non-unique notebook names! "
"Please rename the following: {}".format(", ".join(non_unique_names))
)
for notebook in notebooks:
secho("Converting notebook '{}'...".format(str(notebook)))
output_path = (
kedro_project_path
/ "src"
/ kedro_package_name
/ "nodes"
/ "{}.py".format(notebook.stem)
)
if output_path.is_file():
overwrite = overwrite_flag or click.confirm(
def activate_nbstripout():
"""Install the nbstripout git hook to automatically clean notebooks."""
secho(
(
"Notebook output cells will be automatically cleared before committing"
" to git."
),
fg="yellow",
)
try:
import nbstripout # pylint: disable=unused-import
except ImportError:
raise KedroCliError(NO_DEPENDENCY_MESSAGE.format("nbstripout"))
try:
res = subprocess.run(
["git", "rev-parse", "--git-dir"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
if res.returncode:
raise KedroCliError("Not a git repository. Run `git init` first.")
except FileNotFoundError:
raise KedroCliError("Git executable not found. Install Git first.")
call(["nbstripout", "--install"])
try:
import nbstripout # pylint: disable=unused-import
except ImportError:
raise KedroCliError(NO_DEPENDENCY_MESSAGE.format("nbstripout"))
try:
res = subprocess.run(
["git", "rev-parse", "--git-dir"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
if res.returncode:
raise KedroCliError("Not a git repository. Run `git init` first.")
except FileNotFoundError:
raise KedroCliError("Git executable not found. Install Git first.")
call(["nbstripout", "--install"])
fg="yellow",
)
try:
import nbstripout # pylint: disable=unused-import,unused-variable
except ImportError:
raise KedroCliError(NO_NBSTRIPOUT_MESSAGE)
try:
res = subprocess.run(
["git", "rev-parse", "--git-dir"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
if res.returncode:
raise KedroCliError("Not a git repository. Run `git init` first.")
except FileNotFoundError:
raise KedroCliError("Git executable not found. Install Git first.")
call(["nbstripout", "--install"])
def _handle_exception(msg, end=True):
"""Pretty print the current exception then exit."""
if _VERBOSE:
click.secho(traceback.format_exc(), nl=False, fg="yellow")
else:
etype, value, _ = sys.exc_info()
click.secho(
"".join(*traceback.format_exception_only(etype, value))
+ "Run with --verbose to see the full exception",
fg="yellow",
)
if end:
raise KedroCliError(msg)
click.secho("Error: " + msg, fg="red") # pragma: no cover