janware/jw-pkg
1
mirror of ssh://git.janware.com/janware/proj/jw-pkg synced 2026-04-24 17:23:36 +02:00
Code Activity Actions

lib.SSHClient: Retire non-EC API

Browse source 
Remove .run_cmd(), forcing future clients to use the ExecContext
aligned API.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-03-18 05:57:54 +01:00
commit 5f6ec5a182
1 changed files with 4 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/python/jw/pkg/lib/SSHClient.py
View file

@ -19,19 +19,9 @@ class SSHClient(ExecContext):
self.__username = parsed.username
@abc.abstractmethod
async def _run_cmd(self, cmd: list[str]) -> Result:
async def _run_ssh(self, cmd: list[str]) -> Result:
pass
async def run_cmd(self, *args, **kwargs) -> Result:
kwargs.setdefault('wd', None)
kwargs.setdefault('throw', True)
kwargs.setdefault('verbose', False)
kwargs.setdefault('cmd_input', None)
kwargs.setdefault('env', None)
kwargs.setdefault('title', None)
kwargs.setdefault('output_encoding', None)
return await self._run(*args, **kwargs)
async def _run(
self,
args: list[str],
@ -61,7 +51,7 @@ class SSHClient(ExecContext):
if env is not None:
raise NotImplementedError('Passing an environment to SSH commands is not yet implemented')
stdout_b, stderr_b, status = await self._run_cmd(args, cmd_input=cmd_input)
stdout_b, stderr_b, status = await self._run_ssh(args, cmd_input=cmd_input)
if throw and status:
raise Exception(f'SSH command returned error {status}')
@ -129,7 +119,7 @@ class SSHClientInternal(SSHClient): # export
def __scp(self):
return SCPClient(self.__ssh.get_transport())
async def _run_cmd(self, cmd: list[str], cmd_input: str|None) -> Result:
async def _run_ssh(self, cmd: list[str], cmd_input: str|None) -> Result:
stdin, stdout, stderr = self.__ssh.exec_command(shlex.join(cmd), timeout=self.__timeout)
if cmd_input is not None:
stdin.write(cmd_input)
@ -165,7 +155,7 @@ class SSHClientCmd(SSHClient): # export
self.__askpass_orig[key] = os.getenv(key)
os.environ[key] = val
async def _run_cmd(self, cmd: list[str], cmd_input: str|None) -> Result:
async def _run_ssh(self, cmd: list[str], cmd_input: str|None) -> Result:
self.__init_askpass()
return await run_cmd(['ssh', self.hostname, shlex.join(cmd)],
output_encoding='bytes', cmd_input=cmd_input)