Connection remote classes

This file describes the main package classes API for interacting with remote hosts.

ssh_utilities.remote

Remote working substitutions for some python modules.

Includes: os, pathlib, shutil, subprocess and open from python builtins. Only a subset of API from each module is supported.

class ssh_utilities.remote.SSHConnection(address: str, username: str, password: Optional[str] = None, pkey_file: Union[pathlib.Path, str, None] = None, line_rewrite: bool = True, server_name: Optional[str] = None, quiet: bool = False, thread_safe: bool = False, allow_agent: Optional[bool] = False)

Bases: ssh_utilities.abstract._connection.ConnectionABC

Self keeping ssh connection, to execute commands and file operations.

Parameters:
  • address (str) – server IP address
  • username (str) – server login
  • password (Optional[str]) – server password for login, by default RSA keys is used
  • pkey_file (Optional[Union[str, Path]]) – path to rsa key file
  • line_rewrite (bool) – rewrite lines in output when copying instead of normal mode when lines are outputed one after the other. Default is True
  • warn_on_login (bool) – Display on warnings of firs login
  • server_name (Optional[str]) – input server name that will be later displayed on file operations, if None then IP address will be used
  • thread_safe (bool) – make connection object thread safe so it can be safely accessed from any number of threads, it is disabled by default to avoid performance penalty of threading locks

Warning

At least one of (password, pkey_file, allow_agent) must be specified, priority used is SSH-Agent -> RSA Key -> password

thread_safe parameter is not implemented yet!!!

Raises:ConnectionError – connection to remote could not be established
builtins

Inner class providing access to substitutions for python builtins.

Type:remote.Builtins
close(*, quiet: bool = True)

Close SSH connection.

Parameters:quiet (bool) – whether to print other function messages
os

Inner class providing access to substitutions for python os module.

Type:remote.Os
pathlib

Inner class providing access to substitutions for pathlib module.

Type:remote.Pathlib
sftp

Opens and return sftp channel.

If SFTP coud be open then return SFTPClient instance else return None.

Type:Optional[paramiko.SFTPClient]
Raises:SFTPOpenError – when remote home could not be found
shutil

Inner class providing access to substitutions for shutil module.

Type:remote.Shutil
subprocess

Inner class providing access to substitutions for subprocess module.

Type:remote.Subprocess
class ssh_utilities.remote.Builtins(connection: SSHConnection)

Bases: ssh_utilities.abstract._builtins.BuiltinsABC

Remote replacement for python builtins, mainly the open function.

See also

ssh_utilities.local.Builtins
local version of class with same API
open(filename: _SPATH, mode: str = 'r', buffering: int = -1, encoding: Optional[str] = None, errors: Optional[str] = None, newline: Optional[str] = None) → SFTPFile

Opens remote file, works as python open function.

Can be used both as a function or a decorator.

Parameters:
  • filename (ssh_utilities.typeshed._SPATH) – path to file to be opened
  • mode (str) – select mode to open file. Same as python open modes
  • encoding (Optional[str]) – encoding type to decode file bytes stream
  • buffering (int) – buffer size, 0 turns off buffering, 1 uses line buffering, and any number greater than 1 (>1) uses that specific buffer size
  • errors (Optional[str]) – string that specifies how encoding and decoding errors are to be handled, see builtin function open documentation for more details
  • newline (Optional[str]) – this parameter is not used in ssh implementation
Raises:

FileNotFoundError – when mode is ‘r’ and file does not exist

class ssh_utilities.remote.Os(connection: SSHConnection)

Bases: ssh_utilities.abstract._os.OsABC

Class gathering all remote methods similar to python os module.

See also

ssh_utilities.local.Os
local version of class with same API
chdir(path: _SPATH)

Changes working directory.

Parameters:path (_SPATH) – directory to change to

Warning

This is not guaranted to work on all servers. You should avoud this method or check if you are in the correct directory.

Raises:
  • FileNotFoundError – if directory does not exist
  • NotADirectoryError – if path is not a directory
chmod(path: _SPATH, mode: int, *, dir_fd: Optional[int] = None, follow_symlinks: bool = True)

Change the mode of path to the numeric mode.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path pointing to the file/directory
  • mode (int) – desired mode to set, check python os documentation to see options
  • dir_fd (Optional[int], optional) – not used by ssh implementation, by default None
  • follow_symlinks (bool, optional) – whether to resolve symlinks on the way, by default True
Raises:

FileNotFoundError – if the path does not exist

lchmod(path: _SPATH, mode: int)

Change the mode of path to the numeric mode.

If path is a symlink, this affects the symlink rather than the target.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path pointing to the file/directory
  • mode (int) – desired mode to set, check python os documentation to see options
Raises:

FileNotFoundError – if the path does not exist

listdir(path: _SPATH) → List[str]

Lists contents of specified directory.

Parameters:

path (ssh_utilities.typeshed._SPATH) – directory path

Returns:

list of files, dirs, symlinks …

Return type:

List[str]

Raises:
  • FileNotFoundError – if directory does not exist
  • NotADirectoryError – if path is not a directory
lstat(path: _SPATH, *, dir_fd: Optional[int] = None) → SFTPAttributes

Similar to stat only this does not resolve symlinks.

Parameters:
Returns:

stat object similar to one returned by os.lstat

Return type:

SFTPAttributes

Warning

dir_fd parameter has no effect, it is present only so the signature is compatible with os.lstat

makedirs(path: _SPATH, mode: int = 511, exist_ok: bool = True, quiet: bool = True)

Recursively create directory.

If it already exists, show warning and return.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to directory which should be created
  • mode (int) – create directory with mode, default is 511
  • exist_ok (bool) – if true and directory exists, exception is silently passed when dir already exists
  • quiet (bool) – if True informative messages are suppresssed
Raises:
  • OSError – if directory could not be created
  • FileExistsError – when directory already exists and exist_ok=False
mkdir(path: _SPATH, mode: int = 511)

Create single directory.

If it already exists, show warning and return.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to directory which should be created
  • mode (int) – create directory with mode, default is 511
  • quiet (bool) – if True informative messages are suppresssed
Raises:
  • OSError – if directory could not be created
  • FileNotFoundError – when parent directory is missing and parents=False
name

Try to get remote os name same as os.name function.

Warning

Due to the complexity of the check, this method only checks is remote server is windows by trying to run ver command. If that fails the remote is automatically assumed to be POSIX which should hold true in most cases. If absolute certianty is required you should do your own checks.

Note

This methods main purpose is to help choose the right flavour when instantiating ssh_utilities.path.SSHPath. For its use the provided accuracy should be sufficient.

Returns:remote server os name
Return type:Literal[“nt”, “posix”]
Raises:UnknownOsError – if remote server os name could not be determined
remove(path: _SPATH, *, dir_fd: int = None)

Remove file.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to remove
  • dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

dir_fd parameter is not implemented

Raises:
  • FileNotFoundError – if path does not point to a file
  • IsADirectoryError – if path points to a directory
  • IOError – if some other paramiko related error happens and file could not have been removed.
rename(src: _SPATH, dst: _SPATH, *, src_dir_fd: Optional[int] = None, dst_dir_fd: Optional[int] = None)

Rename the file or directory src to dst.

Parameters:
  • src (ssh_utilities.typeshed._SPATH) – source file or directory
  • dst (ssh_utilities.typeshed._SPATH) – destination file or directory
  • src_dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None
  • dst_dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

src_dir_fd parameter is not implemented dst_dir_fd parameter is not implemented

Raises:
  • FileNotFoundError – raised on win if destination path exists
  • IsADirectoryError – raised on posix if src is file and dst is directory
  • NotADirectoryError – raised on posix in src is dir and dst is file
  • IOError – if some other paramiko related error happens and file could not have been removed.
replace(src: _SPATH, dst: _SPATH, *, src_dir_fd: Optional[int] = None, dst_dir_fd: Optional[int] = None)

Rename the file or directory src to dst.

Parameters:
  • src (ssh_utilities.typeshed._SPATH) – source file or directory
  • dst (ssh_utilities.typeshed._SPATH) – destination file or directory
  • src_dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None
  • dst_dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

src_dir_fd parameter is not implemented dst_dir_fd parameter is not implemented If dst exists and is a file, it will be replaced silently if the user has permission.

Raises:
  • FileNotFoundError – raised on win if destination path exists
  • IsADirectoryError – raised on posix if src is file and dst is directory
  • NotADirectoryError – raised on posix in src is dir and dst is file
  • IOError – if some other paramiko related error happens and file could not have been removed.
rmdir(path: _SPATH, *, dir_fd: int = None)

Remove directory.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to remove
  • dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

dir_fd parameter is not implemented

Raises:
  • FileNotFoundError – if path does not point to a directory
  • OSError – if directory is not empty or some other ssh implementation related error occured
scandir(path: _SPATH) → Iterator[ssh_utilities.remote._os.DirEntryRemote]

Return an iterator of os.DirEntry objects.

These correspond to the entries in the directory given by path.

Parameters:path (ssh_utilities.typeshed._SPATH) – root path directory to scan
Returns:scandir iterator.
Return type:ssh_utilities.abstract._SCANDIR
stat(path: _SPATH, *, dir_fd: Optional[int] = None, follow_symlinks: bool = True) → SFTPAttributes

Replacement for os.stat function.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to file whose stats are desired
  • dir_fd (Any) – not implemented
  • follow_symlinks (bool) – whether to resolve symbolic links along the way
Returns:

stat object similar to one returned by os.stat

Return type:

SFTPAttributes

Warning

dir_fd parameter has no effect, it is present only so the signature is compatible with os.stat

static supports_dir_fd()

Check file descriptor support.

Raises:NotImplementedError – if passing file descriptor is unsupported
static supports_fd()

Check file descriptor support.

Raises:NotImplementedError – if passing file descriptor is unsupported

Make this path a symlink pointing to the given path.

Parameters:
  • src (ssh_utilities.typeshed._SPATH) – target path to which symlink will point
  • dst (ssh_utilities.typeshed._SPATH) – symlink path
  • target_is_directory (bool, optional) – this parameter is ignored in ssh implementation
  • dir_fd (Optional[int], optional) – this parameter is ignored in ssh implementation

Warning

target_is_directory parameter is ignored

Remove file.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to remove
  • dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

dir_fd parameter is not implemented

Raises:
  • FileNotFoundError – if path does not point to a file
  • IsADirectoryError – if path points to a directory
  • IOError – if some other paramiko related error happens and file could not have been removed.
walk(top: _SPATH, topdown: bool = True, onerror=None, followlinks: bool = False) → _WALK

Recursive directory listing.

Parameters:
  • top (ssh_utilities.typeshed._SPATH) – directory to start from
  • topdown (bool, optional) – if true or not specified, the triple for a directory is generated before the triples for any of its subdirectories (directories are generated top-down). This enables you to modify the subdirectories list in place befor iteration continues. If topdown is False, the triple for a directory is generated after the triples for all of its subdirectories, by default True
  • onerror (ssh_utilities.typeshed._ONERROR, optional) – Callable acception one argument of type exception which decides how to handle that exception, by default None
  • followlinks (bool, optional) – follow symbolic links if true, by default False
Returns:

iterator of 3 tuples containing current dir, subdirs and files

Return type:

ssh_utilities.typeshed._WALK

class ssh_utilities.remote.Pathlib(connection: SSHConnection)

Bases: ssh_utilities.abstract._pathlib.PathlibABC

Expose pathlib like API for remote hosts.

See also

ssh_utilities.local.Pathlib
local version of class with same API
Path(path: _SPATH) → ssh_utilities.remote.path.SSHPath

Provides API similar to pathlib.Path only for remote host.

Only for Unix to Unix connections

Parameters:path (ssh_utilities.typeshed._SPATH) – provide initial path
Returns:object representing remote path
Return type:SSHPath
class ssh_utilities.remote.Shutil(connection: SSHConnection)

Bases: ssh_utilities.abstract._shutil.ShutilABC

Class with remote versions of shutil methods.

See also

ssh_utilities.local.Shutil
local version of class with same API
copy(src: _SPATH, dst: _SPATH, *, direction: _DIRECTION, follow_symlinks: bool = True, callback: _CALLBACK = None, quiet: bool = True)

Send files in the chosen direction local <-> remote.

Parameters:

Warning

Unlike shutil this function cannot preserve file permissions (copy) or file metadata (copy2)

Raises:
  • FileNotFoundError – if src is not file
  • ValueError – if direction is not put or get
copy2(src: _SPATH, dst: _SPATH, *, direction: _DIRECTION, follow_symlinks: bool = True, callback: _CALLBACK = None, quiet: bool = True)

Send files in the chosen direction local <-> remote.

Parameters:

Warning

Unlike shutil this function cannot preserve file permissions (copy) or file metadata (copy2)

Raises:
  • FileNotFoundError – if src is not file
  • ValueError – if direction is not put or get
copyfile(src: _SPATH, dst: _SPATH, *, direction: _DIRECTION, follow_symlinks: bool = True, callback: _CALLBACK = None, quiet: bool = True)

Send files in the chosen direction local <-> remote.

Parameters:
Raises:
  • FileNotFoundError – if src is not file
  • IsADirectoryError – if dst is a targer directory not full path
  • ValueError – if direction is not put or get
copyfileobj(fsrc: Union[IO, SFTPFile], fdst: Union[IO, SFTPFile], *, direction: _DIRECTION, length: Optional[int] = None)

Copy the contents of one file-like object to another.

Parameters:
  • fsrc (Union[IO, SFTPFile]) – source file-like object must be local/remote based on direction parameter
  • fdst (Union[IO, SFTPFile]) – source file-like object must be local/remote based on direction parameter
  • direction (_DIRECTION) – either ‘put’ or ‘get’
  • length (int, optional) – [description], by default -1
download_tree(remote_path: _SPATH, local_path: _SPATH, include: _GLOBPAT = None, exclude: _GLOBPAT = None, remove_after: bool = False, quiet: typing_extensions.Literal[True, False, 'stats', 'progress'][True, False, stats, progress] = False)

Download directory tree from remote.

Remote directory must exist otherwise exception is raised.

Parameters:
  • remote_path (ssh_utilities.typeshed._SPATH) – path to directory which should be downloaded
  • local_path (ssh_utilities.typeshed._SPATH) – directory to copy to, must be full path!
  • remove_after (bool) – remove remote copy after directory is uploaded
  • include (ssh_utilities.typeshed._GLOBPAT) – glob pattern of files to include in copy, can be used simultaneously with exclude, default is None = no filtering
  • exclude (ssh_utilities.typeshed._GLOBPAT) – glob pattern of files to exclude in copy, can be used simultaneously with include, default is None = no filtering
  • quiet (Literal[True, False, "stats", "progress"]) – if True informative messages are suppresssed if False all is printed, if stats all statistics except progressbar are suppressed if progress only progressbar is suppressed

Warning

both paths must be full: <some_remote_path>/my_directory -> <some_local_path>/my_directory

Raises:FileNotFoundError – when remote directory does not exist
ignore_patterns(*paterns) → Callable[[Any, Sequence[str]], Set[str]]

Creates a callable for shutil.copytree function to ignore files.

Parameters:*patterns (Sequence[str]) – a list of glob patterns that will ne used to exclude files
Returns:Callable the filters files, when called with a list of strings returns a subset that matches one oth the exclude patterns
Return type:Callable[[Any, Sequence[str]], Set[str]]
rmtree(path: _SPATH, ignore_errors: bool = False, quiet: bool = True)

Recursively remove directory tree.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – directory to be recursively removed
  • ignore_errors (bool) – if True only log warnings do not raise exception
  • quiet (bool) – if True informative messages are suppresssed
Raises:

FileNotFoundError – if some part of deleting filed

upload_tree(local_path: _SPATH, remote_path: _SPATH, include: _GLOBPAT = None, exclude: _GLOBPAT = None, remove_after: bool = False, quiet: typing_extensions.Literal[True, False, 'stats', 'progress'][True, False, stats, progress] = False)

Upload directory tree to remote.

Local path must exist otherwise, exception is raised.

Parameters:
  • local_path (ssh_utilities.typeshed._SPATH) – path to directory which should be uploaded
  • remote_path (ssh_utilities.typeshed._SPATH) – directory to copy to, must be full path!
  • remove_after (bool) – remove local copy after directory is uploaded
  • include (ssh_utilities.typeshed._GLOBPAT) – glob pattern of files to include in copy, can be used simultaneously with exclude, default is None = no filtering
  • exclude (ssh_utilities.typeshed._GLOBPAT) – glob pattern of files to exclude in copy, can be used simultaneously with include, default is None = no filtering
  • quiet (Literal[True, False, "stats", "progress"]) – if True informative messages are suppresssed if False all is printed, if stats all statistics except progressbar are suppressed if progress only progressbar is suppressed

Warning

both paths must be full: <some_local_path>/my_directory -> <some_remote_path>/my_directory

Raises:FileNotFoundError – when local directory does not exist
class ssh_utilities.remote.Subprocess(connection: SSHConnection)

Bases: ssh_utilities.abstract._subprocess.SubprocessABC

Class with similar API to subprocess module.

See also

ssh_utilities.local.Subprocess
local version of class with same API
run(args: _CMD, suppress_out: bool = True, quiet: bool = True, bufsize: int = -1, executable: _SPATH = None, input: Optional[str] = None, stdin: _FILE = None, stdout: _FILE = None, stderr: _FILE = None, capture_output: bool = False, shell: bool = False, cwd: _SPATH = None, timeout: Optional[float] = None, check: bool = False, encoding: Optional[str] = None, errors: Optional[str] = None, text: Optional[bool] = None, env: _ENV = None, universal_newlines: bool = False) → ssh_utilities.utils.CompletedProcess

Excecute command on remote, has simillar API to subprocess run.

Parameters:
  • args (ssh_utilities.typeshed._CMD) – string, Path-like object or a list of strings. If it is a list it will be joined to string with whitespace delimiter.
  • suppress_out (bool) – whether to print command output to console, this is required keyword argument
  • quiet (bool, optional) – whether to print other function messages, by default True
  • bufsize (int, optional) – buffer size, 0 turns off buffering, 1 uses line buffering, and any number greater than 1 (>1) uses that specific buffer size. This applies ti underlying paramiko client as well as stdin, stdout and stderr PIPES, by default -1
  • executable (ssh_utilities.typeshed._SPATH, optional) – [description], by default None
  • input (Optional[str], optional) – [description], by default None
  • stdin (_FILE, optional) – [description], by default None
  • stdout (_FILE, optional) – [description], by default None
  • stderr (_FILE, optional) – [description], by default None
  • capture_output (bool, optional) – if true then lightweight result object with same API as subprocess.CompletedProcess is returned. Same as passing PIPE to stdout and stderr. Both options cannot be used at the same time, by default False
  • shell (bool, optional) – requests a pseudo terminal from server, by default False
  • cwd (_SPATH, optional) – execute command in this directory, by default None
  • timeout (Optional[float], optional) – set, by default None
  • check (bool, optional) – checks for command return code if other than 0 raises CalledProcessError, by default False
  • encoding (Optional[str], optional) – encoding to use when decoding remote output, default is utf-8, by default None
  • errors (Optional[str], optional) –

    string that specifies how encoding and decoding errors are to be handled, see builtin function open documentation for more details, by default None

  • text (Optional[bool], optional) – will select default encoding and open stdin, stdout and stderr streams in text mode. The default encoding, contrary to behaviour of subprocess module is not selected by locale.getpreferredencoding(False) but is always set to utf-8. By default None
  • env (ssh_utilities.typeshed._ENV, optional) – optinal environment variables that will be merged into the existing environment. This is different to subprocess behaviour which creates new environment with only specified variables, by default None
  • universal_newlines (bool, optional) – an alias for text keyword argument, by default None

Warning

New environment variables defined by env might get silently rejected by the server.

Currentlly it is only possible to use input agrgument, not stdin and send data to process only once at the begining. It is still under developement

Returns:

if capture_output is true, returns lightweight result object with same API as subprocess.CompletedProcess

Return type:

CompletedProcess

Raises:
  • ValueError – if stdin and input are specified at the same time
  • ValueError – if capture_output and stdout and/or stderr are specified at the same time
  • ValueError – if text or universal_newlines is used simultaneously with” encoding and/or errors arguments
  • NotImplementedError – if stdin or executable is used - work in progress
  • TypeError – if stdin, stdout or stderr arguments are of wrong type
  • TypeError – if args argument is of wrong type
  • CalledProcessError – if check is true and command exited with non-zero status
  • TimeoutExpired – if the command exceeded allowed run time
class ssh_utilities.remote.OsPath(connection: SSHConnection)

Bases: ssh_utilities.abstract._os_path.OsPathABC

Drop in replacement for os.path module.

exists(path: _SPATH) → bool

Check if path exists in filesystem.

Parameters:path (ssh_utilities.typeshed._SPATH) – path to check
Returns:check result
Return type:bool
getsize(path: _SPATH) → int

Return the size of path in bytes.

Parameters:path (ssh_utilities.typeshed._SPATH) – path to file/directory
Returns:size in bytes
Return type:int
Raises:OsError – if the file does not exist or is inaccessible
isdir(path: _SPATH) → bool

Check if path points to directory.

Parameters:path (ssh_utilities.typeshed._SPATH) – path to check
Returns:check result
Return type:bool
Raises:IOError – if dir could not be accessed
isfile(path: _SPATH) → bool

Check if path points to a file.

Parameters:path (ssh_utilities.typeshed._SPATH) – path to check
Returns:check result
Return type:bool
Raises:IOError – if file could not be accessed

Check if path points to symbolic link.

Parameters:path (ssh_utilities.typeshed._SPATH) – path to check
Returns:check result
Return type:bool
Raises:IOError – if dir could not be accessed
join(path: _SPATH, *paths) → str

Join one or more path components intelligently.

The return value is the concatenation of path and any members of *paths with exactly one directory separator following each non-empty part except the last, meaning that the result will only end in a separator if the last part is empty. If a component is an absolute path, all previous components are thrown away and joining continues from the absolute path component. On Windows, the drive letter is not reset when an absolute path component (e.g., ‘foo’) is encountered. If a component contains a drive letter, all previous components are thrown away and the drive letter is reset. Note that since there is a current directory for each drive, os.path.join(“c:”, “foo”) represents a path relative to the current directory on drive C: (c:foo), not c:/foo.

Parameters:
Returns:

joined path parts

Return type:

str

realpath(path: _SPATH) → str

Return the canonical path of the specified filename.

Eliminates any symbolic links encountered in the path.

Parameters:path (ssh_utilities.typeshed._SPATH) – path to resolve
Returns:string representation of the resolved path
Return type:str

ssh_utilities.remote._builtins

Python builtins remote version.

class ssh_utilities.remote._builtins.Builtins(connection: SSHConnection)

Bases: ssh_utilities.abstract._builtins.BuiltinsABC

Remote replacement for python builtins, mainly the open function.

See also

ssh_utilities.local.Builtins
local version of class with same API
open(filename: _SPATH, mode: str = 'r', buffering: int = -1, encoding: Optional[str] = None, errors: Optional[str] = None, newline: Optional[str] = None) → SFTPFile

Opens remote file, works as python open function.

Can be used both as a function or a decorator.

Parameters:
  • filename (ssh_utilities.typeshed._SPATH) – path to file to be opened
  • mode (str) – select mode to open file. Same as python open modes
  • encoding (Optional[str]) – encoding type to decode file bytes stream
  • buffering (int) – buffer size, 0 turns off buffering, 1 uses line buffering, and any number greater than 1 (>1) uses that specific buffer size
  • errors (Optional[str]) –

    string that specifies how encoding and decoding errors are to be handled, see builtin function open documentation for more details

  • newline (Optional[str]) – this parameter is not used in ssh implementation
Raises:

FileNotFoundError – when mode is ‘r’ and file does not exist

ssh_utilities.remote._os

Remote connection os methods.

class ssh_utilities.remote._os.Os(connection: SSHConnection)

Bases: ssh_utilities.abstract._os.OsABC

Class gathering all remote methods similar to python os module.

See also

ssh_utilities.local.Os
local version of class with same API
chdir(path: _SPATH)

Changes working directory.

Parameters:path (_SPATH) – directory to change to

Warning

This is not guaranted to work on all servers. You should avoud this method or check if you are in the correct directory.

Raises:
  • FileNotFoundError – if directory does not exist
  • NotADirectoryError – if path is not a directory
chmod(path: _SPATH, mode: int, *, dir_fd: Optional[int] = None, follow_symlinks: bool = True)

Change the mode of path to the numeric mode.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path pointing to the file/directory
  • mode (int) – desired mode to set, check python os documentation to see options
  • dir_fd (Optional[int], optional) – not used by ssh implementation, by default None
  • follow_symlinks (bool, optional) – whether to resolve symlinks on the way, by default True
Raises:

FileNotFoundError – if the path does not exist

lchmod(path: _SPATH, mode: int)

Change the mode of path to the numeric mode.

If path is a symlink, this affects the symlink rather than the target.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path pointing to the file/directory
  • mode (int) – desired mode to set, check python os documentation to see options
Raises:

FileNotFoundError – if the path does not exist

listdir(path: _SPATH) → List[str]

Lists contents of specified directory.

Parameters:

path (ssh_utilities.typeshed._SPATH) – directory path

Returns:

list of files, dirs, symlinks …

Return type:

List[str]

Raises:
  • FileNotFoundError – if directory does not exist
  • NotADirectoryError – if path is not a directory
lstat(path: _SPATH, *, dir_fd: Optional[int] = None) → SFTPAttributes

Similar to stat only this does not resolve symlinks.

Parameters:
Returns:

stat object similar to one returned by os.lstat

Return type:

SFTPAttributes

Warning

dir_fd parameter has no effect, it is present only so the signature is compatible with os.lstat

makedirs(path: _SPATH, mode: int = 511, exist_ok: bool = True, quiet: bool = True)

Recursively create directory.

If it already exists, show warning and return.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to directory which should be created
  • mode (int) – create directory with mode, default is 511
  • exist_ok (bool) – if true and directory exists, exception is silently passed when dir already exists
  • quiet (bool) – if True informative messages are suppresssed
Raises:
  • OSError – if directory could not be created
  • FileExistsError – when directory already exists and exist_ok=False
mkdir(path: _SPATH, mode: int = 511)

Create single directory.

If it already exists, show warning and return.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to directory which should be created
  • mode (int) – create directory with mode, default is 511
  • quiet (bool) – if True informative messages are suppresssed
Raises:
  • OSError – if directory could not be created
  • FileNotFoundError – when parent directory is missing and parents=False
name

Try to get remote os name same as os.name function.

Warning

Due to the complexity of the check, this method only checks is remote server is windows by trying to run ver command. If that fails the remote is automatically assumed to be POSIX which should hold true in most cases. If absolute certianty is required you should do your own checks.

Note

This methods main purpose is to help choose the right flavour when instantiating ssh_utilities.path.SSHPath. For its use the provided accuracy should be sufficient.

Returns:remote server os name
Return type:Literal[“nt”, “posix”]
Raises:UnknownOsError – if remote server os name could not be determined
remove(path: _SPATH, *, dir_fd: int = None)

Remove file.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to remove
  • dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

dir_fd parameter is not implemented

Raises:
  • FileNotFoundError – if path does not point to a file
  • IsADirectoryError – if path points to a directory
  • IOError – if some other paramiko related error happens and file could not have been removed.
rename(src: _SPATH, dst: _SPATH, *, src_dir_fd: Optional[int] = None, dst_dir_fd: Optional[int] = None)

Rename the file or directory src to dst.

Parameters:
  • src (ssh_utilities.typeshed._SPATH) – source file or directory
  • dst (ssh_utilities.typeshed._SPATH) – destination file or directory
  • src_dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None
  • dst_dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

src_dir_fd parameter is not implemented dst_dir_fd parameter is not implemented

Raises:
  • FileNotFoundError – raised on win if destination path exists
  • IsADirectoryError – raised on posix if src is file and dst is directory
  • NotADirectoryError – raised on posix in src is dir and dst is file
  • IOError – if some other paramiko related error happens and file could not have been removed.
replace(src: _SPATH, dst: _SPATH, *, src_dir_fd: Optional[int] = None, dst_dir_fd: Optional[int] = None)

Rename the file or directory src to dst.

Parameters:
  • src (ssh_utilities.typeshed._SPATH) – source file or directory
  • dst (ssh_utilities.typeshed._SPATH) – destination file or directory
  • src_dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None
  • dst_dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

src_dir_fd parameter is not implemented dst_dir_fd parameter is not implemented If dst exists and is a file, it will be replaced silently if the user has permission.

Raises:
  • FileNotFoundError – raised on win if destination path exists
  • IsADirectoryError – raised on posix if src is file and dst is directory
  • NotADirectoryError – raised on posix in src is dir and dst is file
  • IOError – if some other paramiko related error happens and file could not have been removed.
rmdir(path: _SPATH, *, dir_fd: int = None)

Remove directory.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to remove
  • dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

dir_fd parameter is not implemented

Raises:
  • FileNotFoundError – if path does not point to a directory
  • OSError – if directory is not empty or some other ssh implementation related error occured
scandir(path: _SPATH) → Iterator[ssh_utilities.remote._os.DirEntryRemote]

Return an iterator of os.DirEntry objects.

These correspond to the entries in the directory given by path.

Parameters:path (ssh_utilities.typeshed._SPATH) – root path directory to scan
Returns:scandir iterator.
Return type:ssh_utilities.abstract._SCANDIR
stat(path: _SPATH, *, dir_fd: Optional[int] = None, follow_symlinks: bool = True) → SFTPAttributes

Replacement for os.stat function.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to file whose stats are desired
  • dir_fd (Any) – not implemented
  • follow_symlinks (bool) – whether to resolve symbolic links along the way
Returns:

stat object similar to one returned by os.stat

Return type:

SFTPAttributes

Warning

dir_fd parameter has no effect, it is present only so the signature is compatible with os.stat

static supports_dir_fd()

Check file descriptor support.

Raises:NotImplementedError – if passing file descriptor is unsupported
static supports_fd()

Check file descriptor support.

Raises:NotImplementedError – if passing file descriptor is unsupported

Make this path a symlink pointing to the given path.

Parameters:
  • src (ssh_utilities.typeshed._SPATH) – target path to which symlink will point
  • dst (ssh_utilities.typeshed._SPATH) – symlink path
  • target_is_directory (bool, optional) – this parameter is ignored in ssh implementation
  • dir_fd (Optional[int], optional) – this parameter is ignored in ssh implementation

Warning

target_is_directory parameter is ignored

Remove file.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – path to remove
  • dir_fd (Optional[int], optional) – file descriptor, not used in ssh implementation, by default None

Warning

dir_fd parameter is not implemented

Raises:
  • FileNotFoundError – if path does not point to a file
  • IsADirectoryError – if path points to a directory
  • IOError – if some other paramiko related error happens and file could not have been removed.
walk(top: _SPATH, topdown: bool = True, onerror=None, followlinks: bool = False) → _WALK

Recursive directory listing.

Parameters:
  • top (ssh_utilities.typeshed._SPATH) – directory to start from
  • topdown (bool, optional) – if true or not specified, the triple for a directory is generated before the triples for any of its subdirectories (directories are generated top-down). This enables you to modify the subdirectories list in place befor iteration continues. If topdown is False, the triple for a directory is generated after the triples for all of its subdirectories, by default True
  • onerror (ssh_utilities.typeshed._ONERROR, optional) – Callable acception one argument of type exception which decides how to handle that exception, by default None
  • followlinks (bool, optional) – follow symbolic links if true, by default False
Returns:

iterator of 3 tuples containing current dir, subdirs and files

Return type:

ssh_utilities.typeshed._WALK

ssh_utilities.remote._pathlib

Module with pathlib functionality for SSHConnection.

class ssh_utilities.remote._pathlib.Pathlib(connection: SSHConnection)

Bases: ssh_utilities.abstract._pathlib.PathlibABC

Expose pathlib like API for remote hosts.

See also

ssh_utilities.local.Pathlib
local version of class with same API
Path(path: _SPATH) → ssh_utilities.remote.path.SSHPath

Provides API similar to pathlib.Path only for remote host.

Only for Unix to Unix connections

Parameters:path (ssh_utilities.typeshed._SPATH) – provide initial path
Returns:object representing remote path
Return type:SSHPath

ssh_utilities.remote._shutil

Module collecting shutil-like remote methods.

class ssh_utilities.remote._shutil.Shutil(connection: SSHConnection)

Bases: ssh_utilities.abstract._shutil.ShutilABC

Class with remote versions of shutil methods.

See also

ssh_utilities.local.Shutil
local version of class with same API
copy(src: _SPATH, dst: _SPATH, *, direction: _DIRECTION, follow_symlinks: bool = True, callback: _CALLBACK = None, quiet: bool = True)

Send files in the chosen direction local <-> remote.

Parameters:

Warning

Unlike shutil this function cannot preserve file permissions (copy) or file metadata (copy2)

Raises:
  • FileNotFoundError – if src is not file
  • ValueError – if direction is not put or get
copy2(src: _SPATH, dst: _SPATH, *, direction: _DIRECTION, follow_symlinks: bool = True, callback: _CALLBACK = None, quiet: bool = True)

Send files in the chosen direction local <-> remote.

Parameters:

Warning

Unlike shutil this function cannot preserve file permissions (copy) or file metadata (copy2)

Raises:
  • FileNotFoundError – if src is not file
  • ValueError – if direction is not put or get
copyfile(src: _SPATH, dst: _SPATH, *, direction: _DIRECTION, follow_symlinks: bool = True, callback: _CALLBACK = None, quiet: bool = True)

Send files in the chosen direction local <-> remote.

Parameters:
Raises:
  • FileNotFoundError – if src is not file
  • IsADirectoryError – if dst is a targer directory not full path
  • ValueError – if direction is not put or get
copyfileobj(fsrc: Union[IO, SFTPFile], fdst: Union[IO, SFTPFile], *, direction: _DIRECTION, length: Optional[int] = None)

Copy the contents of one file-like object to another.

Parameters:
  • fsrc (Union[IO, SFTPFile]) – source file-like object must be local/remote based on direction parameter
  • fdst (Union[IO, SFTPFile]) – source file-like object must be local/remote based on direction parameter
  • direction (_DIRECTION) – either ‘put’ or ‘get’
  • length (int, optional) – [description], by default -1
download_tree(remote_path: _SPATH, local_path: _SPATH, include: _GLOBPAT = None, exclude: _GLOBPAT = None, remove_after: bool = False, quiet: typing_extensions.Literal[True, False, 'stats', 'progress'][True, False, stats, progress] = False)

Download directory tree from remote.

Remote directory must exist otherwise exception is raised.

Parameters:
  • remote_path (ssh_utilities.typeshed._SPATH) – path to directory which should be downloaded
  • local_path (ssh_utilities.typeshed._SPATH) – directory to copy to, must be full path!
  • remove_after (bool) – remove remote copy after directory is uploaded
  • include (ssh_utilities.typeshed._GLOBPAT) – glob pattern of files to include in copy, can be used simultaneously with exclude, default is None = no filtering
  • exclude (ssh_utilities.typeshed._GLOBPAT) – glob pattern of files to exclude in copy, can be used simultaneously with include, default is None = no filtering
  • quiet (Literal[True, False, "stats", "progress"]) – if True informative messages are suppresssed if False all is printed, if stats all statistics except progressbar are suppressed if progress only progressbar is suppressed

Warning

both paths must be full: <some_remote_path>/my_directory -> <some_local_path>/my_directory

Raises:FileNotFoundError – when remote directory does not exist
ignore_patterns(*paterns) → Callable[[Any, Sequence[str]], Set[str]]

Creates a callable for shutil.copytree function to ignore files.

Parameters:*patterns (Sequence[str]) – a list of glob patterns that will ne used to exclude files
Returns:Callable the filters files, when called with a list of strings returns a subset that matches one oth the exclude patterns
Return type:Callable[[Any, Sequence[str]], Set[str]]
rmtree(path: _SPATH, ignore_errors: bool = False, quiet: bool = True)

Recursively remove directory tree.

Parameters:
  • path (ssh_utilities.typeshed._SPATH) – directory to be recursively removed
  • ignore_errors (bool) – if True only log warnings do not raise exception
  • quiet (bool) – if True informative messages are suppresssed
Raises:

FileNotFoundError – if some part of deleting filed

upload_tree(local_path: _SPATH, remote_path: _SPATH, include: _GLOBPAT = None, exclude: _GLOBPAT = None, remove_after: bool = False, quiet: typing_extensions.Literal[True, False, 'stats', 'progress'][True, False, stats, progress] = False)

Upload directory tree to remote.

Local path must exist otherwise, exception is raised.

Parameters:
  • local_path (ssh_utilities.typeshed._SPATH) – path to directory which should be uploaded
  • remote_path (ssh_utilities.typeshed._SPATH) – directory to copy to, must be full path!
  • remove_after (bool) – remove local copy after directory is uploaded
  • include (ssh_utilities.typeshed._GLOBPAT) – glob pattern of files to include in copy, can be used simultaneously with exclude, default is None = no filtering
  • exclude (ssh_utilities.typeshed._GLOBPAT) – glob pattern of files to exclude in copy, can be used simultaneously with include, default is None = no filtering
  • quiet (Literal[True, False, "stats", "progress"]) – if True informative messages are suppresssed if False all is printed, if stats all statistics except progressbar are suppressed if progress only progressbar is suppressed

Warning

both paths must be full: <some_local_path>/my_directory -> <some_remote_path>/my_directory

Raises:FileNotFoundError – when local directory does not exist

ssh_utilities.remote._subprocess

Remote version of subprocess module.

class ssh_utilities.remote._subprocess.Subprocess(connection: SSHConnection)

Bases: ssh_utilities.abstract._subprocess.SubprocessABC

Class with similar API to subprocess module.

See also

ssh_utilities.local.Subprocess
local version of class with same API
run(args: _CMD, suppress_out: bool = True, quiet: bool = True, bufsize: int = -1, executable: _SPATH = None, input: Optional[str] = None, stdin: _FILE = None, stdout: _FILE = None, stderr: _FILE = None, capture_output: bool = False, shell: bool = False, cwd: _SPATH = None, timeout: Optional[float] = None, check: bool = False, encoding: Optional[str] = None, errors: Optional[str] = None, text: Optional[bool] = None, env: _ENV = None, universal_newlines: bool = False) → ssh_utilities.utils.CompletedProcess

Excecute command on remote, has simillar API to subprocess run.

Parameters:
  • args (ssh_utilities.typeshed._CMD) – string, Path-like object or a list of strings. If it is a list it will be joined to string with whitespace delimiter.
  • suppress_out (bool) – whether to print command output to console, this is required keyword argument
  • quiet (bool, optional) – whether to print other function messages, by default True
  • bufsize (int, optional) – buffer size, 0 turns off buffering, 1 uses line buffering, and any number greater than 1 (>1) uses that specific buffer size. This applies ti underlying paramiko client as well as stdin, stdout and stderr PIPES, by default -1
  • executable (ssh_utilities.typeshed._SPATH, optional) – [description], by default None
  • input (Optional[str], optional) – [description], by default None
  • stdin (_FILE, optional) – [description], by default None
  • stdout (_FILE, optional) – [description], by default None
  • stderr (_FILE, optional) – [description], by default None
  • capture_output (bool, optional) – if true then lightweight result object with same API as subprocess.CompletedProcess is returned. Same as passing PIPE to stdout and stderr. Both options cannot be used at the same time, by default False
  • shell (bool, optional) – requests a pseudo terminal from server, by default False
  • cwd (_SPATH, optional) – execute command in this directory, by default None
  • timeout (Optional[float], optional) – set, by default None
  • check (bool, optional) – checks for command return code if other than 0 raises CalledProcessError, by default False
  • encoding (Optional[str], optional) – encoding to use when decoding remote output, default is utf-8, by default None
  • errors (Optional[str], optional) –

    string that specifies how encoding and decoding errors are to be handled, see builtin function open documentation for more details, by default None

  • text (Optional[bool], optional) – will select default encoding and open stdin, stdout and stderr streams in text mode. The default encoding, contrary to behaviour of subprocess module is not selected by locale.getpreferredencoding(False) but is always set to utf-8. By default None
  • env (ssh_utilities.typeshed._ENV, optional) – optinal environment variables that will be merged into the existing environment. This is different to subprocess behaviour which creates new environment with only specified variables, by default None
  • universal_newlines (bool, optional) – an alias for text keyword argument, by default None

Warning

New environment variables defined by env might get silently rejected by the server.

Currentlly it is only possible to use input agrgument, not stdin and send data to process only once at the begining. It is still under developement

Returns:

if capture_output is true, returns lightweight result object with same API as subprocess.CompletedProcess

Return type:

CompletedProcess

Raises:
  • ValueError – if stdin and input are specified at the same time
  • ValueError – if capture_output and stdout and/or stderr are specified at the same time
  • ValueError – if text or universal_newlines is used simultaneously with” encoding and/or errors arguments
  • NotImplementedError – if stdin or executable is used - work in progress
  • TypeError – if stdin, stdout or stderr arguments are of wrong type
  • TypeError – if args argument is of wrong type
  • CalledProcessError – if check is true and command exited with non-zero status
  • TimeoutExpired – if the command exceeded allowed run time

ssh_utilities.remote.path

Implements Path-like object for remote hosts.

class ssh_utilities.remote.path.SSHPath

Bases: pathlib.Path

Object providing similar API to Pathlib.Path but for remote server.

The for all methods wrapper will wrap whole pathlib library and thus will always try to cast every path to SSHPath, so logic must be emplyed in wrapper to prevent this.

Parameters:
  • connection (Connection) – instance of ssh_utilities.Connection to server
  • path (_SPATH) – initial path

Warning

Some methods have changed signature from classmethod -> instancemethod since old approach was not vaiable in this application. These are: home() and cwd().

_2str

Return an attribute of instance, which is of type owner.

_flavour(**kwds)

Internal indicator of special typing constructs. See _doc instance attribute for specific docs.

absolute()

Return an absolute version of this path.

This function works even if the path doesn’t point to anything.

No normalization is done, i.e. all ‘.’ and ‘..’ will be kept along. Use resolve() to get the canonical path to a file.

connection

Return an attribute of instance, which is of type owner.

cwd() → ssh_utilities.remote.path.SSHPath

Returns current working directory.

Warning

This is no longer a class method! Sometimes return value can be empty string for unknown reasons.

Returns:path to current directory
Return type:SSHPath
expanduser()

Return a new path with expanded ~ and ~user constructs.

As returned by os.path.expanduser

group() → str

Return file group.

Returns:string with group name
Return type:str
Raises:NotImplementedError – when used on windows host
home() → ssh_utilities.remote.path.SSHPath

Get home dir on remote server for logged in user.

Warning

This is no longer a class method!

Returns:path to remote home
Return type:Path
open(mode: str = 'r', buffering: int = -1, encoding: Optional[str] = 'utf-8', errors: Optional[str] = None, newline: Optional[str] = None) → SFTPFile

Opens remote file, works as pathlib.Path open function.

Can be used both as a function or a decorator.

Parameters:
  • filename (_SPATH) – path to file to be opened
  • mode (str) – select mode to open file. Same as python open modes
  • encoding (str) – encoding type to decode file bytes stream
  • errors (Optional[str]) – define error handling when decoding raw stream
  • newline (Optional[str]) – define how newline symbols are handled
  • buffering (int) – buffer size, 0 turns off buffering, 1 uses line buffering, and any number greater than 1 (>1) uses that specific buffer size
Raises:

FileNotFoundError – when mode is ‘r’ and file does not exist

owner()

Return file owner.

Returns:string with owner name
Return type:str
Raises:NotImplementedError – when used on windows host
samefile(other_path: _SPATH)

Return whether other_path is the same or not as this file.

As returned by os.path.samefile())

touch(mode: int = 438, exist_ok: bool = True)

Create this file with the given access mode, if it doesn’t exist.

The permissions are unix-style and identical to those used by Python’s os.chmod function.

Parameters:
  • mode (int) – integer number of the desired mode
  • exist_ok (bool) – do not raise an exception when file already exists
Raises:

FileExistsError – when file or directory with same name already exists

ssh_utilities.remote._connection_wrapper

Module taking care of ssh connection resilience to drops.

class ssh_utilities.remote._connection_wrapper.check_connections

Bases: object

A decorator to check SSH connections, implemented as callble class.

If connection is dropped while running function, re-negotiate new connection and run function again.

Parameters:
  • original_function (Callable) – function to watch for dropped connections
  • exclude_exceptions (ssh_utilities.typeshed._EXCTYPE) – tuple of exceptions not to catch
Raises:

exclude_exceptions – The esception specified in this parameter are allowed to bypass this decorator uncaught

Warning

Beware, this function can hide certain errors or cause the code to become stuck in an infinite loop!

References

https://stackoverflow.com/questions/3888158/making-decorators-with-optional-arguments

Examples

First use case is without arguments:

>>> @check_connections
... def function(*args, **kwargs):

Second possible use cases is with arguments:

>>> @check_connections(exclude_exceptions=(<Exception>, ...))
... def function(*args, **kwargs):
static _get_connection(wrapped_instance: _CLASS) → SSHConnection

Find the underlying SSHConnection in inner classes hierarchy.

This is the instance to which the wrapped method belongs to. We have to deal with branching inner classes where attribute ‘c’ is instance of SSHConnection not paramiko SSHClient. As of now we have 2 inner levels so we must iterate until we get to the base.

Parameters:wrapped_instance (_CLASS) – any of: remote.Builtins, remote.Os, remote.Pathlib, remote.Shutil, remote.Subprocess, remote.SSHConnection
Returns:the underlying connection object
Return type:SSHConnection
_negotiate(wrapped_instance: _CLASS) → bool

Negotiate new paramiko ssh connection for SSHConnection class.

Parameters:wrapped_instance (_CLASS) – any of: remote.Builtins, remote.Os, remote.Pathlib, remote.Shutil, remote.Subprocess, remote.SSHConnection
Returns:True if negotiation was succesfull
Return type:bool