-
Notifications
You must be signed in to change notification settings - Fork 17
Expand file tree
/
Copy path__main__.py
More file actions
88 lines (68 loc) · 3.08 KB
/
__main__.py
File metadata and controls
88 lines (68 loc) · 3.08 KB
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
"""Command-line interface."""
from __future__ import annotations
import importlib
from importlib import metadata
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Callable
from typing import Any
import click
KNOWN_MODULES = ["segy.py", "copy.py", "info.py"]
class MyCLI(click.Group):
"""CLI generator via plugin design pattern.
This class dynamically loads command modules from the specified `plugin_folder`. If the
command us another CLI group, the command module must define a `cli = click.Group(...)` and
subsequent commands must be added to this CLI. If it is a single utility it must have a
variable named `cli` for the command to be exposed.
Args:
plugin_folder: Path to the directory containing command modules
*args: Variable length argument list passed to the click.Group.
**kwargs: Arbitrary keyword arguments passed to the click.Group.
"""
def __init__(self, plugin_folder: Path, *args: Any, **kwargs: Any): # noqa: ANN401
super().__init__(*args, **kwargs)
self.plugin_folder = plugin_folder
self.known_modules = KNOWN_MODULES
def list_commands(self, _ctx: click.Context) -> list[str]:
"""List commands available under `commands` module."""
rv = []
for filename in self.plugin_folder.iterdir():
is_known = filename.name in self.known_modules
is_python = filename.suffix == ".py"
if is_known and is_python:
rv.append(filename.stem)
rv.sort()
return rv
def get_command(self, _ctx: click.Context, name: str) -> Callable | None:
"""Get command implementation from `commands` module."""
try:
filepath = self.plugin_folder / f"{name}.py"
if filepath.name not in self.known_modules:
click.echo(f"Command {name} is not safe to execute.")
return None
module_name = f"mdio.commands.{name}"
spec = importlib.util.spec_from_file_location(module_name, str(filepath))
if spec and spec.loader:
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module.cli
except Exception as e:
click.echo(f"Error loading command {name}: {e}")
return None
def get_package_version(package_name: str, default: str = "unknown") -> str:
"""Safely fetch the package version, providing a default if not found."""
try:
return metadata.version(package_name)
except metadata.PackageNotFoundError:
return default
@click.command(cls=MyCLI, plugin_folder=Path(__file__).parent / "commands")
@click.version_option(get_package_version("multidimio"))
def main() -> None:
"""Welcome to MDIO!
MDIO is an open source, cloud-native, and scalable storage engine
for various types of energy data.
MDIO supports importing or exporting various data containers,
hence we allow plugins as subcommands.
From this main command, we can see the MDIO version.
"""