Executes a function in a child process with a specified timeout. Usage example: with_timeout(save_contact, args=(first_name, last_name), timeout_in_seconds=0.5) Args: function: The function to be executed in a child process.
(function, *, args=None, timeout_in_seconds)
| 52 | |
| 53 | |
| 54 | def with_timeout(function, *, args=None, timeout_in_seconds): |
| 55 | """Executes a function in a child process with a specified timeout. |
| 56 | |
| 57 | Usage example: |
| 58 | |
| 59 | with_timeout(save_contact, |
| 60 | args=(first_name, last_name), |
| 61 | timeout_in_seconds=0.5) |
| 62 | |
| 63 | Args: |
| 64 | function: The function to be executed in a child process. |
| 65 | args: Optional `function` arguments as a tuple. |
| 66 | timeout_in_seconds: The execution time limit in seconds. |
| 67 | |
| 68 | Returns: |
| 69 | The return value of the `function`. |
| 70 | |
| 71 | Raises: |
| 72 | TimeoutError: If the execution time of the `function` exceeds the |
| 73 | timeout `seconds`. |
| 74 | """ |
| 75 | process = ProcessWithResult(target=function, args=args or (), daemon=True) |
| 76 | process.start() |
| 77 | process.join(timeout=timeout_in_seconds) |
| 78 | if process.is_alive(): |
| 79 | process.kill() |
| 80 | _wait_for_process_exit(process) |
| 81 | result = process.result() |
| 82 | if result is None: |
| 83 | raise TimeoutError( |
| 84 | f'Process failed to complete in {timeout_in_seconds} seconds') |
| 85 | if not result.was_successful(): |
| 86 | raise result.exception |
| 87 | return result.return_value |
| 88 | |
| 89 | |
| 90 | def _wait_for_process_exit(target_process): |
nothing calls this directly
no test coverage detected