forked from hail-is/hail
-
Notifications
You must be signed in to change notification settings - Fork 0
/
cli.py
155 lines (117 loc) · 5.01 KB
/
cli.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
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
import os
import sys
from typing import Optional, Tuple, Annotated as Ann
from rich import print
import typer
from typer import Argument as Arg
from hailtop.config.variables import ConfigVariable
from .config_variables import config_variables
app = typer.Typer(
name='config',
no_args_is_help=True,
help='Manage Hail configuration.',
pretty_exceptions_show_locals=False,
)
def get_section_key_path(parameter: str) -> Tuple[str, str, Tuple[str, ...]]:
path = parameter.split('/')
if len(path) == 1:
return 'global', path[0], tuple(path)
if len(path) == 2:
return path[0], path[1], tuple(path)
print(
'''
Parameters must contain at most one slash separating the configuration section
from the configuration parameter, for example: "batch/billing_project".
Parameters may also have no slashes, indicating the parameter is a global
parameter, for example: "domain".
A parameter with more than one slash is invalid, for example:
"batch/billing/project".
'''.lstrip(
'\n'
),
file=sys.stderr,
)
sys.exit(1)
def complete_config_variable(incomplete: str):
for var, var_info in config_variables().items():
if var.value.startswith(incomplete):
yield (var.value, var_info.help_msg)
@app.command()
def set(parameter: Ann[ConfigVariable, Arg(help="Configuration variable to set", autocompletion=complete_config_variable)], value: str):
'''Set a Hail configuration parameter.'''
from hailtop.config import get_user_config, get_user_config_path # pylint: disable=import-outside-toplevel
if parameter not in config_variables():
print(f"Error: unknown parameter {parameter!r}", file=sys.stderr)
sys.exit(1)
section, key, _ = get_section_key_path(parameter.value)
config_variable_info = config_variables()[parameter]
validation_func, error_msg = config_variable_info.validation
if not validation_func(value):
print(f"Error: bad value {value!r} for parameter {parameter!r} {error_msg}", file=sys.stderr)
sys.exit(1)
config = get_user_config()
config_file = get_user_config_path()
if section not in config:
config[section] = {}
config[section][key] = value
try:
f = open(config_file, 'w', encoding='utf-8')
except FileNotFoundError:
os.makedirs(config_file.parent, exist_ok=True)
f = open(config_file, 'w', encoding='utf-8')
with f:
config.write(f)
def get_config_variable(incomplete: str):
from hailtop.config import get_user_config # pylint: disable=import-outside-toplevel
config = get_user_config()
elements = []
for section_name, section in config.items():
for item_name, value in section.items():
if section_name == 'global':
path = item_name
else:
path = f'{section_name}/{item_name}'
elements.append((path, value))
config_items = {var.name: var_info.help_msg for var, var_info in config_variables().items()}
for name, _ in elements:
if name.startswith(incomplete):
help_msg = config_items.get(name)
yield (name, help_msg)
@app.command()
def unset(parameter: Ann[str, Arg(help="Configuration variable to unset", autocompletion=get_config_variable)]):
'''Unset a Hail configuration parameter (restore to default behavior).'''
from hailtop.config import get_user_config, get_user_config_path # pylint: disable=import-outside-toplevel
config = get_user_config()
config_file = get_user_config_path()
section, key, _ = get_section_key_path(parameter)
if section in config and key in config[section]:
del config[section][key]
with open(config_file, 'w', encoding='utf-8') as f:
config.write(f)
else:
print(f"WARNING: Unknown parameter {parameter!r}", file=sys.stderr)
@app.command()
def get(parameter: Ann[str, Arg(help="Configuration variable to get", autocompletion=get_config_variable)]):
'''Get the value of a Hail configuration parameter.'''
from hailtop.config import get_user_config # pylint: disable=import-outside-toplevel
config = get_user_config()
section, key, _ = get_section_key_path(parameter)
if section in config and key in config[section]:
print(config[section][key])
@app.command(name='config-location')
def config_location():
'''Print the location of the config file.'''
from hailtop.config import get_user_config_path # pylint: disable=import-outside-toplevel
print(get_user_config_path())
@app.command()
def list(section: Ann[Optional[str], Arg(show_default='all sections')] = None):
'''Lists every config variable in the section.'''
from hailtop.config import get_user_config # pylint: disable=import-outside-toplevel
config = get_user_config()
if section:
for key, value in config.items(section):
print(f'{key}={value}')
else:
for sname, items in config.items():
for key, value in items.items():
print(f'{sname}/{key}={value}')