Safely run a command with retries.
(command: List[str], retry: bool = True)
| 138 | |
| 139 | |
| 140 | def safe_run_command(command: List[str], retry: bool = True) -> Tuple[bool, str, str]: |
| 141 | """Safely run a command with retries.""" |
| 142 | cmd_str = " ".join(command) |
| 143 | logger.info(f"Executing command: {cmd_str}") |
| 144 | |
| 145 | for attempt in range(MAX_RETRIES if retry else 1): |
| 146 | try: |
| 147 | logger.debug(f"Attempt {attempt + 1}/{MAX_RETRIES if retry else 1}") |
| 148 | result = run(command, capture_output=True, text=True) |
| 149 | if result.returncode == 0: |
| 150 | logger.debug("Command executed successfully") |
| 151 | return True, result.stdout, "" |
| 152 | if attempt == MAX_RETRIES - 1 or not retry: |
| 153 | logger.error( |
| 154 | f"Command failed with return code {result.returncode}: {result.stderr}" |
| 155 | ) |
| 156 | return False, "", result.stderr |
| 157 | except Exception as e: |
| 158 | if attempt == MAX_RETRIES - 1 or not retry: |
| 159 | logger.error(f"Command execution failed: {str(e)}") |
| 160 | return False, "", str(e) |
| 161 | logger.debug(f"Retrying in {RETRY_DELAY} seconds...") |
| 162 | time.sleep(RETRY_DELAY) |
| 163 | logger.error("Max retries exceeded") |
| 164 | return False, "", "Max retries exceeded" |
| 165 | |
| 166 | |
| 167 | def fetch_models_once() -> Dict[str, List[str]]: |
no test coverage detected