def _exec_ctx()

in terranova/binds.py [0:0]

• 19 lines of code
• 5 McCabe index (conditional complexity)

    def _exec_ctx(self, *args, **kwargs) -> ContextManager[RunningCommand]:
        """
        Run the command and handle lifecycle in case we kill the parent process.
        Note: It's useful for composite action that need multiple calls.

        Returns:
            process execution context.
        """
        process: RunningCommand | None = None
        try:
            kwargs = {
                **kwargs,
                **{"_bg": True, "_bg_exc": False, "_truncate_exc": False},
            }
            process = self._cmd(*args, **kwargs)
            if isinstance(process, RunningCommand):
                yield process
            else:
                raise ValueError("Not a running command")
        finally:
            if process is not None and process.is_alive():
                process.terminate()
                try:
                    process.wait(timeout=10)
                except TimeoutException:
                    process.kill()