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

lib.SSHClient: Derive from ExecContext

Browse source 
Make SSHClient an ExecContext by implementing _run() and _sudo().

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-03-18 05:52:42 +01:00
commit 284eb30ecf
1 changed files with 29 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -6,11 +6,12 @@ import os, abc, shlex, sys
from .util import run_cmd
from .log import *
from .ExecContext import Result
from .ExecContext import ExecContext, Result
class SSHClient(abc.ABC):
class SSHClient(ExecContext):
def __init__(self, hostname: str) -> None:
super().__init__(interactive=False, verbose_default=False)
self.___ssh = None
self.__hostname = hostname
self.__password: str|None = None
@ -37,16 +38,26 @@ class SSHClient(abc.ABC):
async def _run_cmd(self, cmd: list[str]) -> Result:
pass
async def run_cmd(
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],
wd: str|None = None,
throw: bool = True,
verbose: bool = False,
cmd_input: str|None = None,
env: dict[str, str]|None = None,
title: str=None,
output_encoding: str|None = None, # None => unchanged; "bytes" => return raw bytes
wd: str|None,
throw: bool,
verbose: bool,
cmd_input: str|None,
env: dict[str, str]|None,
title: str,
output_encoding: str|None, # None => unchanged; "bytes" => return raw bytes
) -> Result:
if wd is not None:
@ -81,6 +92,13 @@ class SSHClient(abc.ABC):
return stdout_s, stderr_s, status
async def _sudo(self, cmd: list[str], mod_env: dict[str, str], opts: list[str], *args, **kwargs) -> Result:
if self.username != 'root':
cmd = ['sudo', *opts, *cmd]
if mod_env:
log(WARNING, f'Modifying environment over SSH is not implemented, ignored')
return await self._run(cmd, *args, **kwargs)
class SSHClientInternal(SSHClient): # export
def __init__(self, hostname: str) -> None:
@ -114,7 +132,7 @@ class SSHClientInternal(SSHClient): # export
if cmd_input is not None:
stdin.write(cmd_input)
exit_status = stdout.channel.recv_exit_status()
return stdout.read(), stderr.read(), exit_status
return Result(stdout.read(), stderr.read(), exit_status)
class SSHClientCmd(SSHClient): # export