-
Notifications
You must be signed in to change notification settings - Fork 61
/
__main__.py
47 lines (37 loc) · 1.5 KB
/
__main__.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
"""{{ cookiecutter.project_name }} file for ensuring the package is executable
as `{{ cookiecutter.repo_name }}` and `python -m {{ cookiecutter.python_package }}`
"""
import importlib
from pathlib import Path
from kedro.framework.cli.utils import KedroCliError, load_entry_points
from kedro.framework.project import configure_project
def _find_run_command(package_name):
try:
project_cli = importlib.import_module(f"{package_name}.cli")
# fail gracefully if cli.py does not exist
except ModuleNotFoundError as exc:
if f"{package_name}.cli" not in str(exc):
raise
plugins = load_entry_points("project")
run = _find_run_command_in_plugins(plugins) if plugins else None
if run:
# use run command from installed plugin if it exists
return run
# use run command from `kedro.framework.cli.project`
from kedro.framework.cli.project import run
return run
# fail badly if cli.py exists, but has no `cli` in it
if not hasattr(project_cli, "cli"):
raise KedroCliError(f"Cannot load commands from {package_name}.cli")
return project_cli.run
def _find_run_command_in_plugins(plugins):
for group in plugins:
if "run" in group.commands:
return group.commands["run"]
def main(*args, **kwargs):
package_name = Path(__file__).parent.name
configure_project(package_name)
run = _find_run_command(package_name)
run(*args, **kwargs)
if __name__ == "__main__":
main()