(*args, **kwargs)
| 78 | |
| 79 | def call_in_subprocess(func): |
| 80 | def wrapper(*args, **kwargs): |
| 81 | result_queue = multiprocessing.Queue() |
| 82 | p = multiprocessing.Process( |
| 83 | target=call_fn, args=(func, args, kwargs, result_queue) |
| 84 | ) |
| 85 | p.start() |
| 86 | p.join() |
| 87 | success, result = result_queue.get() |
| 88 | if success: |
| 89 | return result |
| 90 | else: |
| 91 | raise RuntimeError(f"Error in subprocess: {result}") |
| 92 | |
| 93 | return wrapper |
| 94 |
nothing calls this directly
no outgoing calls
no test coverage detected