1
+ from __future__ import annotations
2
+
1
3
import subprocess
2
- from typing import Any , List , Optional , Tuple
4
+ from typing import Any
3
5
6
+ from rrmngmnt import Host
4
7
from simple_logger .logger import get_logger
5
8
6
9
from pyhelper_utils .exceptions import CommandExecFailed
7
- from rrmngmnt import Host
8
10
9
11
LOGGER = get_logger (name = __name__ )
10
12
11
13
TIMEOUT_30MIN = 30 * 60
12
14
13
15
14
16
def run_command (
15
- command : List [str ],
17
+ command : list [str ],
16
18
verify_stderr : bool = True ,
17
19
shell : bool = False ,
18
- timeout : Optional [ int ] = None ,
20
+ timeout : int | None = None ,
19
21
capture_output : bool = True ,
20
22
check : bool = True ,
21
23
hide_log_command : bool = False ,
22
24
log_errors : bool = True ,
23
25
** kwargs : Any ,
24
- ) -> Tuple [bool , str , str ]:
26
+ ) -> tuple [bool , str , str ]:
25
27
"""
26
28
Run command locally.
27
29
@@ -83,11 +85,11 @@ def run_command(
83
85
84
86
def run_ssh_commands (
85
87
host : Host ,
86
- commands : List [ str ],
88
+ commands : list [ Any ],
87
89
get_pty : bool = False ,
88
90
check_rc : bool = True ,
89
91
timeout : int = TIMEOUT_30MIN ,
90
- tcp_timeout : Optional [ float ] = None ,
92
+ tcp_timeout : float | None = None ,
91
93
) -> list :
92
94
"""
93
95
Run commands on remote host via SSH
@@ -110,8 +112,9 @@ def run_ssh_commands(
110
112
Raise:
111
113
CommandExecFailed: If command failed to execute.
112
114
"""
113
- results : List [str ] = []
114
- commands_list : List [List [str ]] = commands if isinstance (commands [0 ], list ) else [commands ]
115
+ results : list [str ] = []
116
+ commands_list : list [list [str ]] = commands if isinstance (commands [0 ], list ) else [commands ]
117
+
115
118
with host .executor ().session (timeout = tcp_timeout ) as ssh_session :
116
119
for cmd in commands_list :
117
120
rc , out , err = ssh_session .run_cmd (cmd = cmd , get_pty = get_pty , timeout = timeout )
0 commit comments