Handle SIGTERM/SIGINT by cleaning up processes first.
(signum: int, frame: Any)
| 98 | self._original_sigint = signal.getsignal(signal.SIGINT) |
| 99 | |
| 100 | def signal_handler(signum: int, frame: Any) -> None: |
| 101 | """Handle SIGTERM/SIGINT by cleaning up processes first.""" |
| 102 | logger.info(f"Received signal {signum}, cleaning up PIO processes...") |
| 103 | self._cleanup_all_processes() |
| 104 | |
| 105 | # Call original handler if it exists |
| 106 | original = ( |
| 107 | self._original_sigterm |
| 108 | if signum == signal.SIGTERM |
| 109 | else self._original_sigint |
| 110 | ) |
| 111 | # signal.getsignal() returns Handlers | int | Callable | None; |
| 112 | # the callable() check + isinstance excludes the int/Handlers cases. |
| 113 | if callable(original) and not isinstance(original, int): |
| 114 | original(signum, frame) |
| 115 | elif original == signal.SIG_DFL: |
| 116 | # Re-raise to trigger default behavior |
| 117 | signal.signal(signum, signal.SIG_DFL) |
| 118 | os.kill(os.getpid(), signum) |
| 119 | |
| 120 | try: |
| 121 | signal.signal(signal.SIGTERM, signal_handler) |
nothing calls this directly
no test coverage detected