Kkit.timeout

This module is used to run a command by subprocess.run with timeout and retry times.

Example:

from Kkit import timeout
result = timeout.run_command_with_timeout(["sleep", "10"], timeout=1, retry_times=3)
#or
result = timeout.run_shell_with_timeout("sleep 10", timeout=1, retry_times=3)

View Source

 1"""
 2This module is used to run a command by `subprocess.run` with timeout and retry times.
 3
 4Example:
 5
 6```python
 7from Kkit import timeout
 8result = timeout.run_command_with_timeout(["sleep", "10"], timeout=1, retry_times=3)
 9#or
10result = timeout.run_shell_with_timeout("sleep 10", timeout=1, retry_times=3)
11```
12"""
13
14import subprocess
15
16
17def run_command_with_timeout(command: list, timeout=1, retry_times=3, **kwargs):
18    """
19    Run a command with timeout and retry times
20
21    Parameters
22    ----------
23    command : list
24        The command to run
25
26    timeout : int
27        The timeout for the command
28
29    retry_times : int
30        The retry times for the command
31
32    **kwargs
33        Other parameters for subprocess.run
34
35    Returns
36    -------
37    subprocess.CompletedProcess or None
38        The result of the command
39    """
40    for i in range(retry_times):
41        try:
42            result = subprocess.run(command, timeout=timeout, **kwargs)
43            return result
44        except subprocess.TimeoutExpired:
45            print(f"command <{' '.join(command)}> timeouts {i+1} times")
46            continue
47    return None
48
49def run_shell_with_timeout(shell: str, timeout=1, retry_times=3, **kwargs):
50    """
51    Run a shell command with timeout and retry times
52
53    Parameters
54    ----------
55    shell : str
56        The shell command to run
57
58    timeout : int
59        The timeout for the command
60
61    retry_times : int
62        The retry times for the command
63        
64    **kwargs
65        Other parameters for subprocess.run
66
67    Returns
68    -------
69    subprocess.CompletedProcess or None
70        The result of the command
71    """
72    for i in range(retry_times):
73        try:
74            result = subprocess.run(shell, shell=True, timeout=timeout, **kwargs)
75            return result
76        except subprocess.TimeoutExpired:
77            print(f"command <{shell}> timeouts {i+1} times")
78            continue
79    return None

def run_command_with_timeout(command: list, timeout=1, retry_times=3, **kwargs): View Source

18def run_command_with_timeout(command: list, timeout=1, retry_times=3, **kwargs):
19    """
20    Run a command with timeout and retry times
21
22    Parameters
23    ----------
24    command : list
25        The command to run
26
27    timeout : int
28        The timeout for the command
29
30    retry_times : int
31        The retry times for the command
32
33    **kwargs
34        Other parameters for subprocess.run
35
36    Returns
37    -------
38    subprocess.CompletedProcess or None
39        The result of the command
40    """
41    for i in range(retry_times):
42        try:
43            result = subprocess.run(command, timeout=timeout, **kwargs)
44            return result
45        except subprocess.TimeoutExpired:
46            print(f"command <{' '.join(command)}> timeouts {i+1} times")
47            continue
48    return None

Run a command with timeout and retry times

Parameters

command : list The command to run

timeout : int The timeout for the command

retry_times : int The retry times for the command

**kwargs Other parameters for subprocess.run

Returns

subprocess.CompletedProcess or None The result of the command

def run_shell_with_timeout(shell: str, timeout=1, retry_times=3, **kwargs): View Source

50def run_shell_with_timeout(shell: str, timeout=1, retry_times=3, **kwargs):
51    """
52    Run a shell command with timeout and retry times
53
54    Parameters
55    ----------
56    shell : str
57        The shell command to run
58
59    timeout : int
60        The timeout for the command
61
62    retry_times : int
63        The retry times for the command
64        
65    **kwargs
66        Other parameters for subprocess.run
67
68    Returns
69    -------
70    subprocess.CompletedProcess or None
71        The result of the command
72    """
73    for i in range(retry_times):
74        try:
75            result = subprocess.run(shell, shell=True, timeout=timeout, **kwargs)
76            return result
77        except subprocess.TimeoutExpired:
78            print(f"command <{shell}> timeouts {i+1} times")
79            continue
80    return None