|
1 | 1 | import json |
2 | | -import subprocess |
3 | 2 |
|
4 | 3 | from data import TOOLS |
| 4 | +from lib.commands import local_cmd |
5 | 5 | from lib.typing import JSONType |
6 | 6 |
|
7 | | -from typing import Dict, Literal, overload |
| 7 | +from typing import Literal, overload |
8 | 8 |
|
9 | | -# TODO: either: |
10 | | -# * replace simple_output and use_json by a single output type |
11 | | -# * make sure that simple_output=False and use_json=True are not being used together |
12 | | - |
13 | | -@overload |
14 | | -def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: Literal[True] = True, |
15 | | - use_json: Literal[False] = False) -> str: |
16 | | - ... |
17 | | -@overload |
18 | | -def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: Literal[True] = True, |
19 | | - use_json: Literal[True]) -> JSONType: |
20 | | - ... |
21 | 9 | @overload |
22 | | -def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: Literal[False], |
23 | | - use_json: bool = False) -> subprocess.CompletedProcess[bytes]: |
| 10 | +def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_json: Literal[False] = False) -> str: |
24 | 11 | ... |
25 | 12 | @overload |
26 | | -def xo_cli(action: str, args: Dict[str, str] = {}, *, check: bool = True, simple_output: bool = True, |
27 | | - use_json: bool = False) -> subprocess.CompletedProcess[bytes] | JSONType | str: |
| 13 | +def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_json: Literal[True]) -> JSONType: |
28 | 14 | ... |
29 | | -def xo_cli( |
30 | | - action: str, args: dict[str, str] = {}, check: bool = True, simple_output: bool = True, use_json: bool = False |
31 | | -) -> subprocess.CompletedProcess[bytes] | JSONType | str: |
32 | | - run_array = [TOOLS.get('xo-cli', 'xo-cli'), action] |
| 15 | + |
| 16 | +def xo_cli(action: str, args: dict[str, str] = {}, *, check: bool = True, use_json: bool = False) -> JSONType | str: |
| 17 | + cmd = [TOOLS.get('xo-cli', 'xo-cli'), action] |
33 | 18 | if use_json: |
34 | | - run_array += ['--json'] |
35 | | - run_array += ["%s=%s" % (key, value) for key, value in args.items()] |
36 | | - res = subprocess.run( |
37 | | - run_array, |
38 | | - stdout=subprocess.PIPE, |
39 | | - stderr=subprocess.STDOUT, |
40 | | - check=check |
41 | | - ) |
42 | | - if simple_output: |
43 | | - output = res.stdout.decode().strip() |
44 | | - if use_json: |
45 | | - return json.loads(output) |
46 | | - return output |
47 | | - return res |
| 19 | + cmd += ['--json'] |
| 20 | + cmd += ["%s=%s" % (key, value) for key, value in args.items()] |
| 21 | + |
| 22 | + res = local_cmd(cmd, check=check) |
| 23 | + |
| 24 | + if use_json: |
| 25 | + return json.loads(res.stdout) |
| 26 | + |
| 27 | + return res.stdout |
48 | 28 |
|
49 | 29 | def xo_object_exists(uuid: str) -> bool: |
50 | | - lst = json.loads(xo_cli('--list-objects', {'uuid': uuid})) |
| 30 | + lst = xo_cli('list-objects', {'uuid': uuid}, use_json=True) |
| 31 | + assert isinstance(lst, list) |
51 | 32 | return len(lst) > 0 |
0 commit comments